Merge branch 'main' into ui-refactor

management/internals/shared/grpc/proxy.go

@@ -9,6 +9,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"net/url"
@@ -136,9 +137,12 @@ type proxyConnection struct {
 	tokenID      string
 	capabilities *proto.ProxyCapabilities
 	stream       proto.ProxyService_GetMappingUpdateServer
-	sendChan     chan *proto.GetMappingUpdateResponse
-	ctx          context.Context
-	cancel       context.CancelFunc
+	// syncStream is set when the proxy connected via SyncMappings.
+	// When non-nil, the sender goroutine uses this instead of stream.
+	syncStream proto.ProxyService_SyncMappingsServer
+	sendChan   chan *proto.GetMappingUpdateResponse
+	ctx        context.Context
+	cancel     context.CancelFunc
 }
 
 func enforceAccountScope(ctx context.Context, requestAccountID string) error {
@@ -206,145 +210,323 @@ func (s *ProxyServiceServer) SetProxyController(proxyController proxy.Controller
 	s.proxyController = proxyController
 }
 
+// proxyConnectParams holds the validated parameters extracted from either
+// a GetMappingUpdateRequest or a SyncMappingsInit message.
+type proxyConnectParams struct {
+	proxyID      string
+	address      string
+	capabilities *proto.ProxyCapabilities
+}
+
 // GetMappingUpdate handles the control stream with proxy clients
 func (s *ProxyServiceServer) GetMappingUpdate(req *proto.GetMappingUpdateRequest, stream proto.ProxyService_GetMappingUpdateServer) error {
-	ctx := stream.Context()
-
-	peerInfo := PeerIPFromContext(ctx)
-	log.Infof("New proxy connection from %s", peerInfo)
-
-	proxyID := req.GetProxyId()
-	if proxyID == "" {
-		return status.Errorf(codes.InvalidArgument, "proxy_id is required")
-	}
-
-	proxyAddress := req.GetAddress()
-	if !isProxyAddressValid(proxyAddress) {
-		return status.Errorf(codes.InvalidArgument, "proxy address is invalid")
-	}
-
-	var accountID *string
-	token := GetProxyTokenFromContext(ctx)
-	if token != nil && token.AccountID != nil {
-		accountID = token.AccountID
-
-		available, err := s.proxyManager.IsClusterAddressAvailable(ctx, proxyAddress, *accountID)
-		if err != nil {
-			return status.Errorf(codes.Internal, "check cluster address: %v", err)
-		}
-		if !available {
-			return status.Errorf(codes.AlreadyExists, "cluster address %s is already in use", proxyAddress)
-		}
-	}
-
-	var tokenID string
-	if token != nil {
-		tokenID = token.ID
-	}
-
-	sessionID := uuid.NewString()
-
-	if old, loaded := s.connectedProxies.Load(proxyID); loaded {
-		oldConn := old.(*proxyConnection)
-		log.WithFields(log.Fields{
-			"proxy_id":       proxyID,
-			"old_session_id": oldConn.sessionID,
-			"new_session_id": sessionID,
-		}).Info("Superseding existing proxy connection")
-		oldConn.cancel()
-	}
-
-	connCtx, cancel := context.WithCancel(ctx)
-	conn := &proxyConnection{
-		proxyID:      proxyID,
-		sessionID:    sessionID,
-		address:      proxyAddress,
-		accountID:    accountID,
-		tokenID:      tokenID,
-		capabilities: req.GetCapabilities(),
-		stream:       stream,
-		sendChan:     make(chan *proto.GetMappingUpdateResponse, 100),
-		ctx:          connCtx,
-		cancel:       cancel,
-	}
-
-	var caps *proxy.Capabilities
-	if c := req.GetCapabilities(); c != nil {
-		caps = &proxy.Capabilities{
-			SupportsCustomPorts: c.SupportsCustomPorts,
-			RequireSubdomain:    c.RequireSubdomain,
-			SupportsCrowdsec:    c.SupportsCrowdsec,
-		}
-	}
-	proxyRecord, err := s.proxyManager.Connect(ctx, proxyID, sessionID, proxyAddress, peerInfo, accountID, caps)
+	params, err := s.validateProxyConnect(req.GetProxyId(), req.GetAddress(), stream.Context())
 	if err != nil {
-		cancel()
-		if accountID != nil {
-			return status.Errorf(codes.Internal, "failed to register BYOP proxy: %v", err)
-		}
-		log.WithContext(ctx).Warnf("failed to register proxy %s in database: %v", proxyID, err)
-		return status.Errorf(codes.Internal, "register proxy in database: %v", err)
+		return err
+	}
+	params.capabilities = req.GetCapabilities()
+
+	conn, proxyRecord, err := s.registerProxyConnection(stream.Context(), params, &proxyConnection{
+		stream: stream,
+	})
+	if err != nil {
+		return err
 	}
 
-	s.connectedProxies.Store(proxyID, conn)
-	if err := s.proxyController.RegisterProxyToCluster(ctx, conn.address, proxyID); err != nil {
-		log.WithContext(ctx).Warnf("Failed to register proxy %s in cluster: %v", proxyID, err)
-	}
-
-	if err := s.sendSnapshot(ctx, conn); err != nil {
-		if s.connectedProxies.CompareAndDelete(proxyID, conn) {
-			if unregErr := s.proxyController.UnregisterProxyFromCluster(context.Background(), conn.address, proxyID); unregErr != nil {
-				log.WithContext(ctx).Debugf("cleanup after snapshot failure for proxy %s: %v", proxyID, unregErr)
-			}
-		}
-		cancel()
-		if disconnErr := s.proxyManager.Disconnect(context.Background(), proxyID, sessionID); disconnErr != nil {
-			log.WithContext(ctx).Debugf("cleanup after snapshot failure for proxy %s: %v", proxyID, disconnErr)
-		}
-		return fmt.Errorf("send snapshot to proxy %s: %w", proxyID, err)
+	if err := s.sendSnapshot(stream.Context(), conn); err != nil {
+		s.cleanupFailedSnapshot(stream.Context(), conn)
+		return fmt.Errorf("send snapshot to proxy %s: %w", params.proxyID, err)
 	}
 
 	errChan := make(chan error, 2)
 	go s.sender(conn, errChan)
 
-	log.WithFields(log.Fields{
-		"proxy_id":      proxyID,
-		"session_id":    sessionID,
-		"address":       proxyAddress,
-		"cluster_addr":  proxyAddress,
-		"account_id":    accountID,
-		"total_proxies": len(s.GetConnectedProxies()),
-	}).Info("Proxy registered in cluster")
-	defer func() {
-		if !s.connectedProxies.CompareAndDelete(proxyID, conn) {
-			log.Infof("Proxy %s session %s: skipping cleanup, superseded by new connection", proxyID, sessionID)
-			cancel()
+	return s.serveProxyConnection(conn, proxyRecord, errChan, false)
+}
+
+// SyncMappings implements the bidirectional SyncMappings RPC.
+// It mirrors GetMappingUpdate but provides application-level back-pressure:
+// management waits for an ack from the proxy before sending the next batch.
+func (s *ProxyServiceServer) SyncMappings(stream proto.ProxyService_SyncMappingsServer) error {
+	init, err := recvSyncInit(stream)
+	if err != nil {
+		return err
+	}
+
+	params, err := s.validateProxyConnect(init.GetProxyId(), init.GetAddress(), stream.Context())
+	if err != nil {
+		return err
+	}
+	params.capabilities = init.GetCapabilities()
+
+	conn, proxyRecord, err := s.registerProxyConnection(stream.Context(), params, &proxyConnection{
+		syncStream: stream,
+	})
+	if err != nil {
+		return err
+	}
+
+	if err := s.sendSnapshotSync(stream.Context(), conn, stream); err != nil {
+		s.cleanupFailedSnapshot(stream.Context(), conn)
+		return fmt.Errorf("send snapshot to proxy %s: %w", params.proxyID, err)
+	}
+
+	errChan := make(chan error, 2)
+	go s.sender(conn, errChan)
+	go s.drainRecv(stream, errChan)
+
+	return s.serveProxyConnection(conn, proxyRecord, errChan, true)
+}
+
+// recvSyncInit receives and validates the first message on a SyncMappings stream.
+func recvSyncInit(stream proto.ProxyService_SyncMappingsServer) (*proto.SyncMappingsInit, error) {
+	firstMsg, err := stream.Recv()
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "receive init: %v", err)
+	}
+	init := firstMsg.GetInit()
+	if init == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "first message must be init")
+	}
+	return init, nil
+}
+
+// validateProxyConnect validates the proxy ID and address, and checks cluster
+// address availability for account-scoped tokens.
+func (s *ProxyServiceServer) validateProxyConnect(proxyID, address string, ctx context.Context) (proxyConnectParams, error) {
+	if proxyID == "" {
+		return proxyConnectParams{}, status.Errorf(codes.InvalidArgument, "proxy_id is required")
+	}
+	if !isProxyAddressValid(address) {
+		return proxyConnectParams{}, status.Errorf(codes.InvalidArgument, "proxy address is invalid")
+	}
+
+	token := GetProxyTokenFromContext(ctx)
+	if token != nil && token.AccountID != nil {
+		available, err := s.proxyManager.IsClusterAddressAvailable(ctx, address, *token.AccountID)
+		if err != nil {
+			return proxyConnectParams{}, status.Errorf(codes.Internal, "check cluster address: %v", err)
+		}
+		if !available {
+			return proxyConnectParams{}, status.Errorf(codes.AlreadyExists, "cluster address %s is already in use", address)
+		}
+	}
+
+	return proxyConnectParams{proxyID: proxyID, address: address}, nil
+}
+
+// registerProxyConnection creates a proxyConnection, registers it with the
+// proxy manager and cluster, and stores it in connectedProxies. The caller
+// provides a partially initialised connSeed with stream-specific fields set;
+// the remaining fields are filled in here.
+func (s *ProxyServiceServer) registerProxyConnection(ctx context.Context, params proxyConnectParams, connSeed *proxyConnection) (*proxyConnection, *proxy.Proxy, error) {
+	peerInfo := PeerIPFromContext(ctx)
+
+	var accountID *string
+	var tokenID string
+	if token := GetProxyTokenFromContext(ctx); token != nil {
+		if token.AccountID != nil {
+			accountID = token.AccountID
+		}
+		tokenID = token.ID
+	}
+
+	sessionID := uuid.NewString()
+	s.supersedePriorConnection(params.proxyID, sessionID)
+
+	connCtx, cancel := context.WithCancel(ctx)
+	connSeed.proxyID = params.proxyID
+	connSeed.sessionID = sessionID
+	connSeed.address = params.address
+	connSeed.accountID = accountID
+	connSeed.tokenID = tokenID
+	connSeed.capabilities = params.capabilities
+	connSeed.sendChan = make(chan *proto.GetMappingUpdateResponse, 100)
+	connSeed.ctx = connCtx
+	connSeed.cancel = cancel
+
+	var caps *proxy.Capabilities
+	if c := params.capabilities; c != nil {
+		caps = &proxy.Capabilities{
+			SupportsCustomPorts: c.SupportsCustomPorts,
+			RequireSubdomain:    c.RequireSubdomain,
+			SupportsCrowdsec:    c.SupportsCrowdsec,
+			Private:             c.Private,
+		}
+	}
+
+	proxyRecord, err := s.proxyManager.Connect(ctx, params.proxyID, sessionID, params.address, peerInfo, accountID, caps)
+	if err != nil {
+		cancel()
+		if accountID != nil {
+			return nil, nil, status.Errorf(codes.Internal, "failed to register BYOP proxy: %v", err)
+		}
+		log.WithContext(ctx).Warnf("failed to register proxy %s in database: %v", params.proxyID, err)
+		return nil, nil, status.Errorf(codes.Internal, "register proxy in database: %v", err)
+	}
+
+	s.connectedProxies.Store(params.proxyID, connSeed)
+	if err := s.proxyController.RegisterProxyToCluster(ctx, params.address, params.proxyID); err != nil {
+		log.WithContext(ctx).Warnf("Failed to register proxy %s in cluster: %v", params.proxyID, err)
+	}
+
+	return connSeed, proxyRecord, nil
+}
+
+// supersedePriorConnection cancels any existing connection for the given proxy.
+func (s *ProxyServiceServer) supersedePriorConnection(proxyID, newSessionID string) {
+	if old, loaded := s.connectedProxies.Load(proxyID); loaded {
+		oldConn := old.(*proxyConnection)
+		log.WithFields(log.Fields{
+			"proxy_id":       proxyID,
+			"old_session_id": oldConn.sessionID,
+			"new_session_id": newSessionID,
+		}).Info("Superseding existing proxy connection")
+		oldConn.cancel()
+	}
+}
+
+// cleanupFailedSnapshot removes the connection from the cluster and store
+// after a snapshot send failure.
+func (s *ProxyServiceServer) cleanupFailedSnapshot(ctx context.Context, conn *proxyConnection) {
+	if s.connectedProxies.CompareAndDelete(conn.proxyID, conn) {
+		if err := s.proxyController.UnregisterProxyFromCluster(context.Background(), conn.address, conn.proxyID); err != nil {
+			log.WithContext(ctx).Debugf("cleanup after snapshot failure for proxy %s: %v", conn.proxyID, err)
+		}
+	}
+	conn.cancel()
+	if err := s.proxyManager.Disconnect(context.Background(), conn.proxyID, conn.sessionID); err != nil {
+		log.WithContext(ctx).Debugf("cleanup after snapshot failure for proxy %s: %v", conn.proxyID, err)
+	}
+}
+
+// drainRecv consumes and discards messages from a bidirectional stream.
+// The proxy sends an ack for every incremental update; we don't need them
+// after the snapshot phase. Recv errors are forwarded to errChan.
+func (s *ProxyServiceServer) drainRecv(stream proto.ProxyService_SyncMappingsServer, errChan chan<- error) {
+	for {
+		if _, err := stream.Recv(); err != nil {
+			errChan <- err
 			return
 		}
+	}
+}
 
-		if err := s.proxyController.UnregisterProxyFromCluster(context.Background(), conn.address, proxyID); err != nil {
-			log.Warnf("Failed to unregister proxy %s from cluster: %v", proxyID, err)
-		}
-		if err := s.proxyManager.Disconnect(context.Background(), proxyID, sessionID); err != nil {
-			log.Warnf("Failed to mark proxy %s as disconnected: %v", proxyID, err)
-		}
+// serveProxyConnection runs the post-snapshot lifecycle: heartbeat, sender,
+// and wait for termination. When bidi is true, normal stream closure (EOF,
+// canceled) is treated as a clean disconnect rather than an error.
+func (s *ProxyServiceServer) serveProxyConnection(conn *proxyConnection, proxyRecord *proxy.Proxy, errChan <-chan error, bidi bool) error {
+	log.WithFields(log.Fields{
+		"proxy_id":      conn.proxyID,
+		"session_id":    conn.sessionID,
+		"address":       conn.address,
+		"cluster_addr":  conn.address,
+		"account_id":    conn.accountID,
+		"total_proxies": len(s.GetConnectedProxies()),
+	}).Info("Proxy registered in cluster")
 
-		cancel()
-		log.Infof("Proxy %s session %s disconnected", proxyID, sessionID)
-	}()
-
-	go s.heartbeat(connCtx, conn, proxyRecord)
+	defer s.disconnectProxy(conn)
+	go s.heartbeat(conn.ctx, conn, proxyRecord)
 
 	select {
 	case err := <-errChan:
-		log.WithContext(ctx).Warnf("Failed to send update: %v", err)
-		return fmt.Errorf("send update to proxy %s: %w", proxyID, err)
-	case <-connCtx.Done():
-		log.WithContext(ctx).Infof("Proxy %s context canceled", proxyID)
-		return connCtx.Err()
+		if bidi && isStreamClosed(err) {
+			log.Infof("Proxy %s stream closed", conn.proxyID)
+			return nil
+		}
+		log.Warnf("Failed to send update: %v", err)
+		return fmt.Errorf("send update to proxy %s: %w", conn.proxyID, err)
+	case <-conn.ctx.Done():
+		log.Infof("Proxy %s context canceled", conn.proxyID)
+		return conn.ctx.Err()
 	}
 }
 
+// disconnectProxy removes the connection from cluster and store, unless it
+// has already been superseded by a newer connection.
+func (s *ProxyServiceServer) disconnectProxy(conn *proxyConnection) {
+	if !s.connectedProxies.CompareAndDelete(conn.proxyID, conn) {
+		log.Infof("Proxy %s session %s: skipping cleanup, superseded by new connection", conn.proxyID, conn.sessionID)
+		conn.cancel()
+		return
+	}
+
+	if err := s.proxyController.UnregisterProxyFromCluster(context.Background(), conn.address, conn.proxyID); err != nil {
+		log.Warnf("Failed to unregister proxy %s from cluster: %v", conn.proxyID, err)
+	}
+	if err := s.proxyManager.Disconnect(context.Background(), conn.proxyID, conn.sessionID); err != nil {
+		log.Warnf("Failed to mark proxy %s as disconnected: %v", conn.proxyID, err)
+	}
+
+	conn.cancel()
+	log.Infof("Proxy %s session %s disconnected", conn.proxyID, conn.sessionID)
+}
+
+// sendSnapshotSync sends the initial snapshot with back-pressure: it sends
+// one batch, then waits for the proxy to ack before sending the next.
+func (s *ProxyServiceServer) sendSnapshotSync(ctx context.Context, conn *proxyConnection, stream proto.ProxyService_SyncMappingsServer) error {
+	if !isProxyAddressValid(conn.address) {
+		return fmt.Errorf("proxy address is invalid")
+	}
+	if s.snapshotBatchSize <= 0 {
+		return fmt.Errorf("invalid snapshot batch size: %d", s.snapshotBatchSize)
+	}
+
+	mappings, err := s.snapshotServiceMappings(ctx, conn)
+	if err != nil {
+		return err
+	}
+
+	for i := 0; i < len(mappings); i += s.snapshotBatchSize {
+		end := i + s.snapshotBatchSize
+		if end > len(mappings) {
+			end = len(mappings)
+		}
+		for _, m := range mappings[i:end] {
+			token, err := s.tokenStore.GenerateToken(m.AccountId, m.Id, s.proxyTokenTTL())
+			if err != nil {
+				return fmt.Errorf("generate auth token for service %s: %w", m.Id, err)
+			}
+			m.AuthToken = token
+		}
+		if err := stream.Send(&proto.SyncMappingsResponse{
+			Mapping:             mappings[i:end],
+			InitialSyncComplete: end == len(mappings),
+		}); err != nil {
+			return fmt.Errorf("send snapshot batch: %w", err)
+		}
+
+		if err := waitForAck(stream); err != nil {
+			return err
+		}
+	}
+
+	if len(mappings) == 0 {
+		if err := stream.Send(&proto.SyncMappingsResponse{
+			InitialSyncComplete: true,
+		}); err != nil {
+			return fmt.Errorf("send snapshot completion: %w", err)
+		}
+
+		if err := waitForAck(stream); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func waitForAck(stream proto.ProxyService_SyncMappingsServer) error {
+	msg, err := stream.Recv()
+	if err != nil {
+		return fmt.Errorf("receive ack: %w", err)
+	}
+	if msg.GetAck() == nil {
+		return fmt.Errorf("expected ack, got %T", msg.GetMsg())
+	}
+	return nil
+}
+
 // heartbeat updates the proxy's last_seen timestamp every minute and
 // disconnects the proxy if its access token has been revoked.
 func (s *ProxyServiceServer) heartbeat(ctx context.Context, conn *proxyConnection, p *proxy.Proxy) {
@@ -381,6 +563,9 @@ func (s *ProxyServiceServer) sendSnapshot(ctx context.Context, conn *proxyConnec
 	if !isProxyAddressValid(conn.address) {
 		return fmt.Errorf("proxy address is invalid")
 	}
+	if s.snapshotBatchSize <= 0 {
+		return fmt.Errorf("invalid snapshot batch size: %d", s.snapshotBatchSize)
+	}
 
 	mappings, err := s.snapshotServiceMappings(ctx, conn)
 	if err != nil {
@@ -460,12 +645,26 @@ func isProxyAddressValid(addr string) bool {
 	return err == nil
 }
 
-// sender handles sending messages to proxy
+// isStreamClosed returns true for errors that indicate normal stream
+// termination: io.EOF, context cancellation, or gRPC Canceled.
+func isStreamClosed(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
+		return true
+	}
+	return status.Code(err) == codes.Canceled
+}
+
+// sender handles sending messages to proxy.
+// When conn.syncStream is set the message is sent as SyncMappingsResponse;
+// otherwise the legacy GetMappingUpdateResponse stream is used.
 func (s *ProxyServiceServer) sender(conn *proxyConnection, errChan chan<- error) {
 	for {
 		select {
 		case resp := <-conn.sendChan:
-			if err := conn.stream.Send(resp); err != nil {
+			if err := conn.sendResponse(resp); err != nil {
 				errChan <- err
 				return
 			}
@@ -475,6 +674,17 @@ func (s *ProxyServiceServer) sender(conn *proxyConnection, errChan chan<- error)
 	}
 }
 
+// sendResponse sends a mapping update on whichever stream the proxy connected with.
+func (conn *proxyConnection) sendResponse(resp *proto.GetMappingUpdateResponse) error {
+	if conn.syncStream != nil {
+		return conn.syncStream.Send(&proto.SyncMappingsResponse{
+			Mapping:             resp.Mapping,
+			InitialSyncComplete: resp.InitialSyncComplete,
+		})
+	}
+	return conn.stream.Send(resp)
+}
+
 // SendAccessLog processes access log from proxy
 func (s *ProxyServiceServer) SendAccessLog(ctx context.Context, req *proto.SendAccessLogRequest) (*proto.SendAccessLogResponse, error) {
 	accessLog := req.GetLog()
@@ -541,10 +751,15 @@ func (s *ProxyServiceServer) SendServiceUpdate(update *proto.GetMappingUpdateRes
 				return true
 			}
 			connUpdate = &proto.GetMappingUpdateResponse{
-				Mapping:              filtered,
-				InitialSyncComplete:  update.InitialSyncComplete,
+				Mapping:             filtered,
+				InitialSyncComplete: update.InitialSyncComplete,
 			}
 		}
+		// Drop mappings the proxy lacks capability for (e.g. private without SupportsPrivateService).
+		connUpdate = filterMappingsForProxy(conn, connUpdate)
+		if connUpdate == nil || len(connUpdate.Mapping) == 0 {
+			return true
+		}
 		resp := s.perProxyMessage(connUpdate, conn.proxyID)
 		if resp == nil {
 			log.Warnf("Token generation failed for proxy %s, disconnecting to force resync", conn.proxyID)
@@ -673,16 +888,20 @@ func (s *ProxyServiceServer) SendServiceUpdateToCluster(ctx context.Context, upd
 	}
 }
 
-// proxyAcceptsMapping returns whether the proxy should receive this mapping.
-// Old proxies that never reported capabilities are skipped for non-TLS L4
-// mappings with a custom listen port, since they don't understand the
-// protocol. Proxies that report capabilities (even SupportsCustomPorts=false)
-// are new enough to handle the mapping. TLS uses SNI routing and works on
-// any proxy. Delete operations are always sent so proxies can clean up.
+// proxyAcceptsMapping returns whether the proxy can receive this mapping.
+// Private mappings require SupportsPrivateService; custom-port L4 mappings
+// require SupportsCustomPorts. Remove operations always pass so proxies can
+// clean up.
 func proxyAcceptsMapping(conn *proxyConnection, mapping *proto.ProxyMapping) bool {
 	if mapping.Type == proto.ProxyMappingUpdateType_UPDATE_TYPE_REMOVED {
 		return true
 	}
+	if mapping.GetPrivate() {
+		caps := conn.capabilities
+		if caps == nil || caps.SupportsPrivateService == nil || !*caps.SupportsPrivateService {
+			return false
+		}
+	}
 	if mapping.ListenPort == 0 || mapping.Mode == "tls" {
 		return true
 	}
@@ -691,6 +910,29 @@ func proxyAcceptsMapping(conn *proxyConnection, mapping *proto.ProxyMapping) boo
 	return conn.capabilities != nil && conn.capabilities.SupportsCustomPorts != nil
 }
 
+// filterMappingsForProxy drops mappings the proxy cannot safely receive
+// (e.g. private mappings to a proxy without SupportsPrivateService).
+// Returns the input unchanged when no filtering is needed.
+func filterMappingsForProxy(conn *proxyConnection, update *proto.GetMappingUpdateResponse) *proto.GetMappingUpdateResponse {
+	if update == nil || len(update.Mapping) == 0 {
+		return update
+	}
+	kept := make([]*proto.ProxyMapping, 0, len(update.Mapping))
+	for _, m := range update.Mapping {
+		if !proxyAcceptsMapping(conn, m) {
+			continue
+		}
+		kept = append(kept, m)
+	}
+	if len(kept) == len(update.Mapping) {
+		return update
+	}
+	return &proto.GetMappingUpdateResponse{
+		Mapping:             kept,
+		InitialSyncComplete: update.InitialSyncComplete,
+	}
+}
+
 // perProxyMessage returns a copy of update with a fresh one-time token for
 // create/update operations. For delete operations the original mapping is
 // used unchanged because proxies do not need to authenticate for removal.
@@ -752,7 +994,10 @@ func (s *ProxyServiceServer) Authenticate(ctx context.Context, req *proto.Authen
 
 	authenticated, userId, method := s.authenticateRequest(ctx, req, service)
 
-	token, err := s.generateSessionToken(ctx, authenticated, service, userId, method)
+	// Non-OIDC schemes (PIN/Password/Header) authenticate against per-service
+	// secrets and have no user-level group context, so groups stay nil. Email
+	// is also empty — these schemes don't resolve a user record at sign time.
+	token, err := s.generateSessionToken(ctx, authenticated, service, userId, "", method, nil, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -841,7 +1086,7 @@ func (s *ProxyServiceServer) logAuthenticationError(ctx context.Context, err err
 	}
 }
 
-func (s *ProxyServiceServer) generateSessionToken(ctx context.Context, authenticated bool, service *rpservice.Service, userId string, method proxyauth.Method) (string, error) {
+func (s *ProxyServiceServer) generateSessionToken(ctx context.Context, authenticated bool, service *rpservice.Service, userId, userEmail string, method proxyauth.Method, groupIDs, groupNames []string) (string, error) {
 	if !authenticated || service.SessionPrivateKey == "" {
 		return "", nil
 	}
@@ -849,8 +1094,11 @@ func (s *ProxyServiceServer) generateSessionToken(ctx context.Context, authentic
 	token, err := sessionkey.SignToken(
 		service.SessionPrivateKey,
 		userId,
+		userEmail,
 		service.Domain,
 		method,
+		groupIDs,
+		groupNames,
 		proxyauth.DefaultSessionExpiry,
 	)
 	if err != nil {
@@ -861,6 +1109,26 @@ func (s *ProxyServiceServer) generateSessionToken(ctx context.Context, authentic
 	return token, nil
 }
 
+// pairGroupIDsAndNames splits a slice of resolved *types.Group records
+// into parallel id and name slices. ids[i] and names[i] always pair to
+// the same group. nil entries (orphan ids the manager couldn't resolve)
+// are skipped so the consumer can rely on positional pairing.
+func pairGroupIDsAndNames(groups []*types.Group) (ids, names []string) {
+	if len(groups) == 0 {
+		return nil, nil
+	}
+	ids = make([]string, 0, len(groups))
+	names = make([]string, 0, len(groups))
+	for _, g := range groups {
+		if g == nil {
+			continue
+		}
+		ids = append(ids, g.ID)
+		names = append(names, g.Name)
+	}
+	return ids, names
+}
+
 // SendStatusUpdate handles status updates from proxy clients.
 func (s *ProxyServiceServer) SendStatusUpdate(ctx context.Context, req *proto.SendStatusUpdateRequest) (*proto.SendStatusUpdateResponse, error) {
 	if err := enforceAccountScope(ctx, req.GetAccountId()); err != nil {
@@ -1125,7 +1393,9 @@ func (s *ProxyServiceServer) ValidateState(state string) (verifier, redirectURL
 	return verifier, redirectURL, nil
 }
 
-// GenerateSessionToken creates a signed session JWT for the given domain and user.
+// GenerateSessionToken creates a signed session JWT for the given domain and
+// user. The user's group memberships are embedded in the token so policy-aware
+// middlewares on the proxy can authorise without an extra management round-trip.
 func (s *ProxyServiceServer) GenerateSessionToken(ctx context.Context, domain, userID string, method proxyauth.Method) (string, error) {
 	service, err := s.getServiceByDomain(ctx, domain)
 	if err != nil {
@@ -1136,11 +1406,29 @@ func (s *ProxyServiceServer) GenerateSessionToken(ctx context.Context, domain, u
 		return "", fmt.Errorf("no session key configured for domain: %s", domain)
 	}
 
+	var (
+		email      string
+		groupIDs   []string
+		groupNames []string
+	)
+	if s.usersManager != nil {
+		user, userGroups, uerr := s.usersManager.GetUserWithGroups(ctx, userID)
+		if uerr != nil {
+			log.WithContext(ctx).Debugf("session token mint: lookup user %s: %v", userID, uerr)
+		} else if user != nil {
+			email = user.Email
+			groupIDs, groupNames = pairGroupIDsAndNames(userGroups)
+		}
+	}
+
 	return sessionkey.SignToken(
 		service.SessionPrivateKey,
 		userID,
+		email,
 		domain,
 		method,
+		groupIDs,
+		groupNames,
 		proxyauth.DefaultSessionExpiry,
 	)
 }
@@ -1244,7 +1532,7 @@ func (s *ProxyServiceServer) ValidateSession(ctx context.Context, req *proto.Val
 		}, nil
 	}
 
-	userID, _, err := proxyauth.ValidateSessionJWT(sessionToken, domain, pubKeyBytes)
+	userID, _, _, _, _, err := proxyauth.ValidateSessionJWT(sessionToken, domain, pubKeyBytes)
 	if err != nil {
 		log.WithFields(log.Fields{
 			"domain": domain,
@@ -1257,7 +1545,7 @@ func (s *ProxyServiceServer) ValidateSession(ctx context.Context, req *proto.Val
 		}, nil
 	}
 
-	user, err := s.usersManager.GetUser(ctx, userID)
+	user, userGroups, err := s.usersManager.GetUserWithGroups(ctx, userID)
 	if err != nil {
 		log.WithFields(log.Fields{
 			"domain":  domain,
@@ -1291,12 +1579,15 @@ func (s *ProxyServiceServer) ValidateSession(ctx context.Context, req *proto.Val
 			"user_id": userID,
 			"error":   err.Error(),
 		}).Debug("ValidateSession: access denied")
+		groupIDs, groupNames := pairGroupIDsAndNames(userGroups)
 		//nolint:nilerr
 		return &proto.ValidateSessionResponse{
-			Valid:        false,
-			UserId:       user.Id,
-			UserEmail:    user.Email,
-			DeniedReason: "not_in_group",
+			Valid:          false,
+			UserId:         user.Id,
+			UserEmail:      user.Email,
+			DeniedReason:   "not_in_group",
+			PeerGroupIds:   groupIDs,
+			PeerGroupNames: groupNames,
 		}, nil
 	}
 
@@ -1306,10 +1597,13 @@ func (s *ProxyServiceServer) ValidateSession(ctx context.Context, req *proto.Val
 		"email":   user.Email,
 	}).Debug("ValidateSession: access granted")
 
+	groupIDs, groupNames := pairGroupIDsAndNames(userGroups)
 	return &proto.ValidateSessionResponse{
-		Valid:     true,
-		UserId:    user.Id,
-		UserEmail: user.Email,
+		Valid:          true,
+		UserId:         user.Id,
+		UserEmail:      user.Email,
+		PeerGroupIds:   groupIDs,
+		PeerGroupNames: groupNames,
 	}, nil
 }
 
@@ -1342,3 +1636,154 @@ func (s *ProxyServiceServer) checkGroupAccess(service *rpservice.Service, user *
 }
 
 func ptr[T any](v T) *T { return &v }
+
+// ValidateTunnelPeer resolves an inbound peer by its WireGuard tunnel IP and
+// checks the peer's group membership against the service's access groups.
+// Peers without a user (machine agents, automation workloads) are first-class
+// callers; authorisation runs off peer-group memberships rather than the
+// optional owning user's auto-groups. On success a session JWT is minted so
+// the proxy can install a cookie and skip subsequent management round-trips.
+func (s *ProxyServiceServer) ValidateTunnelPeer(ctx context.Context, req *proto.ValidateTunnelPeerRequest) (*proto.ValidateTunnelPeerResponse, error) {
+	domain := req.GetDomain()
+	tunnelIPStr := req.GetTunnelIp()
+
+	if domain == "" || tunnelIPStr == "" {
+		return &proto.ValidateTunnelPeerResponse{
+			Valid:        false,
+			DeniedReason: "missing domain or tunnel_ip",
+		}, nil
+	}
+
+	tunnelIP := net.ParseIP(tunnelIPStr)
+	if tunnelIP == nil {
+		return &proto.ValidateTunnelPeerResponse{
+			Valid:        false,
+			DeniedReason: "invalid_tunnel_ip",
+		}, nil
+	}
+
+	service, err := s.getServiceByDomain(ctx, domain)
+	if err != nil {
+		log.WithFields(log.Fields{"domain": domain, "error": err.Error()}).Debug("ValidateTunnelPeer: service not found")
+		//nolint:nilerr
+		return &proto.ValidateTunnelPeerResponse{
+			Valid:        false,
+			DeniedReason: "service_not_found",
+		}, nil
+	}
+
+	// Mirror ValidateSession: account-scoped (BYOP) proxy tokens may only
+	// validate and mint session cookies for their own account's domains.
+	if err := enforceAccountScope(ctx, service.AccountID); err != nil {
+		return nil, err
+	}
+
+	peer, err := s.peersManager.GetPeerByTunnelIP(ctx, service.AccountID, tunnelIP)
+	if err != nil || peer == nil {
+		log.WithFields(log.Fields{"domain": domain, "tunnel_ip": tunnelIPStr}).Debug("ValidateTunnelPeer: peer not found")
+		//nolint:nilerr
+		return &proto.ValidateTunnelPeerResponse{
+			Valid:        false,
+			DeniedReason: "peer_not_found",
+		}, nil
+	}
+
+	_, peerGroups, err := s.peersManager.GetPeerWithGroups(ctx, service.AccountID, peer.ID)
+	if err != nil {
+		log.WithFields(log.Fields{"domain": domain, "peer_id": peer.ID, "error": err.Error()}).Debug("ValidateTunnelPeer: peer groups lookup failed")
+		//nolint:nilerr
+		return &proto.ValidateTunnelPeerResponse{
+			Valid:        false,
+			DeniedReason: "peer_not_found",
+		}, nil
+	}
+
+	groupIDs, groupNames := pairGroupIDsAndNames(peerGroups)
+
+	// Resolve the principal: when the peer is linked to a user, the human
+	// is the principal so multiple peers owned by the same user share a
+	// single identity. Unlinked peers (machine agents) are their own
+	// principal keyed on peer.ID. displayIdentity is what upstream gateways
+	// tag spend with — user.Email when linked, peer.Name when not.
+	principalID := peer.ID
+	displayIdentity := peer.Name
+	if peer.UserID != "" {
+		if user, uerr := s.usersManager.GetUser(ctx, peer.UserID); uerr == nil && user != nil {
+			principalID = user.Id
+			if user.Email != "" {
+				displayIdentity = user.Email
+			}
+		}
+	}
+
+	if err := checkPeerGroupAccess(service, groupIDs); err != nil {
+		log.WithFields(log.Fields{"domain": domain, "peer_id": peer.ID, "error": err.Error()}).Debug("ValidateTunnelPeer: access denied")
+		//nolint:nilerr
+		return &proto.ValidateTunnelPeerResponse{
+			Valid:          false,
+			UserId:         principalID,
+			UserEmail:      displayIdentity,
+			DeniedReason:   "not_in_group",
+			PeerGroupIds:   groupIDs,
+			PeerGroupNames: groupNames,
+		}, nil
+	}
+
+	token, err := s.generateSessionToken(ctx, true, service, principalID, displayIdentity, proxyauth.MethodOIDC, groupIDs, groupNames)
+	if err != nil {
+		return nil, err
+	}
+
+	log.WithFields(log.Fields{
+		"domain":       domain,
+		"tunnel_ip":    tunnelIPStr,
+		"peer_id":      peer.ID,
+		"principal_id": principalID,
+	}).Debug("ValidateTunnelPeer: access granted")
+
+	return &proto.ValidateTunnelPeerResponse{
+		Valid:          true,
+		UserId:         principalID,
+		UserEmail:      displayIdentity,
+		SessionToken:   token,
+		PeerGroupIds:   groupIDs,
+		PeerGroupNames: groupNames,
+	}, nil
+}
+
+// checkPeerGroupAccess gates ValidateTunnelPeer by the service's required
+// groups. Private services authorise against AccessGroups (empty list fails
+// closed — Validate() rejects that at save time but the RPC is the security
+// boundary and must not trust upstream state). Bearer-auth services authorise
+// against DistributionGroups when populated. Non-private non-bearer services
+// are open.
+func checkPeerGroupAccess(service *rpservice.Service, peerGroupIDs []string) error {
+	if service.Private {
+		if len(service.AccessGroups) == 0 {
+			return fmt.Errorf("private service has no access groups")
+		}
+		return matchAnyGroup(service.AccessGroups, peerGroupIDs)
+	}
+	if service.Auth.BearerAuth != nil && service.Auth.BearerAuth.Enabled && len(service.Auth.BearerAuth.DistributionGroups) > 0 {
+		return matchAnyGroup(service.Auth.BearerAuth.DistributionGroups, peerGroupIDs)
+	}
+	return nil
+}
+
+// matchAnyGroup returns nil when peerGroupIDs intersects allowedGroups,
+// else a non-nil error.
+func matchAnyGroup(allowedGroups, peerGroupIDs []string) error {
+	if len(allowedGroups) == 0 {
+		return fmt.Errorf("no allowed groups configured")
+	}
+	allowed := make(map[string]struct{}, len(allowedGroups))
+	for _, g := range allowedGroups {
+		allowed[g] = struct{}{}
+	}
+	for _, g := range peerGroupIDs {
+		if _, ok := allowed[g]; ok {
+			return nil
+		}
+	}
+	return fmt.Errorf("peer not in allowed groups")
+}

management/internals/shared/grpc/proxy_group_access_test.go

@@ -109,7 +109,7 @@ func (m *mockReverseProxyManager) GetServiceByDomain(_ context.Context, domain s
 	return nil, errors.New("service not found for domain: " + domain)
 }
 
-func (m *mockReverseProxyManager) GetActiveClusters(_ context.Context, _, _ string) ([]proxy.Cluster, error) {
+func (m *mockReverseProxyManager) GetClusters(_ context.Context, _, _ string) ([]proxy.Cluster, error) {
 	return nil, nil
 }
 
@@ -129,6 +129,14 @@ func (m *mockUsersManager) GetUser(ctx context.Context, userID string) (*types.U
 	return user, nil
 }
 
+func (m *mockUsersManager) GetUserWithGroups(ctx context.Context, userID string) (*types.User, []*types.Group, error) {
+	user, err := m.GetUser(ctx, userID)
+	if err != nil {
+		return nil, nil, err
+	}
+	return user, nil, nil
+}
+
 func TestValidateUserGroupAccess(t *testing.T) {
 	tests := []struct {
 		name             string
@@ -420,3 +428,46 @@ func TestGetAccountProxyByDomain(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckPeerGroupAccess(t *testing.T) {
+	t.Run("private with empty AccessGroups denies", func(t *testing.T) {
+		svc := &service.Service{Private: true, AccessGroups: nil}
+		err := checkPeerGroupAccess(svc, []string{"grp-admins"})
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "no access groups")
+	})
+
+	t.Run("private with peer in AccessGroups allows", func(t *testing.T) {
+		svc := &service.Service{Private: true, AccessGroups: []string{"grp-admins", "grp-ops"}}
+		assert.NoError(t, checkPeerGroupAccess(svc, []string{"grp-other", "grp-ops"}))
+	})
+
+	t.Run("private with peer outside AccessGroups denies", func(t *testing.T) {
+		svc := &service.Service{Private: true, AccessGroups: []string{"grp-admins"}}
+		assert.Error(t, checkPeerGroupAccess(svc, []string{"grp-other"}))
+	})
+
+	t.Run("bearer enabled with empty DistributionGroups allows", func(t *testing.T) {
+		svc := &service.Service{
+			Auth: service.AuthConfig{BearerAuth: &service.BearerAuthConfig{Enabled: true}},
+		}
+		assert.NoError(t, checkPeerGroupAccess(svc, []string{"grp-anyone"}))
+	})
+
+	t.Run("bearer enabled gates on DistributionGroups", func(t *testing.T) {
+		svc := &service.Service{
+			Auth: service.AuthConfig{
+				BearerAuth: &service.BearerAuthConfig{
+					Enabled:            true,
+					DistributionGroups: []string{"grp-allowed"},
+				},
+			},
+		}
+		assert.NoError(t, checkPeerGroupAccess(svc, []string{"grp-allowed"}))
+		assert.Error(t, checkPeerGroupAccess(svc, []string{"grp-other"}))
+	})
+
+	t.Run("non-private non-bearer is open", func(t *testing.T) {
+		assert.NoError(t, checkPeerGroupAccess(&service.Service{}, nil))
+	})
+}

management/internals/shared/grpc/sync_mappings_test.go

@@ -0,0 +1,411 @@
+package grpc
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+
+	rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
+	"github.com/netbirdio/netbird/shared/management/proto"
+)
+
+// syncRecordingStream is a mock ProxyService_SyncMappingsServer that records
+// sent messages and returns pre-loaded ack responses from Recv.
+type syncRecordingStream struct {
+	grpc.ServerStream
+
+	mu       sync.Mutex
+	sent     []*proto.SyncMappingsResponse
+	recvMsgs []*proto.SyncMappingsRequest
+	recvIdx  int
+}
+
+func (s *syncRecordingStream) Send(m *proto.SyncMappingsResponse) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.sent = append(s.sent, m)
+	return nil
+}
+
+func (s *syncRecordingStream) Recv() (*proto.SyncMappingsRequest, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.recvIdx >= len(s.recvMsgs) {
+		return nil, fmt.Errorf("no more recv messages")
+	}
+	msg := s.recvMsgs[s.recvIdx]
+	s.recvIdx++
+	return msg, nil
+}
+
+func (s *syncRecordingStream) Context() context.Context     { return context.Background() }
+func (s *syncRecordingStream) SetHeader(metadata.MD) error  { return nil }
+func (s *syncRecordingStream) SendHeader(metadata.MD) error { return nil }
+func (s *syncRecordingStream) SetTrailer(metadata.MD)       {}
+func (s *syncRecordingStream) SendMsg(any) error            { return nil }
+func (s *syncRecordingStream) RecvMsg(any) error            { return nil }
+
+func ackMsg() *proto.SyncMappingsRequest {
+	return &proto.SyncMappingsRequest{
+		Msg: &proto.SyncMappingsRequest_Ack{Ack: &proto.SyncMappingsAck{}},
+	}
+}
+
+func TestSendSnapshotSync_BatchesWithAcks(t *testing.T) {
+	const cluster = "cluster.example.com"
+	const batchSize = 3
+	const totalServices = 7 // 3 + 3 + 1 → 3 batches, 3 acks (one per batch, including final)
+
+	ctrl := gomock.NewController(t)
+	mgr := rpservice.NewMockManager(ctrl)
+	mgr.EXPECT().GetGlobalServices(gomock.Any()).Return(makeServices(totalServices, cluster), nil)
+
+	s := newSnapshotTestServer(t, batchSize)
+	s.serviceManager = mgr
+
+	stream := &syncRecordingStream{
+		recvMsgs: []*proto.SyncMappingsRequest{ackMsg(), ackMsg(), ackMsg()},
+	}
+	conn := &proxyConnection{
+		proxyID:    "proxy-a",
+		address:    cluster,
+		syncStream: stream,
+	}
+
+	err := s.sendSnapshotSync(context.Background(), conn, stream)
+	require.NoError(t, err)
+
+	require.Len(t, stream.sent, 3, "should send ceil(7/3) = 3 batches")
+
+	assert.Len(t, stream.sent[0].Mapping, 3)
+	assert.False(t, stream.sent[0].InitialSyncComplete)
+
+	assert.Len(t, stream.sent[1].Mapping, 3)
+	assert.False(t, stream.sent[1].InitialSyncComplete)
+
+	assert.Len(t, stream.sent[2].Mapping, 1)
+	assert.True(t, stream.sent[2].InitialSyncComplete)
+
+	// All 3 acks consumed — including the final batch.
+	assert.Equal(t, 3, stream.recvIdx)
+}
+
+func TestSendSnapshotSync_SingleBatchWaitsForAck(t *testing.T) {
+	const cluster = "cluster.example.com"
+	const batchSize = 100
+	const totalServices = 5
+
+	ctrl := gomock.NewController(t)
+	mgr := rpservice.NewMockManager(ctrl)
+	mgr.EXPECT().GetGlobalServices(gomock.Any()).Return(makeServices(totalServices, cluster), nil)
+
+	s := newSnapshotTestServer(t, batchSize)
+	s.serviceManager = mgr
+
+	stream := &syncRecordingStream{
+		recvMsgs: []*proto.SyncMappingsRequest{ackMsg()},
+	}
+	conn := &proxyConnection{
+		proxyID:    "proxy-a",
+		address:    cluster,
+		syncStream: stream,
+	}
+
+	err := s.sendSnapshotSync(context.Background(), conn, stream)
+	require.NoError(t, err)
+
+	require.Len(t, stream.sent, 1)
+	assert.Len(t, stream.sent[0].Mapping, totalServices)
+	assert.True(t, stream.sent[0].InitialSyncComplete)
+	assert.Equal(t, 1, stream.recvIdx, "final batch ack must be consumed")
+}
+
+func TestSendSnapshotSync_EmptySnapshot(t *testing.T) {
+	const cluster = "cluster.example.com"
+
+	ctrl := gomock.NewController(t)
+	mgr := rpservice.NewMockManager(ctrl)
+	mgr.EXPECT().GetGlobalServices(gomock.Any()).Return(nil, nil)
+
+	s := newSnapshotTestServer(t, 500)
+	s.serviceManager = mgr
+
+	stream := &syncRecordingStream{
+		recvMsgs: []*proto.SyncMappingsRequest{ackMsg()},
+	}
+	conn := &proxyConnection{
+		proxyID:    "proxy-a",
+		address:    cluster,
+		syncStream: stream,
+	}
+
+	err := s.sendSnapshotSync(context.Background(), conn, stream)
+	require.NoError(t, err)
+
+	require.Len(t, stream.sent, 1, "empty snapshot must still send sync-complete")
+	assert.Empty(t, stream.sent[0].Mapping)
+	assert.True(t, stream.sent[0].InitialSyncComplete)
+	assert.Equal(t, 1, stream.recvIdx, "empty snapshot ack must be consumed")
+}
+
+func TestSendSnapshotSync_MissingAckReturnsError(t *testing.T) {
+	const cluster = "cluster.example.com"
+	const batchSize = 2
+	const totalServices = 4 // 2 batches → 1 ack needed, but we provide none
+
+	ctrl := gomock.NewController(t)
+	mgr := rpservice.NewMockManager(ctrl)
+	mgr.EXPECT().GetGlobalServices(gomock.Any()).Return(makeServices(totalServices, cluster), nil)
+
+	s := newSnapshotTestServer(t, batchSize)
+	s.serviceManager = mgr
+
+	// No acks available — Recv will return error.
+	stream := &syncRecordingStream{}
+	conn := &proxyConnection{
+		proxyID:    "proxy-a",
+		address:    cluster,
+		syncStream: stream,
+	}
+
+	err := s.sendSnapshotSync(context.Background(), conn, stream)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "receive ack")
+	// First batch should have been sent before the error.
+	require.Len(t, stream.sent, 1)
+}
+
+func TestSendSnapshotSync_WrongMessageInsteadOfAck(t *testing.T) {
+	const cluster = "cluster.example.com"
+	const batchSize = 2
+	const totalServices = 4
+
+	ctrl := gomock.NewController(t)
+	mgr := rpservice.NewMockManager(ctrl)
+	mgr.EXPECT().GetGlobalServices(gomock.Any()).Return(makeServices(totalServices, cluster), nil)
+
+	s := newSnapshotTestServer(t, batchSize)
+	s.serviceManager = mgr
+
+	// Send an init message instead of an ack.
+	stream := &syncRecordingStream{
+		recvMsgs: []*proto.SyncMappingsRequest{
+			{Msg: &proto.SyncMappingsRequest_Init{Init: &proto.SyncMappingsInit{ProxyId: "bad"}}},
+		},
+	}
+	conn := &proxyConnection{
+		proxyID:    "proxy-a",
+		address:    cluster,
+		syncStream: stream,
+	}
+
+	err := s.sendSnapshotSync(context.Background(), conn, stream)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "expected ack")
+}
+
+func TestSendSnapshotSync_BackPressureOrdering(t *testing.T) {
+	// Verify batches are sent strictly sequentially — batch N+1 is not sent
+	// until the ack for batch N is received, including the final batch.
+	const cluster = "cluster.example.com"
+	const batchSize = 2
+	const totalServices = 6 // 3 batches, 3 acks
+
+	ctrl := gomock.NewController(t)
+	mgr := rpservice.NewMockManager(ctrl)
+	mgr.EXPECT().GetGlobalServices(gomock.Any()).Return(makeServices(totalServices, cluster), nil)
+
+	s := newSnapshotTestServer(t, batchSize)
+	s.serviceManager = mgr
+
+	var mu sync.Mutex
+	var events []string
+
+	// Build a stream that logs send/recv events so we can verify ordering.
+	ackCh := make(chan struct{}, 3)
+	stream := &orderTrackingStream{
+		mu:     &mu,
+		events: &events,
+		ackCh:  ackCh,
+	}
+	conn := &proxyConnection{
+		proxyID:    "proxy-a",
+		address:    cluster,
+		syncStream: stream,
+	}
+
+	// Feed acks asynchronously after a short delay to simulate real proxy.
+	go func() {
+		for range 3 {
+			time.Sleep(10 * time.Millisecond)
+			ackCh <- struct{}{}
+		}
+	}()
+
+	err := s.sendSnapshotSync(context.Background(), conn, stream)
+	require.NoError(t, err)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	// Expected: send, recv-ack, send, recv-ack, send, recv-ack.
+	require.Len(t, events, 6)
+	assert.Equal(t, "send", events[0])
+	assert.Equal(t, "recv", events[1])
+	assert.Equal(t, "send", events[2])
+	assert.Equal(t, "recv", events[3])
+	assert.Equal(t, "send", events[4])
+	assert.Equal(t, "recv", events[5])
+}
+
+// orderTrackingStream logs "send" and "recv" events and blocks Recv until
+// an ack is signaled via ackCh.
+type orderTrackingStream struct {
+	grpc.ServerStream
+	mu     *sync.Mutex
+	events *[]string
+	ackCh  chan struct{}
+}
+
+func (s *orderTrackingStream) Send(_ *proto.SyncMappingsResponse) error {
+	s.mu.Lock()
+	*s.events = append(*s.events, "send")
+	s.mu.Unlock()
+	return nil
+}
+
+func (s *orderTrackingStream) Recv() (*proto.SyncMappingsRequest, error) {
+	<-s.ackCh
+	s.mu.Lock()
+	*s.events = append(*s.events, "recv")
+	s.mu.Unlock()
+	return ackMsg(), nil
+}
+
+func (s *orderTrackingStream) Context() context.Context     { return context.Background() }
+func (s *orderTrackingStream) SetHeader(metadata.MD) error  { return nil }
+func (s *orderTrackingStream) SendHeader(metadata.MD) error { return nil }
+func (s *orderTrackingStream) SetTrailer(metadata.MD)       {}
+func (s *orderTrackingStream) SendMsg(any) error            { return nil }
+func (s *orderTrackingStream) RecvMsg(any) error            { return nil }
+
+func TestSendSnapshotSync_TokensGeneratedPerBatch(t *testing.T) {
+	const cluster = "cluster.example.com"
+	const batchSize = 2
+	const totalServices = 4
+	const ttl = 100 * time.Millisecond
+	const ackDelay = 200 * time.Millisecond
+
+	ctrl := gomock.NewController(t)
+	mgr := rpservice.NewMockManager(ctrl)
+	mgr.EXPECT().GetGlobalServices(gomock.Any()).Return(makeServices(totalServices, cluster), nil)
+
+	s := newSnapshotTestServer(t, batchSize)
+	s.serviceManager = mgr
+	s.tokenTTL = ttl
+
+	// Build a stream that validates tokens immediately on Send, then
+	// delays the ack to ensure the next batch's tokens are generated fresh.
+	var validateErrs []error
+	ackCh := make(chan struct{}, 2)
+	stream := &tokenValidatingSyncStream{
+		tokenStore:   s.tokenStore,
+		validateErrs: &validateErrs,
+		ackCh:        ackCh,
+	}
+	conn := &proxyConnection{
+		proxyID:    "proxy-a",
+		address:    cluster,
+		syncStream: stream,
+	}
+
+	go func() {
+		// Delay first ack so that if tokens were all generated upfront they'd expire.
+		time.Sleep(ackDelay)
+		ackCh <- struct{}{}
+		// Final batch ack — immediate.
+		ackCh <- struct{}{}
+	}()
+
+	err := s.sendSnapshotSync(context.Background(), conn, stream)
+	require.NoError(t, err)
+	require.Empty(t, validateErrs,
+		"tokens must remain valid: per-batch generation guarantees freshness")
+}
+
+type tokenValidatingSyncStream struct {
+	grpc.ServerStream
+	tokenStore   *OneTimeTokenStore
+	validateErrs *[]error
+	ackCh        chan struct{}
+}
+
+func (s *tokenValidatingSyncStream) Send(m *proto.SyncMappingsResponse) error {
+	for _, mapping := range m.Mapping {
+		if err := s.tokenStore.ValidateAndConsume(mapping.AuthToken, mapping.AccountId, mapping.Id); err != nil {
+			*s.validateErrs = append(*s.validateErrs, fmt.Errorf("svc %s: %w", mapping.Id, err))
+		}
+	}
+	return nil
+}
+
+func (s *tokenValidatingSyncStream) Recv() (*proto.SyncMappingsRequest, error) {
+	<-s.ackCh
+	return ackMsg(), nil
+}
+
+func (s *tokenValidatingSyncStream) Context() context.Context     { return context.Background() }
+func (s *tokenValidatingSyncStream) SetHeader(metadata.MD) error  { return nil }
+func (s *tokenValidatingSyncStream) SendHeader(metadata.MD) error { return nil }
+func (s *tokenValidatingSyncStream) SetTrailer(metadata.MD)       {}
+func (s *tokenValidatingSyncStream) SendMsg(any) error            { return nil }
+func (s *tokenValidatingSyncStream) RecvMsg(any) error            { return nil }
+
+func TestConnectionSendResponse_RoutesToSyncStream(t *testing.T) {
+	stream := &syncRecordingStream{}
+	conn := &proxyConnection{
+		syncStream: stream,
+	}
+
+	resp := &proto.GetMappingUpdateResponse{
+		Mapping: []*proto.ProxyMapping{
+			{Id: "svc-1", AccountId: "acct-1", Domain: "example.com"},
+		},
+		InitialSyncComplete: true,
+	}
+
+	err := conn.sendResponse(resp)
+	require.NoError(t, err)
+
+	require.Len(t, stream.sent, 1)
+	assert.Len(t, stream.sent[0].Mapping, 1)
+	assert.Equal(t, "svc-1", stream.sent[0].Mapping[0].Id)
+	assert.True(t, stream.sent[0].InitialSyncComplete)
+}
+
+func TestConnectionSendResponse_RoutesToLegacyStream(t *testing.T) {
+	stream := &recordingStream{}
+	conn := &proxyConnection{
+		stream: stream,
+	}
+
+	resp := &proto.GetMappingUpdateResponse{
+		Mapping: []*proto.ProxyMapping{
+			{Id: "svc-2", AccountId: "acct-2"},
+		},
+	}
+
+	err := conn.sendResponse(resp)
+	require.NoError(t, err)
+
+	require.Len(t, stream.messages, 1)
+	assert.Equal(t, "svc-2", stream.messages[0].Mapping[0].Id)
+}

management/internals/shared/grpc/validate_session_test.go

@@ -102,7 +102,7 @@ func generateSessionKeyPair(t *testing.T) (string, string) {
 
 func createSessionToken(t *testing.T, privKeyB64, userID, domain string) string {
 	t.Helper()
-	token, err := sessionkey.SignToken(privKeyB64, userID, domain, auth.MethodOIDC, time.Hour)
+	token, err := sessionkey.SignToken(privKeyB64, userID, domain, auth.MethodOIDC, nil, time.Hour)
 	require.NoError(t, err)
 	return token
 }
@@ -125,6 +125,7 @@ func TestValidateSession_UserAllowed(t *testing.T) {
 	assert.True(t, resp.Valid, "User should be allowed access")
 	assert.Equal(t, "allowedUserId", resp.UserId)
 	assert.Empty(t, resp.DeniedReason)
+	assert.Equal(t, []string{"allowedGroupId"}, resp.GetPeerGroupIds(), "PeerGroupIds must mirror the resolved user's group memberships")
 }
 
 func TestValidateSession_UserNotInAllowedGroup(t *testing.T) {
@@ -145,6 +146,7 @@ func TestValidateSession_UserNotInAllowedGroup(t *testing.T) {
 	assert.False(t, resp.Valid, "User not in group should be denied")
 	assert.Equal(t, "not_in_group", resp.DeniedReason)
 	assert.Equal(t, "nonGroupUserId", resp.UserId)
+	assert.Empty(t, resp.GetPeerGroupIds(), "PeerGroupIds must mirror the resolved user's actual (empty) memberships on denial")
 }
 
 func TestValidateSession_UserInDifferentAccount(t *testing.T) {
@@ -322,7 +324,7 @@ func (m *testValidateSessionServiceManager) GetServiceByDomain(ctx context.Conte
 	return m.store.GetServiceByDomain(ctx, domain)
 }
 
-func (m *testValidateSessionServiceManager) GetActiveClusters(_ context.Context, _, _ string) ([]proxy.Cluster, error) {
+func (m *testValidateSessionServiceManager) GetClusters(_ context.Context, _, _ string) ([]proxy.Cluster, error) {
 	return nil, nil
 }