[management] Move the activity policy out of the gRPC service

Recording proxy usage is business logic, and it had ended up in the RPC handler: the throttle interval, the service-user skip, the exclusion rule for embedded and browser peers, and a store handle to write through. It moves to a reverseproxy module manager, matching how accesslogs, domain, service and proxy are already structured, and the RPC keeps only what is its own: calling the manager and deciding the request must not fail when the write does. The proxy service goes back to holding ProxyTokenChecker rather than a widened store interface. The policy tests move with the policy. The handler tests now assert only that a granted request reaches the manager, which is all the transport decides.
2026-10-06 21:49:08 +02:00 · 2026-08-09 08:20:40 +00:00
parent d2f93fcd90
commit 25b1081933
7 changed files with 301 additions and 143 deletions
@@ -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"
@@ -61,22 +62,6 @@ type ProxyTokenChecker interface {
 	IsProxyAccessTokenValid(ctx context.Context, tokenID string) (bool, error)
 }

-// ProxyStore is the slice of the management store this service reaches
-// directly. It is declared here, and satisfied by the store, so the reverse
-// proxy owns the surface it needs instead of widening manager interfaces the
-// rest of management shares.
-type ProxyStore interface {
-	ProxyTokenChecker
-	// SaveUserLastLogin is the same write the dashboard and device login paths
-	// use, reused here so a proxy SSO sign-in lands in the one place activity
-	// accounting reads.
-	SaveUserLastLogin(ctx context.Context, accountID, userID string, lastLogin time.Time) error
-	// RefreshPeerLastSeen stamps only LastSeen. SavePeerStatus is not usable
-	// here: it rewrites the connected flag and session token from a caller
-	// snapshot, which would race the sync stream that owns them.
-	RefreshPeerLastSeen(ctx context.Context, accountID, peerID string) error
-}
-
 // ProxyServiceServer implements the ProxyService gRPC server
 // AgentNetworkSynthesizer produces in-memory reverse-proxy services from
 // Agent Network provider/policy state for the proxy snapshot path; synthesised
@@ -130,11 +115,14 @@ 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

 	// Checker for proxy access token validity
-	proxyStore ProxyStore
+	tokenChecker ProxyTokenChecker

 	// OIDC configuration for proxy authentication
 	oidcConfig ProxyOIDCConfig
@@ -205,7 +193,7 @@ func enforceAccountScope(ctx context.Context, requestAccountID string) error {
 }

 // NewProxyServiceServer creates a new proxy service server.
-func NewProxyServiceServer(accessLogMgr accesslogs.Manager, tokenStore *OneTimeTokenStore, pkceStore *PKCEVerifierStore, oidcConfig ProxyOIDCConfig, peersManager peers.Manager, usersManager users.Manager, idpManager idp.Manager, proxyMgr proxy.Manager, proxyStore ProxyStore) *ProxyServiceServer {
+func NewProxyServiceServer(accessLogMgr accesslogs.Manager, tokenStore *OneTimeTokenStore, pkceStore *PKCEVerifierStore, oidcConfig ProxyOIDCConfig, peersManager peers.Manager, usersManager users.Manager, idpManager idp.Manager, proxyMgr proxy.Manager, tokenChecker ProxyTokenChecker) *ProxyServiceServer {
 	ctx, cancel := context.WithCancel(context.Background())
 	s := &ProxyServiceServer{
 		accessLogManager:  accessLogMgr,
@@ -216,7 +204,7 @@ func NewProxyServiceServer(accessLogMgr accesslogs.Manager, tokenStore *OneTimeT
 		usersManager:      usersManager,
 		idpManager:        idpManager,
 		proxyManager:      proxyMgr,
-		proxyStore:        proxyStore,
+		tokenChecker:      tokenChecker,
 		snapshotBatchSize: snapshotBatchSizeFromEnv(),
 		cancel:            cancel,
 	}
@@ -252,6 +240,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
@@ -711,8 +706,8 @@ func (s *ProxyServiceServer) heartbeat(ctx context.Context, conn *proxyConnectio
 				log.WithContext(ctx).Debugf("Failed to update proxy %s heartbeat: %v", p.ID, err)
 			}

-			if conn.tokenID != "" && s.proxyStore != nil {
-				valid, err := s.proxyStore.IsProxyAccessTokenValid(ctx, conn.tokenID)
+			if conn.tokenID != "" && s.tokenChecker != nil {
+				valid, err := s.tokenChecker.IsProxyAccessTokenValid(ctx, conn.tokenID)
 				if err != nil {
 					log.WithContext(ctx).Warnf("failed to check token validity for proxy %s: %v", conn.proxyID, err)
 					continue
@@ -1707,17 +1702,14 @@ func (s *ProxyServiceServer) GenerateSessionToken(ctx context.Context, domain, u
 	return token, nil
 }

-// recordUserLogin marks a completed reverse proxy SSO login on the user. The
-// timestamp is what activity accounting reads to count someone who only ever
-// reaches proxy-protected services and never opens the dashboard. Service users
-// have no interactive login to record. A failure is logged and dropped: the
-// next login marks it again and no authorization decision reads it.
+// 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.proxyStore == nil || user.IsServiceUser {
+	if s.activityManager == nil {
 		return
 	}

-	if err := s.proxyStore.SaveUserLastLogin(ctx, accountID, user.Id, time.Now().UTC()); err != nil {
+	if err := s.activityManager.RecordUserLogin(ctx, accountID, user); err != nil {
 		log.WithContext(ctx).Debugf("record proxy login for user %s: %v", user.Id, err)
 	}
 }
@@ -2088,41 +2080,18 @@ func (s *ProxyServiceServer) ValidateTunnelPeer(ctx context.Context, req *proto.
 	}, nil
 }

-// proxyPeerSeenInterval 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
-// all day; an hour still sits well inside the window activity accounting asks
-// about.
-const proxyPeerSeenInterval = time.Hour
-
-// recordPeerSeen marks a peer as seen when it reaches a private service over
-// the mesh, which is what lets its owner count as active. Peers that activity
-// accounting excludes are skipped rather than written for nothing, and so is a
-// peer already seen inside the interval — the row is in hand, so the throttle
-// costs nothing. A failure is logged and dropped: the next request marks it
-// again.
+// 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.proxyStore == nil || !peerCountsTowardActivity(peer) {
+	if s.activityManager == nil {
 		return
 	}

-	if peer.Status != nil && time.Since(peer.Status.LastSeen) < proxyPeerSeenInterval {
-		return
-	}
-
-	if err := s.proxyStore.RefreshPeerLastSeen(ctx, accountID, peer.ID); err != nil {
+	if err := s.activityManager.RecordPeerSeen(ctx, accountID, peer); err != nil {
 		log.WithContext(ctx).Debugf("record proxy activity for peer %s: %v", peer.ID, err)
 	}
 }

-// peerCountsTowardActivity 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 peerCountsTowardActivity(peer *peer.Peer) bool {
-	return !peer.ProxyMeta.Embedded && peer.Meta.KernelVersion != "wasm"
-}
-
 // 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
@@ -156,17 +156,11 @@ type mockTunnelPeersManager struct {
 	groupsErr error
 }

-// mockProxyStore records the activity writes the proxy makes so tests can
-// assert what was written, not merely that something was called.
-type mockProxyStore struct {
-	loginMarks []loginMark
-	seenMarks  []seenMark
-}
-
-type loginMark struct {
-	accountID string
-	userID    string
-	at        time.Time
+// 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 {
@@ -174,17 +168,12 @@ type seenMark struct {
 	peerID    string
 }

-func (m *mockProxyStore) IsProxyAccessTokenValid(_ context.Context, _ string) (bool, error) {
-	return true, nil
-}
-
-func (m *mockProxyStore) SaveUserLastLogin(_ context.Context, accountID, userID string, lastLogin time.Time) error {
-	m.loginMarks = append(m.loginMarks, loginMark{accountID: accountID, userID: userID, at: lastLogin})
+func (m *mockActivityManager) RecordUserLogin(_ context.Context, _ string, _ *types.User) error {
 	return nil
 }

-func (m *mockProxyStore) RefreshPeerLastSeen(_ context.Context, accountID, peerID string) error {
-	m.seenMarks = append(m.seenMarks, seenMark{accountID: accountID, peerID: peerID})
+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
 }

@@ -778,10 +767,9 @@ func TestValidateTunnelPeerOwnerStatus(t *testing.T) {
 	}
 }

-// TestValidateTunnelPeerRecordsActivity covers the activity write on the mesh
-// fast-path: a peer reaching a private service is what lets its owner count as
-// active, but only peers that activity accounting actually counts are written,
-// and only once per interval.
+// 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"
@@ -789,73 +777,31 @@ func TestValidateTunnelPeerRecordsActivity(t *testing.T) {
 		peerID    = "peer1"
 	)

-	tests := []struct {
-		name       string
-		peer       *peer.Peer
-		expectMark bool
-	}{
-		{
-			name:       "peer seen long ago is marked",
-			peer:       &peer.Peer{ID: peerID, Name: "agent", Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}},
-			expectMark: true,
+	activityManager := &mockActivityManager{}
+	server := &ProxyServiceServer{
+		activityManager: activityManager,
+		serviceManager: &mockReverseProxyManager{
+			proxiesByAccount: map[string][]*service.Service{
+				accountID: {{Domain: domain, AccountID: accountID}},
+			},
 		},
-		{
-			name:       "peer never seen is marked",
-			peer:       &peer.Peer{ID: peerID, Name: "agent", Status: &peer.PeerStatus{}},
-			expectMark: true,
-		},
-		{
-			// The throttle. The peer row is already in hand, so a recently seen
-			// peer costs nothing to skip.
-			name:       "peer seen inside the interval is skipped",
-			peer:       &peer.Peer{ID: peerID, Name: "agent", Status: &peer.PeerStatus{LastSeen: time.Now().Add(-10 * time.Minute)}},
-			expectMark: false,
-		},
-		{
-			name:       "embedded proxy peer is skipped",
-			peer:       &peer.Peer{ID: peerID, Name: "embedded", ProxyMeta: peer.ProxyMeta{Embedded: true}, Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}},
-			expectMark: false,
-		},
-		{
-			name:       "browser client is skipped",
-			peer:       &peer.Peer{ID: peerID, Name: "browser", Meta: peer.PeerSystemMeta{KernelVersion: "wasm"}, Status: &peer.PeerStatus{LastSeen: time.Now().Add(-3 * time.Hour)}},
-			expectMark: false,
+		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{}},
 	}

-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			proxyStore := &mockProxyStore{}
-			server := &ProxyServiceServer{
-				proxyStore: proxyStore,
-				serviceManager: &mockReverseProxyManager{
-					proxiesByAccount: map[string][]*service.Service{
-						accountID: {{Domain: domain, AccountID: accountID}},
-					},
-				},
-				peersManager: &mockTunnelPeersManager{peer: tt.peer},
-				usersManager: &mockUsersManager{users: map[string]*types.User{}},
-			}
+	resp, err := server.ValidateTunnelPeer(context.Background(), &proto.ValidateTunnelPeerRequest{
+		Domain:   domain,
+		TunnelIp: "100.64.0.1",
+	})

-			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.NoError(t, err)
-			require.True(t, resp.GetValid(), "peer should be granted access")
-
-			if !tt.expectMark {
-				assert.Empty(t, proxyStore.seenMarks, "peer should not have been marked seen")
-				return
-			}
-
-			require.Len(t, proxyStore.seenMarks, 1, "peer should have been marked seen exactly once")
-			mark := proxyStore.seenMarks[0]
-			assert.Equal(t, accountID, mark.accountID, "activity must be recorded against the service account")
-			assert.Equal(t, peerID, mark.peerID, "activity must be recorded against the calling peer")
-		})
-	}
+	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
@@ -866,9 +812,9 @@ func TestValidateTunnelPeerDeniedRecordsNoActivity(t *testing.T) {
 		accountID = "account1"
 	)

-	proxyStore := &mockProxyStore{}
+	activityManager := &mockActivityManager{}
 	server := &ProxyServiceServer{
-		proxyStore: proxyStore,
+		activityManager: activityManager,
 		serviceManager: &mockReverseProxyManager{
 			proxiesByAccount: map[string][]*service.Service{
 				accountID: {{Domain: domain, AccountID: accountID}},
@@ -890,7 +836,7 @@ func TestValidateTunnelPeerDeniedRecordsNoActivity(t *testing.T) {

 	require.NoError(t, err)
 	require.False(t, resp.GetValid(), "blocked owner should be denied")
-	assert.Empty(t, proxyStore.seenMarks, "a denied peer must not be marked seen")
+	assert.Empty(t, activityManager.seenMarks, "a denied peer must not be marked seen")
 }

 func TestGetAccountProxyByDomain(t *testing.T) {