diff --git a/management/internals/modules/reverseproxy/activity/interface.go b/management/internals/modules/reverseproxy/activity/interface.go new file mode 100644 index 000000000..e2e682dad --- /dev/null +++ b/management/internals/modules/reverseproxy/activity/interface.go @@ -0,0 +1,25 @@ +// Package activity records that a principal used a reverse proxy service, so +// that activity accounting counts people and devices which reach services +// through the proxy but never touch the dashboard or the management API. +package activity + +import ( + "context" + + "github.com/netbirdio/netbird/management/server/peer" + "github.com/netbirdio/netbird/management/server/types" +) + +// Manager records reverse proxy usage against the timestamps activity +// accounting reads. Both methods are best effort from the caller's point of +// view: a lost record is corrected by the next request, and no authorization +// decision reads them back. +type Manager interface { + // RecordUserLogin records a completed SSO sign-in to a proxied service. + // Service users have no interactive login and are ignored. + RecordUserLogin(ctx context.Context, accountID string, user *types.User) error + // RecordPeerSeen records that a peer reached a private service over the + // mesh, which is what lets its owner count as active. Peers activity + // accounting excludes, and peers already seen recently, are ignored. + RecordPeerSeen(ctx context.Context, accountID string, peer *peer.Peer) error +} diff --git a/management/internals/modules/reverseproxy/activity/manager/manager.go b/management/internals/modules/reverseproxy/activity/manager/manager.go new file mode 100644 index 000000000..978604fa0 --- /dev/null +++ b/management/internals/modules/reverseproxy/activity/manager/manager.go @@ -0,0 +1,66 @@ +package manager + +import ( + "context" + "time" + + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/activity" + "github.com/netbirdio/netbird/management/server/peer" + "github.com/netbirdio/netbird/management/server/store" + "github.com/netbirdio/netbird/management/server/types" +) + +// peerSeenInterval is how stale a peer's LastSeen must be before reaching a +// private service refreshes it. Positive tunnel validations are cached on the +// proxy for five minutes, so without a floor a busy peer would rewrite its row +// behind every request; an hour still sits well inside the window activity +// accounting asks about. +const peerSeenInterval = time.Hour + +type managerImpl struct { + store store.Store +} + +// NewManager returns the activity manager backed by the management store. +func NewManager(store store.Store) activity.Manager { + return &managerImpl{store: store} +} + +// RecordUserLogin stamps the login the same way the dashboard and device login +// paths do, so a person who only ever reaches proxied services still has a +// login on record. +func (m *managerImpl) RecordUserLogin(ctx context.Context, accountID string, user *types.User) error { + if user == nil || user.IsServiceUser { + return nil + } + + return m.store.SaveUserLastLogin(ctx, accountID, user.Id, time.Now().UTC()) +} + +// RecordPeerSeen stamps LastSeen, the column a peer activates its owner +// through. The peer the caller already holds answers the throttle without a +// query, so a peer seen inside the interval costs nothing to skip; the same +// cutoff goes to the store, which enforces it inside the UPDATE so concurrent +// requests for one peer cannot each write off their own stale read. +func (m *managerImpl) RecordPeerSeen(ctx context.Context, accountID string, peer *peer.Peer) error { + if peer == nil || !countsTowardActivity(peer) { + return nil + } + + staleBefore := time.Now().UTC().Add(-peerSeenInterval) + if peer.Status != nil && peer.Status.LastSeen.After(staleBefore) { + return nil + } + + _, err := m.store.RefreshPeerLastSeen(ctx, accountID, peer.ID, staleBefore) + + return err +} + +// countsTowardActivity reports whether the peer represents a device a person +// actually runs. Embedded proxy peers are infrastructure and browser (WASM) +// clients are ephemeral sessions, so activity accounting ignores both and a +// write for them could never count. +func countsTowardActivity(peer *peer.Peer) bool { + return !peer.ProxyMeta.Embedded && peer.Meta.KernelVersion != "wasm" +} diff --git a/management/internals/modules/reverseproxy/activity/manager/manager_test.go b/management/internals/modules/reverseproxy/activity/manager/manager_test.go new file mode 100644 index 000000000..c89bce0a1 --- /dev/null +++ b/management/internals/modules/reverseproxy/activity/manager/manager_test.go @@ -0,0 +1,149 @@ +package manager + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/netbirdio/netbird/management/server/peer" + "github.com/netbirdio/netbird/management/server/store" + "github.com/netbirdio/netbird/management/server/types" +) + +// recordingStore captures the two writes the activity manager makes. The +// embedded interface satisfies the rest and panics if anything else is called, +// which keeps the manager honest about its surface. +type recordingStore struct { + store.Store + logins []loginWrite + seen []seenWrite +} + +type loginWrite struct { + accountID string + userID string + at time.Time +} + +type seenWrite struct { + accountID string + peerID string + staleBefore time.Time +} + +func (s *recordingStore) SaveUserLastLogin(_ context.Context, accountID, userID string, lastLogin time.Time) error { + s.logins = append(s.logins, loginWrite{accountID: accountID, userID: userID, at: lastLogin}) + return nil +} + +func (s *recordingStore) RefreshPeerLastSeen(_ context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) { + s.seen = append(s.seen, seenWrite{accountID: accountID, peerID: peerID, staleBefore: staleBefore}) + return true, nil +} + +func TestRecordUserLogin(t *testing.T) { + tests := []struct { + name string + user *types.User + expectWrite bool + }{ + { + name: "regular user is recorded", + user: &types.User{Id: "user1", AccountID: "account1"}, + expectWrite: true, + }, + { + // Activity accounting never counts service users, so a row for one + // would be noise. + name: "service user is ignored", + user: &types.User{Id: "svc1", AccountID: "account1", IsServiceUser: true}, + expectWrite: false, + }, + { + name: "missing user is ignored", + user: nil, + expectWrite: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + st := &recordingStore{} + require.NoError(t, NewManager(st).RecordUserLogin(context.Background(), "account1", tt.user)) + + if !tt.expectWrite { + assert.Empty(t, st.logins, "no login should have been recorded") + return + } + + require.Len(t, st.logins, 1, "exactly one login should have been recorded") + assert.Equal(t, "account1", st.logins[0].accountID, "login must be recorded against the service account") + assert.Equal(t, tt.user.Id, st.logins[0].userID, "login must be recorded against the signing-in user") + assert.Equal(t, time.UTC, st.logins[0].at.Location(), "timestamps are written in UTC") + assert.WithinDuration(t, time.Now().UTC(), st.logins[0].at, time.Minute, "login should be stamped now") + }) + } +} + +func TestRecordPeerSeen(t *testing.T) { + tests := []struct { + name string + peer *peer.Peer + expectWrite bool + }{ + { + name: "peer seen long ago is recorded", + peer: &peer.Peer{ID: "peer1", Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}}, + expectWrite: true, + }, + { + name: "peer never seen is recorded", + peer: &peer.Peer{ID: "peer1", Status: &peer.PeerStatus{}}, + expectWrite: true, + }, + { + // The throttle. The caller already holds the peer, so skipping a + // recently seen one costs nothing. + name: "peer seen inside the interval is skipped", + peer: &peer.Peer{ID: "peer1", Status: &peer.PeerStatus{LastSeen: time.Now().Add(-10 * time.Minute)}}, + expectWrite: false, + }, + { + name: "embedded proxy peer is skipped", + peer: &peer.Peer{ID: "peer1", ProxyMeta: peer.ProxyMeta{Embedded: true}, Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}}, + expectWrite: false, + }, + { + name: "browser client is skipped", + peer: &peer.Peer{ID: "peer1", Meta: peer.PeerSystemMeta{KernelVersion: "wasm"}, Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}}, + expectWrite: false, + }, + { + name: "missing peer is ignored", + peer: nil, + expectWrite: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + st := &recordingStore{} + require.NoError(t, NewManager(st).RecordPeerSeen(context.Background(), "account1", tt.peer)) + + if !tt.expectWrite { + assert.Empty(t, st.seen, "no activity should have been recorded") + return + } + + require.Len(t, st.seen, 1, "exactly one activity write should have been recorded") + assert.Equal(t, "account1", st.seen[0].accountID, "activity must be recorded against the service account") + assert.Equal(t, tt.peer.ID, st.seen[0].peerID, "activity must be recorded against the calling peer") + assert.Equal(t, time.UTC, st.seen[0].staleBefore.Location(), "cutoffs are passed in UTC") + assert.WithinDuration(t, time.Now().UTC().Add(-peerSeenInterval), st.seen[0].staleBefore, time.Minute, + "the store must enforce the same interval the local check applies") + }) + } +} diff --git a/management/internals/server/boot.go b/management/internals/server/boot.go index 2a1e521aa..e8f4964c6 100644 --- a/management/internals/server/boot.go +++ b/management/internals/server/boot.go @@ -27,6 +27,8 @@ import ( "github.com/netbirdio/netbird/management/internals/modules/agentnetwork" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" + proxyactivity "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/activity" + proxyactivitymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/activity/manager" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc" "github.com/netbirdio/netbird/management/server/activity" @@ -231,6 +233,7 @@ func (s *BaseServer) ReverseProxyGRPCServer() *nbgrpc.ProxyServiceServer { proxyService := nbgrpc.NewProxyServiceServer(s.AccessLogsManager(), s.ProxyTokenStore(), s.PKCEVerifierStore(), s.proxyOIDCConfig(), s.PeersManager(), s.UsersManager(), s.IdpManager(), s.ProxyManager(), s.Store()) s.AfterInit(func(s *BaseServer) { proxyService.SetServiceManager(s.ServiceManager()) + proxyService.SetActivityManager(s.ProxyActivityManager()) proxyService.SetProxyController(s.ServiceProxyController()) proxyService.SetAgentNetworkSynthesizer(newAgentNetworkSynthesizer(s.Store())) proxyService.SetAgentNetworkLimitsService(s.AgentNetworkManager()) @@ -290,6 +293,13 @@ func (s *BaseServer) PKCEVerifierStore() *nbgrpc.PKCEVerifierStore { }) } +// ProxyActivityManager records reverse proxy usage for activity accounting. +func (s *BaseServer) ProxyActivityManager() proxyactivity.Manager { + return Create(s, func() proxyactivity.Manager { + return proxyactivitymanager.NewManager(s.Store()) + }) +} + func (s *BaseServer) AccessLogsManager() accesslogs.Manager { return Create(s, func() accesslogs.Manager { accessLogManager := accesslogsmanager.NewManager(s.Store(), s.PermissionsManager(), s.GeoLocationManager()) diff --git a/management/internals/shared/grpc/proxy.go b/management/internals/shared/grpc/proxy.go index b5d6590f8..40b0914ef 100644 --- a/management/internals/shared/grpc/proxy.go +++ b/management/internals/shared/grpc/proxy.go @@ -32,6 +32,7 @@ import ( "github.com/netbirdio/netbird/management/internals/modules/agentnetwork" "github.com/netbirdio/netbird/management/internals/modules/peers" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/activity" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/sessionkey" @@ -128,6 +129,9 @@ type ProxyServiceServer struct { // Manager for IdP-enriched user data (may be nil when no IdP is configured) idpManager idp.Manager + // Manager that records reverse proxy usage for activity accounting + activityManager activity.Manager + // Store for one-time authentication tokens tokenStore *OneTimeTokenStore @@ -250,6 +254,13 @@ func (s *ProxyServiceServer) SetServiceManager(manager rpservice.Manager) { s.serviceManager = manager } +// SetActivityManager wires the manager that records reverse proxy usage. +func (s *ProxyServiceServer) SetActivityManager(manager activity.Manager) { + s.mu.Lock() + defer s.mu.Unlock() + s.activityManager = manager +} + // SetAgentNetworkSynthesizer wires the agent-network service synthesiser. // Optional — when nil the snapshot path skips agent-network synthesis. The // modules layer injects this after both the proxy server and the agent-network @@ -1717,7 +1728,7 @@ func (s *ProxyServiceServer) GenerateSessionToken(ctx context.Context, domain, u groupIDs, groupNames := pairGroupIDsAndNames(userGroups) - return sessionkey.SignToken( + token, err := sessionkey.SignToken( service.SessionPrivateKey, userID, user.Email, @@ -1727,6 +1738,25 @@ func (s *ProxyServiceServer) GenerateSessionToken(ctx context.Context, domain, u groupNames, proxyauth.DefaultSessionExpiry, ) + if err != nil { + return "", err + } + + s.recordUserLogin(ctx, service.AccountID, user) + + return token, nil +} + +// recordUserLogin hands the sign-in to the activity manager. The RPC must not +// fail on it, so the error is logged and dropped here rather than returned. +func (s *ProxyServiceServer) recordUserLogin(ctx context.Context, accountID string, user *types.User) { + if s.activityManager == nil { + return + } + + if err := s.activityManager.RecordUserLogin(ctx, accountID, user); err != nil { + log.WithContext(ctx).Debugf("record proxy login for user %s: %v", user.Id, err) + } } // ValidateUserGroupAccess checks if a user has access to a service. @@ -2076,6 +2106,8 @@ func (s *ProxyServiceServer) ValidateTunnelPeer(ctx context.Context, req *proto. return nil, err } + s.recordPeerSeen(ctx, service.AccountID, peer) + log.WithFields(log.Fields{ "domain": domain, "tunnel_ip": tunnelIPStr, @@ -2093,6 +2125,18 @@ func (s *ProxyServiceServer) ValidateTunnelPeer(ctx context.Context, req *proto. }, nil } +// recordPeerSeen hands the mesh request to the activity manager. The RPC must +// not fail on it, so the error is logged and dropped here rather than returned. +func (s *ProxyServiceServer) recordPeerSeen(ctx context.Context, accountID string, peer *peer.Peer) { + if s.activityManager == nil { + return + } + + if err := s.activityManager.RecordPeerSeen(ctx, accountID, peer); err != nil { + log.WithContext(ctx).Debugf("record proxy activity for peer %s: %v", peer.ID, err) + } +} + // resolvePeerOwner returns the user a peer is linked to, once per request so // the status gate and the identity resolution below share a single lookup. // Unlinked peers (machine agents) have no owner. A lookup that fails returns diff --git a/management/internals/shared/grpc/proxy_group_access_test.go b/management/internals/shared/grpc/proxy_group_access_test.go index cb30a07b9..5f0c31eb6 100644 --- a/management/internals/shared/grpc/proxy_group_access_test.go +++ b/management/internals/shared/grpc/proxy_group_access_test.go @@ -5,6 +5,7 @@ import ( "errors" "net" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -155,6 +156,27 @@ type mockTunnelPeersManager struct { groupsErr error } +// mockActivityManager records what the RPC handed to the activity manager. The +// policy (throttling, exclusions) is the manager's and is tested there; these +// tests only pin which requests reach it. +type mockActivityManager struct { + seenMarks []seenMark +} + +type seenMark struct { + accountID string + peerID string +} + +func (m *mockActivityManager) RecordUserLogin(_ context.Context, _ string, _ *types.User) error { + return nil +} + +func (m *mockActivityManager) RecordPeerSeen(_ context.Context, accountID string, peer *peer.Peer) error { + m.seenMarks = append(m.seenMarks, seenMark{accountID: accountID, peerID: peer.ID}) + return nil +} + func (m *mockTunnelPeersManager) GetPeerByTunnelIP(_ context.Context, _ string, _ net.IP) (*peer.Peer, error) { return m.peer, m.peerErr } @@ -745,6 +767,78 @@ func TestValidateTunnelPeerOwnerStatus(t *testing.T) { } } +// TestValidateTunnelPeerRecordsActivity pins that a granted mesh request is +// handed to the activity manager. Which of those the manager then writes is its +// own decision, covered by its tests. +func TestValidateTunnelPeerRecordsActivity(t *testing.T) { + const ( + domain = "app.example.com" + accountID = "account1" + peerID = "peer1" + ) + + activityManager := &mockActivityManager{} + server := &ProxyServiceServer{ + activityManager: activityManager, + serviceManager: &mockReverseProxyManager{ + proxiesByAccount: map[string][]*service.Service{ + accountID: {{Domain: domain, AccountID: accountID}}, + }, + }, + peersManager: &mockTunnelPeersManager{ + peer: &peer.Peer{ID: peerID, Name: "agent", Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}}, + }, + usersManager: &mockUsersManager{users: map[string]*types.User{}}, + } + + resp, err := server.ValidateTunnelPeer(context.Background(), &proto.ValidateTunnelPeerRequest{ + Domain: domain, + TunnelIp: "100.64.0.1", + }) + + require.NoError(t, err) + require.True(t, resp.GetValid(), "peer should be granted access") + + require.Len(t, activityManager.seenMarks, 1, "a granted peer should reach the activity manager once") + assert.Equal(t, accountID, activityManager.seenMarks[0].accountID, "activity must be attributed to the service account") + assert.Equal(t, peerID, activityManager.seenMarks[0].peerID, "activity must be attributed to the calling peer") +} + +// TestValidateTunnelPeerDeniedRecordsNoActivity keeps the write on the granted +// path only: a refused peer is not evidence its owner was active. +func TestValidateTunnelPeerDeniedRecordsNoActivity(t *testing.T) { + const ( + domain = "app.example.com" + accountID = "account1" + ) + + activityManager := &mockActivityManager{} + server := &ProxyServiceServer{ + activityManager: activityManager, + serviceManager: &mockReverseProxyManager{ + proxiesByAccount: map[string][]*service.Service{ + accountID: {{Domain: domain, AccountID: accountID}}, + }, + }, + peersManager: &mockTunnelPeersManager{ + peer: &peer.Peer{ID: "peer1", Name: "agent", UserID: "user1", Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}}, + }, + // The owner is blocked, so the tunnel gate denies before the mint. + usersManager: &mockUsersManager{users: map[string]*types.User{ + "user1": {Id: "user1", AccountID: accountID, Blocked: true}, + }}, + } + + resp, err := server.ValidateTunnelPeer(context.Background(), &proto.ValidateTunnelPeerRequest{ + Domain: domain, + TunnelIp: "100.64.0.1", + }) + + require.NoError(t, err) + require.False(t, resp.GetValid(), "blocked owner should be denied") + assert.Empty(t, activityManager.seenMarks, "a denied peer must not be marked seen") +} + func TestGetAccountProxyByDomain(t *testing.T) { tests := []struct { name string diff --git a/management/server/http/handlers/proxy/auth_callback_integration_test.go b/management/server/http/handlers/proxy/auth_callback_integration_test.go index 51e54ebce..1dbfca4cd 100644 --- a/management/server/http/handlers/proxy/auth_callback_integration_test.go +++ b/management/server/http/handlers/proxy/auth_callback_integration_test.go @@ -19,6 +19,7 @@ import ( "github.com/stretchr/testify/require" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" + activitymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/activity/manager" nbproxy "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc" @@ -221,6 +222,7 @@ func setupAuthCallbackTest(t *testing.T) *testSetup { ) proxyService.SetServiceManager(&testServiceManager{store: testStore}) + proxyService.SetActivityManager(activitymanager.NewManager(testStore)) handler := NewAuthCallbackHandler(proxyService, nil) @@ -538,6 +540,55 @@ func TestAuthCallback_UserAllowedToLogin(t *testing.T) { // TestAuthCallback_UserDeniedByAccountStatus asserts that a user whose account // is pending approval or blocked never receives a session token from the OIDC // callback, and that the redirect carries a description the proxy can render. +// TestAuthCallback_RecordsUserLogin drives the real OIDC callback and asserts +// the login lands on the user row. That timestamp is what activity accounting +// reads, and it is the only signal that can ever count someone who reaches +// proxy-protected services from a browser and never opens the dashboard. +func TestAuthCallback_RecordsUserLogin(t *testing.T) { + setup := setupAuthCallbackTest(t) + defer setup.cleanup() + + ctx := context.Background() + + before, err := setup.store.GetUserByUserID(ctx, store.LockingStrengthNone, "allowedUserId") + require.NoError(t, err) + require.Nil(t, before.LastLogin, "fixture user starts with no login on record") + + setup.oidcServer.tokenSubject = "allowedUserId" + state := createTestState(t, setup.proxyService, "https://test-proxy.example.com/dashboard") + + req := httptest.NewRequest(http.MethodGet, "/reverse-proxy/callback?code=test-auth-code&state="+url.QueryEscape(state), nil) + rec := httptest.NewRecorder() + setup.router.ServeHTTP(rec, req) + require.Equal(t, http.StatusFound, rec.Code) + + after, err := setup.store.GetUserByUserID(ctx, store.LockingStrengthNone, "allowedUserId") + require.NoError(t, err) + require.NotNil(t, after.LastLogin, "a completed proxy SSO login must be recorded on the user") + require.WithinDuration(t, time.Now().UTC(), after.LastLogin.UTC(), time.Minute, "login should be stamped at sign-in time") +} + +// TestAuthCallback_DeniedUserLoginNotRecorded keeps the write on the granted +// path: a refused sign-in is not a login. +func TestAuthCallback_DeniedUserLoginNotRecorded(t *testing.T) { + setup := setupAuthCallbackTest(t) + defer setup.cleanup() + + ctx := context.Background() + + setup.oidcServer.tokenSubject = "blockedUserId" + state := createTestState(t, setup.proxyService, "https://test-proxy.example.com/dashboard") + + req := httptest.NewRequest(http.MethodGet, "/reverse-proxy/callback?code=test-auth-code&state="+url.QueryEscape(state), nil) + rec := httptest.NewRecorder() + setup.router.ServeHTTP(rec, req) + require.Equal(t, http.StatusFound, rec.Code) + + after, err := setup.store.GetUserByUserID(ctx, store.LockingStrengthNone, "blockedUserId") + require.NoError(t, err) + require.Nil(t, after.LastLogin, "a denied user must not be recorded as having logged in") +} + func TestAuthCallback_UserDeniedByAccountStatus(t *testing.T) { tests := []struct { name string diff --git a/management/server/store/sql_store.go b/management/server/store/sql_store.go index 67f13a32c..99bb2c2c1 100644 --- a/management/server/store/sql_store.go +++ b/management/server/store/sql_store.go @@ -599,6 +599,34 @@ func (s *SqlStore) ApproveAccountPeers(ctx context.Context, accountID string) (i return int(result.RowsAffected), nil } +// RefreshPeerLastSeen updates only peer_status_last_seen. Every other status +// column is left untouched: peer_status_connected and +// peer_status_session_started_at belong to the sync stream that owns the +// session, and a blind write here would corrupt the fencing +// MarkPeerConnectedIfNewerSession relies on. +// +// LastSeen comes from the database clock for the same reason it does there: a +// Go-side timestamp is taken before the write and can land after a connect that +// used CURRENT_TIMESTAMP, dragging the column backwards. +// +// staleBefore carries the caller's throttle into the same statement, so +// concurrent requests for one peer collapse into a single write instead of +// each racing on its own stale read. The column is nullable — Status is an +// embedded pointer, so a peer stored without one leaves it NULL — and NULL +// loses every comparison, hence the explicit branch for a peer never seen. +func (s *SqlStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) { + result := s.db.WithContext(ctx). + Model(&nbpeer.Peer{}). + Where(accountAndIDQueryCondition, accountID, peerID). + Where("(peer_status_last_seen IS NULL OR peer_status_last_seen < ?)", staleBefore). + Update("peer_status_last_seen", gorm.Expr("CURRENT_TIMESTAMP")) + if result.Error != nil { + return false, status.Errorf(status.Internal, "refresh peer last seen: %v", result.Error) + } + + return result.RowsAffected > 0, nil +} + // SaveUsers saves the given list of users to the database. func (s *SqlStore) SaveUsers(ctx context.Context, users []*types.User) error { if len(users) == 0 { diff --git a/management/server/store/sql_store_activity_test.go b/management/server/store/sql_store_activity_test.go new file mode 100644 index 000000000..b7aef47b1 --- /dev/null +++ b/management/server/store/sql_store_activity_test.go @@ -0,0 +1,122 @@ +package store + +import ( + "context" + "net/netip" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + nbpeer "github.com/netbirdio/netbird/management/server/peer" + "github.com/netbirdio/netbird/management/server/types" +) + +const activityAccountID = "activityAccountId" + +func newActivityTestStore(t *testing.T) Store { + t.Helper() + + store, cleanUp, err := NewTestStoreFromSQL(context.Background(), "", t.TempDir()) + require.NoError(t, err) + t.Cleanup(cleanUp) + + require.NoError(t, store.SaveAccount(context.Background(), &types.Account{ + Id: activityAccountID, + Domain: "activity.example.com", + CreatedAt: time.Now().UTC(), + })) + + return store +} + +func TestRefreshPeerLastSeen(t *testing.T) { + ctx := context.Background() + store := newActivityTestStore(t) + stored := time.Now().UTC().Add(-3 * time.Hour) + require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(stored))) + + refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour)) + require.NoError(t, err) + assert.True(t, refreshed, "a peer seen three hours ago is stale enough to refresh") + + peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer") + require.NoError(t, err) + assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should be stamped at write time") + assert.True(t, peer.Status.LastSeen.After(stored), "last seen must move forward") +} + +// TestRefreshPeerLastSeenHonoursCutoff covers the throttle the caller relies on: +// two concurrent requests both read the same stale peer, but only the statement +// that still finds LastSeen behind the cutoff writes. +func TestRefreshPeerLastSeenHonoursCutoff(t *testing.T) { + ctx := context.Background() + store := newActivityTestStore(t) + stored := time.Now().UTC().Add(-10 * time.Minute) + require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(stored))) + + refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour)) + require.NoError(t, err) + assert.False(t, refreshed, "a peer seen inside the interval must not be written") + + peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer") + require.NoError(t, err) + assert.WithinDuration(t, stored, peer.Status.LastSeen.UTC(), time.Second, "last seen must be left where it was") +} + +// TestRefreshPeerLastSeenRecordsNeverSeenPeer covers the nullable column. Status +// is an embedded pointer, so a peer stored without one leaves last seen NULL, +// and NULL loses the cutoff comparison — such a peer would never record its +// first activity. +func TestRefreshPeerLastSeenRecordsNeverSeenPeer(t *testing.T) { + ctx := context.Background() + store := newActivityTestStore(t) + stored := activityPeer(time.Time{}) + stored.Status = nil + require.NoError(t, store.AddPeerToAccount(ctx, stored)) + + refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour)) + require.NoError(t, err) + assert.True(t, refreshed, "a peer that was never seen must record its first activity") + + peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer") + require.NoError(t, err) + assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should be stamped at write time") +} + +// TestRefreshPeerLastSeenLeavesSessionStateAlone pins the column boundary: the +// connected flag and the session token belong to the sync stream that owns the +// peer's session, and a blind write here would corrupt its fencing. This is why +// SavePeerStatus is not reused for an activity bump. +func TestRefreshPeerLastSeenLeavesSessionStateAlone(t *testing.T) { + ctx := context.Background() + store := newActivityTestStore(t) + + stored := activityPeer(time.Date(2026, 3, 1, 9, 0, 0, 0, time.UTC)) + stored.Status.Connected = true + stored.Status.SessionStartedAt = 1234567890 + require.NoError(t, store.AddPeerToAccount(ctx, stored)) + + refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour)) + require.NoError(t, err) + require.True(t, refreshed, "the peer is stale enough to refresh") + + peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer") + require.NoError(t, err) + assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should move forward") + assert.True(t, peer.Status.Connected, "connected flag must survive an activity write") + assert.Equal(t, int64(1234567890), peer.Status.SessionStartedAt, "session token must survive an activity write") +} + +func activityPeer(lastSeen time.Time) *nbpeer.Peer { + return &nbpeer.Peer{ + ID: "activityPeer", + AccountID: activityAccountID, + Key: "activityPeerKey", + IP: netip.MustParseAddr("100.64.0.9"), + Name: "activity-peer", + DNSLabel: "activity-peer", + Status: &nbpeer.PeerStatus{LastSeen: lastSeen}, + } +} diff --git a/management/server/store/store.go b/management/server/store/store.go index ed0bed9fc..869e8dab5 100644 --- a/management/server/store/store.go +++ b/management/server/store/store.go @@ -180,6 +180,14 @@ type Store interface { // Returns true when the update happened, false when this stream lost // the race against a newer session. MarkPeerConnectedIfNewerSession(ctx context.Context, accountID, peerID string, newSessionStartedAt int64) (bool, error) + // RefreshPeerLastSeen records that a peer was just seen, stamping the + // database clock like the other status writers. Connected and + // SessionStartedAt are left alone, so this never interferes with the + // session-ownership protocol MarkPeerConnectedIfNewerSession implements. + // The write only lands when the stored LastSeen is older than + // staleBefore, which keeps a caller's throttle atomic under concurrent + // requests for the same peer. Returns true when the update happened. + RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) // MarkPeerDisconnectedIfSameSession sets the peer to disconnected and // resets SessionStartedAt to zero, but only when the stored // SessionStartedAt equals the given sessionStartedAt. LastSeen is diff --git a/management/server/store/store_mock.go b/management/server/store/store_mock.go index 1f17a0724..3d9e160ba 100644 --- a/management/server/store/store_mock.go +++ b/management/server/store/store_mock.go @@ -3261,6 +3261,21 @@ func (mr *MockStoreMockRecorder) MarkProxyAccessTokenUsed(ctx, tokenID interface return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkProxyAccessTokenUsed", reflect.TypeOf((*MockStore)(nil).MarkProxyAccessTokenUsed), ctx, tokenID) } +// RefreshPeerLastSeen mocks base method. +func (m *MockStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "RefreshPeerLastSeen", ctx, accountID, peerID, staleBefore) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// RefreshPeerLastSeen indicates an expected call of RefreshPeerLastSeen. +func (mr *MockStoreMockRecorder) RefreshPeerLastSeen(ctx, accountID, peerID, staleBefore interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPeerLastSeen", reflect.TypeOf((*MockStore)(nil).RefreshPeerLastSeen), ctx, accountID, peerID, staleBefore) +} + // RemovePeerFromAllGroups mocks base method. func (m *MockStore) RemovePeerFromAllGroups(ctx context.Context, peerID string) error { m.ctrl.T.Helper()