Merge branch 'main' into refactor-combined

management/internals/modules/peers/ephemeral/manager/ephemeral.go

@@ -11,6 +11,7 @@ import (
 	"github.com/netbirdio/netbird/management/internals/modules/peers/ephemeral"
 	"github.com/netbirdio/netbird/management/server/activity"
 	nbpeer "github.com/netbirdio/netbird/management/server/peer"
+	"github.com/netbirdio/netbird/management/server/telemetry"
 
 	"github.com/netbirdio/netbird/management/server/store"
 )
@@ -47,6 +48,11 @@ type EphemeralManager struct {
 
 	lifeTime      time.Duration
 	cleanupWindow time.Duration
+
+	// metrics is nil-safe; methods on telemetry.EphemeralPeersMetrics
+	// no-op when the receiver is nil so deployments without an app
+	// metrics provider work unchanged.
+	metrics *telemetry.EphemeralPeersMetrics
 }
 
 // NewEphemeralManager instantiate new EphemeralManager
@@ -60,6 +66,15 @@ func NewEphemeralManager(store store.Store, peersManager peers.Manager) *Ephemer
 	}
 }
 
+// SetMetrics attaches a metrics collector. Safe to call once before
+// LoadInitialPeers; later attachment is fine but earlier loads won't be
+// reflected in the gauge. Pass nil to detach.
+func (e *EphemeralManager) SetMetrics(m *telemetry.EphemeralPeersMetrics) {
+	e.peersLock.Lock()
+	e.metrics = m
+	e.peersLock.Unlock()
+}
+
 // LoadInitialPeers load from the database the ephemeral type of peers and schedule a cleanup procedure to the head
 // of the linked list (to the most deprecated peer). At the end of cleanup it schedules the next cleanup to the new
 // head.
@@ -97,7 +112,9 @@ func (e *EphemeralManager) OnPeerConnected(ctx context.Context, peer *nbpeer.Pee
 	e.peersLock.Lock()
 	defer e.peersLock.Unlock()
 
-	e.removePeer(peer.ID)
+	if e.removePeer(peer.ID) {
+		e.metrics.DecPending(1)
+	}
 
 	// stop the unnecessary timer
 	if e.headPeer == nil && e.timer != nil {
@@ -123,6 +140,7 @@ func (e *EphemeralManager) OnPeerDisconnected(ctx context.Context, peer *nbpeer.
 	}
 
 	e.addPeer(peer.AccountID, peer.ID, e.newDeadLine())
+	e.metrics.IncPending()
 	if e.timer == nil {
 		delay := e.headPeer.deadline.Sub(timeNow()) + e.cleanupWindow
 		if delay < 0 {
@@ -145,6 +163,7 @@ func (e *EphemeralManager) loadEphemeralPeers(ctx context.Context) {
 	for _, p := range peers {
 		e.addPeer(p.AccountID, p.ID, t)
 	}
+	e.metrics.AddPending(int64(len(peers)))
 
 	log.WithContext(ctx).Debugf("loaded ephemeral peer(s): %d", len(peers))
 }
@@ -181,6 +200,15 @@ func (e *EphemeralManager) cleanup(ctx context.Context) {
 
 	e.peersLock.Unlock()
 
+	// Drop the gauge by the number of entries we just took off the list,
+	// regardless of whether the subsequent DeletePeers call succeeds. The
+	// list invariant is what the gauge tracks; failed delete batches are
+	// counted separately via CountCleanupError so we can still see them.
+	if len(deletePeers) > 0 {
+		e.metrics.CountCleanupRun()
+		e.metrics.DecPending(int64(len(deletePeers)))
+	}
+
 	peerIDsPerAccount := make(map[string][]string)
 	for id, p := range deletePeers {
 		peerIDsPerAccount[p.accountID] = append(peerIDsPerAccount[p.accountID], id)
@@ -191,7 +219,10 @@ func (e *EphemeralManager) cleanup(ctx context.Context) {
 		err := e.peersManager.DeletePeers(ctx, accountID, peerIDs, activity.SystemInitiator, true)
 		if err != nil {
 			log.WithContext(ctx).Errorf("failed to delete ephemeral peers: %s", err)
+			e.metrics.CountCleanupError()
+			continue
 		}
+		e.metrics.CountPeersCleaned(int64(len(peerIDs)))
 	}
 }
 
@@ -211,9 +242,12 @@ func (e *EphemeralManager) addPeer(accountID string, peerID string, deadline tim
 	e.tailPeer = ep
 }
 
-func (e *EphemeralManager) removePeer(id string) {
+// removePeer drops the entry from the linked list. Returns true if a
+// matching entry was found and removed so callers can keep the pending
+// metric gauge in sync.
+func (e *EphemeralManager) removePeer(id string) bool {
 	if e.headPeer == nil {
-		return
+		return false
 	}
 
 	if e.headPeer.id == id {
@@ -221,7 +255,7 @@ func (e *EphemeralManager) removePeer(id string) {
 		if e.tailPeer.id == id {
 			e.tailPeer = nil
 		}
-		return
+		return true
 	}
 
 	for p := e.headPeer; p.next != nil; p = p.next {
@@ -231,9 +265,10 @@ func (e *EphemeralManager) removePeer(id string) {
 				e.tailPeer = p
 			}
 			p.next = p.next.next
-			return
+			return true
 		}
 	}
+	return false
 }
 
 func (e *EphemeralManager) isPeerOnList(id string) bool {

management/internals/modules/peers/manager.go

@@ -5,6 +5,7 @@ package peers
 import (
 	"context"
 	"fmt"
+	"net"
 	"time"
 
 	"github.com/rs/xid"
@@ -35,6 +36,14 @@ type Manager interface {
 	SetAccountManager(accountManager account.Manager)
 	GetPeerID(ctx context.Context, peerKey string) (string, error)
 	CreateProxyPeer(ctx context.Context, accountID string, peerKey string, cluster string) error
+	// GetPeerByTunnelIP looks up a peer in accountID by its WireGuard tunnel IP.
+	// Returns nil with an error when no match exists. No permission check;
+	// callers (the proxy's ValidateTunnelPeer RPC) are trusted server components.
+	GetPeerByTunnelIP(ctx context.Context, accountID string, ip net.IP) (*peer.Peer, error)
+	// GetPeerWithGroups returns the peer and the list of *types.Group it belongs
+	// to. Used by the proxy's auth path to authorise a request by the calling
+	// peer's group memberships.
+	GetPeerWithGroups(ctx context.Context, accountID, peerID string) (*peer.Peer, []*types.Group, error)
 }
 
 type managerImpl struct {
@@ -99,6 +108,26 @@ func (m *managerImpl) GetPeersByGroupIDs(ctx context.Context, accountID string,
 	return m.store.GetPeersByGroupIDs(ctx, accountID, groupsIDs)
 }
 
+// GetPeerByTunnelIP delegates to the store's indexed lookup.
+func (m *managerImpl) GetPeerByTunnelIP(ctx context.Context, accountID string, ip net.IP) (*peer.Peer, error) {
+	return m.store.GetPeerByIP(ctx, store.LockingStrengthNone, accountID, ip)
+}
+
+// GetPeerWithGroups returns the peer plus its group memberships. Any store
+// error returns (nil, nil, err) so callers never receive a valid peer
+// alongside a non-nil error.
+func (m *managerImpl) GetPeerWithGroups(ctx context.Context, accountID, peerID string) (*peer.Peer, []*types.Group, error) {
+	p, err := m.store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
+	if err != nil {
+		return nil, nil, err
+	}
+	groups, err := m.store.GetPeerGroups(ctx, store.LockingStrengthNone, accountID, peerID)
+	if err != nil {
+		return nil, nil, err
+	}
+	return p, groups, nil
+}
+
 func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 	settings, err := m.store.GetAccountSettings(ctx, store.LockingStrengthNone, accountID)
 	if err != nil {

management/internals/modules/peers/manager_mock.go

@@ -6,6 +6,7 @@ package peers
 
 import (
 	context "context"
+	net "net"
 	reflect "reflect"
 
 	gomock "github.com/golang/mock/gomock"
@@ -13,6 +14,7 @@ import (
 	account "github.com/netbirdio/netbird/management/server/account"
 	integrated_validator "github.com/netbirdio/netbird/management/server/integrations/integrated_validator"
 	peer "github.com/netbirdio/netbird/management/server/peer"
+	types "github.com/netbirdio/netbird/management/server/types"
 )
 
 // MockManager is a mock of Manager interface.
@@ -38,6 +40,20 @@ func (m *MockManager) EXPECT() *MockManagerMockRecorder {
 	return m.recorder
 }
 
+// CreateProxyPeer mocks base method.
+func (m *MockManager) CreateProxyPeer(ctx context.Context, accountID, peerKey, cluster string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CreateProxyPeer", ctx, accountID, peerKey, cluster)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// CreateProxyPeer indicates an expected call of CreateProxyPeer.
+func (mr *MockManagerMockRecorder) CreateProxyPeer(ctx, accountID, peerKey, cluster interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateProxyPeer", reflect.TypeOf((*MockManager)(nil).CreateProxyPeer), ctx, accountID, peerKey, cluster)
+}
+
 // DeletePeers mocks base method.
 func (m *MockManager) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 	m.ctrl.T.Helper()
@@ -97,6 +113,21 @@ func (mr *MockManagerMockRecorder) GetPeerAccountID(ctx, peerID interface{}) *go
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPeerAccountID", reflect.TypeOf((*MockManager)(nil).GetPeerAccountID), ctx, peerID)
 }
 
+// GetPeerByTunnelIP mocks base method.
+func (m *MockManager) GetPeerByTunnelIP(ctx context.Context, accountID string, ip net.IP) (*peer.Peer, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetPeerByTunnelIP", ctx, accountID, ip)
+	ret0, _ := ret[0].(*peer.Peer)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetPeerByTunnelIP indicates an expected call of GetPeerByTunnelIP.
+func (mr *MockManagerMockRecorder) GetPeerByTunnelIP(ctx, accountID, ip interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPeerByTunnelIP", reflect.TypeOf((*MockManager)(nil).GetPeerByTunnelIP), ctx, accountID, ip)
+}
+
 // GetPeerID mocks base method.
 func (m *MockManager) GetPeerID(ctx context.Context, peerKey string) (string, error) {
 	m.ctrl.T.Helper()
@@ -112,6 +143,22 @@ func (mr *MockManagerMockRecorder) GetPeerID(ctx, peerKey interface{}) *gomock.C
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPeerID", reflect.TypeOf((*MockManager)(nil).GetPeerID), ctx, peerKey)
 }
 
+// GetPeerWithGroups mocks base method.
+func (m *MockManager) GetPeerWithGroups(ctx context.Context, accountID, peerID string) (*peer.Peer, []*types.Group, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetPeerWithGroups", ctx, accountID, peerID)
+	ret0, _ := ret[0].(*peer.Peer)
+	ret1, _ := ret[1].([]*types.Group)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
+}
+
+// GetPeerWithGroups indicates an expected call of GetPeerWithGroups.
+func (mr *MockManagerMockRecorder) GetPeerWithGroups(ctx, accountID, peerID interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPeerWithGroups", reflect.TypeOf((*MockManager)(nil).GetPeerWithGroups), ctx, accountID, peerID)
+}
+
 // GetPeersByGroupIDs mocks base method.
 func (m *MockManager) GetPeersByGroupIDs(ctx context.Context, accountID string, groupsIDs []string) ([]*peer.Peer, error) {
 	m.ctrl.T.Helper()
@@ -162,17 +209,3 @@ func (mr *MockManagerMockRecorder) SetNetworkMapController(networkMapController
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetNetworkMapController", reflect.TypeOf((*MockManager)(nil).SetNetworkMapController), networkMapController)
 }
-
-// CreateProxyPeer mocks base method.
-func (m *MockManager) CreateProxyPeer(ctx context.Context, accountID string, peerKey string, cluster string) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CreateProxyPeer", ctx, accountID, peerKey, cluster)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// CreateProxyPeer indicates an expected call of CreateProxyPeer.
-func (mr *MockManagerMockRecorder) CreateProxyPeer(ctx, accountID, peerKey, cluster interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateProxyPeer", reflect.TypeOf((*MockManager)(nil).CreateProxyPeer), ctx, accountID, peerKey, cluster)
-}

management/internals/modules/reverseproxy/domain/domain.go

@@ -23,6 +23,8 @@ type Domain struct {
 	// SupportsCrowdSec is populated at query time from proxy cluster capabilities.
 	// Not persisted.
 	SupportsCrowdSec *bool `gorm:"-"`
+	// SupportsPrivate is populated at query time from proxy cluster capabilities. Not persisted.
+	SupportsPrivate *bool `gorm:"-"`
 }
 
 // EventMeta returns activity event metadata for a domain

management/internals/modules/reverseproxy/domain/manager/api.go

@@ -49,6 +49,7 @@ func domainToApi(d *domain.Domain) api.ReverseProxyDomain {
 		SupportsCustomPorts: d.SupportsCustomPorts,
 		RequireSubdomain:    d.RequireSubdomain,
 		SupportsCrowdsec:    d.SupportsCrowdSec,
+		SupportsPrivate:     d.SupportsPrivate,
 	}
 	if d.TargetCluster != "" {
 		resp.TargetCluster = &d.TargetCluster

management/internals/modules/reverseproxy/domain/manager/manager.go

@@ -35,6 +35,7 @@ type proxyManager interface {
 	ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
 	ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
 	ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
+	ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
 }
 
 type Manager struct {
@@ -93,6 +94,7 @@ func (m Manager) GetDomains(ctx context.Context, accountID, userID string) ([]*d
 		d.SupportsCustomPorts = m.proxyManager.ClusterSupportsCustomPorts(ctx, cluster)
 		d.RequireSubdomain = m.proxyManager.ClusterRequireSubdomain(ctx, cluster)
 		d.SupportsCrowdSec = m.proxyManager.ClusterSupportsCrowdSec(ctx, cluster)
+		d.SupportsPrivate = m.proxyManager.ClusterSupportsPrivate(ctx, cluster)
 		ret = append(ret, d)
 	}
 
@@ -109,6 +111,7 @@ func (m Manager) GetDomains(ctx context.Context, accountID, userID string) ([]*d
 		if d.TargetCluster != "" {
 			cd.SupportsCustomPorts = m.proxyManager.ClusterSupportsCustomPorts(ctx, d.TargetCluster)
 			cd.SupportsCrowdSec = m.proxyManager.ClusterSupportsCrowdSec(ctx, d.TargetCluster)
+			cd.SupportsPrivate = m.proxyManager.ClusterSupportsPrivate(ctx, d.TargetCluster)
 		}
 		// Custom domains never require a subdomain by default since
 		// the account owns them and should be able to use the bare domain.
@@ -304,10 +307,27 @@ func (m Manager) getClusterAllowList(ctx context.Context, accountID string) ([]s
 	if err != nil {
 		return nil, fmt.Errorf("get BYOP cluster addresses: %w", err)
 	}
-	if len(byopAddresses) > 0 {
-		return byopAddresses, nil
+	publicAddresses, err := m.proxyManager.GetActiveClusterAddresses(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("get public cluster addresses: %w", err)
 	}
-	return m.proxyManager.GetActiveClusterAddresses(ctx)
+	seen := make(map[string]struct{}, len(byopAddresses)+len(publicAddresses))
+	merged := make([]string, 0, len(byopAddresses)+len(publicAddresses))
+	for _, addr := range byopAddresses {
+		if _, ok := seen[addr]; ok {
+			continue
+		}
+		seen[addr] = struct{}{}
+		merged = append(merged, addr)
+	}
+	for _, addr := range publicAddresses {
+		if _, ok := seen[addr]; ok {
+			continue
+		}
+		seen[addr] = struct{}{}
+		merged = append(merged, addr)
+	}
+	return merged, nil
 }
 
 func extractClusterFromCustomDomains(serviceDomain string, customDomains []*domain.Domain) (string, bool) {

management/internals/modules/reverseproxy/domain/manager/manager_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 type mockProxyManager struct {
-	getActiveClusterAddressesFunc          func(ctx context.Context) ([]string, error)
+	getActiveClusterAddressesFunc           func(ctx context.Context) ([]string, error)
 	getActiveClusterAddressesForAccountFunc func(ctx context.Context, accountID string) ([]string, error)
 }
 
@@ -40,22 +40,41 @@ func (m *mockProxyManager) ClusterSupportsCrowdSec(_ context.Context, _ string)
 	return nil
 }
 
-func TestGetClusterAllowList_BYOPProxy(t *testing.T) {
+func (m *mockProxyManager) ClusterSupportsPrivate(_ context.Context, _ string) *bool {
+	return nil
+}
+
+func TestGetClusterAllowList_BYOPMergedWithPublic(t *testing.T) {
 	pm := &mockProxyManager{
 		getActiveClusterAddressesForAccountFunc: func(_ context.Context, accID string) ([]string, error) {
 			assert.Equal(t, "acc-123", accID)
 			return []string{"byop.example.com"}, nil
 		},
 		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
-			t.Fatal("should not call GetActiveClusterAddresses when BYOP addresses exist")
-			return nil, nil
+			return []string{"eu.proxy.netbird.io"}, nil
 		},
 	}
 
 	mgr := Manager{proxyManager: pm}
 	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
 	require.NoError(t, err)
-	assert.Equal(t, []string{"byop.example.com"}, result)
+	assert.Equal(t, []string{"byop.example.com", "eu.proxy.netbird.io"}, result)
+}
+
+func TestGetClusterAllowList_DeduplicatesBYOPAndPublic(t *testing.T) {
+	pm := &mockProxyManager{
+		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
+			return []string{"shared.example.com", "byop.example.com"}, nil
+		},
+		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
+			return []string{"shared.example.com", "eu.proxy.netbird.io"}, nil
+		},
+	}
+
+	mgr := Manager{proxyManager: pm}
+	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"shared.example.com", "byop.example.com", "eu.proxy.netbird.io"}, result)
 }
 
 func TestGetClusterAllowList_NoBYOP_FallbackToShared(t *testing.T) {
@@ -79,10 +98,6 @@ func TestGetClusterAllowList_BYOPError_ReturnsError(t *testing.T) {
 		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
 			return nil, errors.New("db error")
 		},
-		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
-			t.Fatal("should not call GetActiveClusterAddresses when BYOP lookup fails")
-			return nil, nil
-		},
 	}
 
 	mgr := Manager{proxyManager: pm}
@@ -92,6 +107,23 @@ func TestGetClusterAllowList_BYOPError_ReturnsError(t *testing.T) {
 	assert.Contains(t, err.Error(), "BYOP cluster addresses")
 }
 
+func TestGetClusterAllowList_PublicError_ReturnsError(t *testing.T) {
+	pm := &mockProxyManager{
+		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
+			return []string{"byop.example.com"}, nil
+		},
+		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
+			return nil, errors.New("db error")
+		},
+	}
+
+	mgr := Manager{proxyManager: pm}
+	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
+	require.Error(t, err)
+	assert.Nil(t, result)
+	assert.Contains(t, err.Error(), "public cluster addresses")
+}
+
 func TestGetClusterAllowList_BYOPEmptySlice_FallbackToShared(t *testing.T) {
 	pm := &mockProxyManager{
 		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
@@ -108,3 +140,18 @@ func TestGetClusterAllowList_BYOPEmptySlice_FallbackToShared(t *testing.T) {
 	assert.Equal(t, []string{"eu.proxy.netbird.io"}, result)
 }
 
+func TestGetClusterAllowList_PublicEmpty_BYOPOnly(t *testing.T) {
+	pm := &mockProxyManager{
+		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
+			return []string{"byop.example.com"}, nil
+		},
+		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
+			return nil, nil
+		},
+	}
+
+	mgr := Manager{proxyManager: pm}
+	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"byop.example.com"}, result)
+}

management/internals/modules/reverseproxy/proxy/manager.go

@@ -19,6 +19,7 @@ type Manager interface {
 	ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
 	ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
 	ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
+	ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
 	CleanupStale(ctx context.Context, inactivityDuration time.Duration) error
 	GetAccountProxy(ctx context.Context, accountID string) (*Proxy, error)
 	CountAccountProxies(ctx context.Context, accountID string) (int64, error)

management/internals/modules/reverseproxy/proxy/manager/manager.go

@@ -17,10 +17,11 @@ type store interface {
 	UpdateProxyHeartbeat(ctx context.Context, p *proxy.Proxy) error
 	GetActiveProxyClusterAddresses(ctx context.Context) ([]string, error)
 	GetActiveProxyClusterAddressesForAccount(ctx context.Context, accountID string) ([]string, error)
-	GetActiveProxyClusters(ctx context.Context, accountID string) ([]proxy.Cluster, error)
+	GetProxyClusters(ctx context.Context, accountID string) ([]proxy.Cluster, error)
 	GetClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
 	GetClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
 	GetClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
+	GetClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
 	CleanupStaleProxies(ctx context.Context, inactivityDuration time.Duration) error
 	GetProxyByAccountID(ctx context.Context, accountID string) (*proxy.Proxy, error)
 	CountProxiesByAccountID(ctx context.Context, accountID string) (int64, error)
@@ -137,6 +138,11 @@ func (m Manager) ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string
 	return m.store.GetClusterSupportsCrowdSec(ctx, clusterAddr)
 }
 
+// ClusterSupportsPrivate reports whether any active proxy claims the private capability (nil = unreported).
+func (m Manager) ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool {
+	return m.store.GetClusterSupportsPrivate(ctx, clusterAddr)
+}
+
 // CleanupStale removes proxies that haven't sent heartbeat in the specified duration
 func (m *Manager) CleanupStale(ctx context.Context, inactivityDuration time.Duration) error {
 	if err := m.store.CleanupStaleProxies(ctx, inactivityDuration); err != nil {
@@ -178,4 +184,3 @@ func (m *Manager) DeleteAccountCluster(ctx context.Context, clusterAddress, acco
 	}
 	return nil
 }
-

management/internals/modules/reverseproxy/proxy/manager/manager_test.go

@@ -15,16 +15,16 @@ import (
 )
 
 type mockStore struct {
-	saveProxyFunc                              func(ctx context.Context, p *proxy.Proxy) error
-	disconnectProxyFunc                        func(ctx context.Context, proxyID, sessionID string) error
-	updateProxyHeartbeatFunc                   func(ctx context.Context, p *proxy.Proxy) error
-	getActiveProxyClusterAddressesFunc         func(ctx context.Context) ([]string, error)
-	getActiveProxyClusterAddressesForAccFunc   func(ctx context.Context, accountID string) ([]string, error)
-	cleanupStaleProxiesFunc                    func(ctx context.Context, d time.Duration) error
-	getProxyByAccountIDFunc                    func(ctx context.Context, accountID string) (*proxy.Proxy, error)
-	countProxiesByAccountIDFunc                func(ctx context.Context, accountID string) (int64, error)
-	isClusterAddressConflictingFunc            func(ctx context.Context, clusterAddress, accountID string) (bool, error)
-	deleteAccountClusterFunc                   func(ctx context.Context, clusterAddress, accountID string) error
+	saveProxyFunc                            func(ctx context.Context, p *proxy.Proxy) error
+	disconnectProxyFunc                      func(ctx context.Context, proxyID, sessionID string) error
+	updateProxyHeartbeatFunc                 func(ctx context.Context, p *proxy.Proxy) error
+	getActiveProxyClusterAddressesFunc       func(ctx context.Context) ([]string, error)
+	getActiveProxyClusterAddressesForAccFunc func(ctx context.Context, accountID string) ([]string, error)
+	cleanupStaleProxiesFunc                  func(ctx context.Context, d time.Duration) error
+	getProxyByAccountIDFunc                  func(ctx context.Context, accountID string) (*proxy.Proxy, error)
+	countProxiesByAccountIDFunc              func(ctx context.Context, accountID string) (int64, error)
+	isClusterAddressConflictingFunc          func(ctx context.Context, clusterAddress, accountID string) (bool, error)
+	deleteAccountClusterFunc                 func(ctx context.Context, clusterAddress, accountID string) error
 }
 
 func (m *mockStore) SaveProxy(ctx context.Context, p *proxy.Proxy) error {
@@ -57,7 +57,7 @@ func (m *mockStore) GetActiveProxyClusterAddressesForAccount(ctx context.Context
 	}
 	return nil, nil
 }
-func (m *mockStore) GetActiveProxyClusters(_ context.Context, _ string) ([]proxy.Cluster, error) {
+func (m *mockStore) GetProxyClusters(_ context.Context, _ string) ([]proxy.Cluster, error) {
 	return nil, nil
 }
 func (m *mockStore) CleanupStaleProxies(ctx context.Context, d time.Duration) error {
@@ -99,6 +99,9 @@ func (m *mockStore) GetClusterRequireSubdomain(_ context.Context, _ string) *boo
 func (m *mockStore) GetClusterSupportsCrowdSec(_ context.Context, _ string) *bool {
 	return nil
 }
+func (m *mockStore) GetClusterSupportsPrivate(_ context.Context, _ string) *bool {
+	return nil
+}
 
 func newTestManager(s store) *Manager {
 	meter := noop.NewMeterProvider().Meter("test")

management/internals/modules/reverseproxy/proxy/manager_mock.go

@@ -92,6 +92,20 @@ func (mr *MockManagerMockRecorder) ClusterSupportsCrowdSec(ctx, clusterAddr inte
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterSupportsCrowdSec", reflect.TypeOf((*MockManager)(nil).ClusterSupportsCrowdSec), ctx, clusterAddr)
 }
 
+// ClusterSupportsPrivate mocks base method.
+func (m *MockManager) ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ClusterSupportsPrivate", ctx, clusterAddr)
+	ret0, _ := ret[0].(*bool)
+	return ret0
+}
+
+// ClusterSupportsPrivate indicates an expected call of ClusterSupportsPrivate.
+func (mr *MockManagerMockRecorder) ClusterSupportsPrivate(ctx, clusterAddr interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterSupportsPrivate", reflect.TypeOf((*MockManager)(nil).ClusterSupportsPrivate), ctx, clusterAddr)
+}
+
 // Connect mocks base method.
 func (m *MockManager) Connect(ctx context.Context, proxyID, sessionID, clusterAddress, ipAddress string, accountID *string, capabilities *Capabilities) (*Proxy, error) {
 	m.ctrl.T.Helper()

management/internals/modules/reverseproxy/proxy/proxy.go

@@ -20,6 +20,9 @@ type Capabilities struct {
 	RequireSubdomain *bool
 	// SupportsCrowdsec indicates whether this proxy has CrowdSec configured.
 	SupportsCrowdsec *bool
+	// Private indicates whether this proxy supports inbound access via Wireguard
+	// tunnel and netbird-only authentication policies
+	Private *bool
 }
 
 // Proxy represents a reverse proxy instance
@@ -42,10 +45,34 @@ func (Proxy) TableName() string {
 	return "proxies"
 }
 
+// ClusterType is the source of a proxy cluster.
+type ClusterType string
+
+const (
+	// ClusterTypeAccount is a cluster operated by the account itself (BYOP) —
+	// at least one proxy row in the cluster carries a non-NULL account_id.
+	ClusterTypeAccount ClusterType = "account"
+	// ClusterTypeShared is a cluster operated by NetBird and shared across
+	// accounts — all proxy rows in the cluster have account_id IS NULL.
+	ClusterTypeShared ClusterType = "shared"
+)
+
 // Cluster represents a group of proxy nodes serving the same address.
+//
+// Online and ConnectedProxies derive from the same 2-min active window
+// the rest of the module uses, but Cluster rows are not gated on it —
+// the cluster listing surfaces offline clusters too so operators can
+// see and clean them up. The 1-hour heartbeat reaper still bounds the
+// table eventually.
 type Cluster struct {
 	ID               string
 	Address          string
+	Type             ClusterType
+	Online           bool
 	ConnectedProxies int
-	SelfHosted       bool
+	// *bool: nil = no proxy reported the capability; the dashboard renders that as unknown.
+	SupportsCustomPorts *bool
+	RequireSubdomain    *bool
+	SupportsCrowdSec    *bool
+	Private             *bool
 }

management/internals/modules/reverseproxy/service/interface.go

@@ -9,7 +9,7 @@ import (
 )
 
 type Manager interface {
-	GetActiveClusters(ctx context.Context, accountID, userID string) ([]proxy.Cluster, error)
+	GetClusters(ctx context.Context, accountID, userID string) ([]proxy.Cluster, error)
 	DeleteAccountCluster(ctx context.Context, accountID, userID, clusterAddress string) error
 	GetAllServices(ctx context.Context, accountID, userID string) ([]*Service, error)
 	GetService(ctx context.Context, accountID, userID, serviceID string) (*Service, error)

management/internals/modules/reverseproxy/service/interface_mock.go

@@ -65,20 +65,6 @@ func (mr *MockManagerMockRecorder) CreateServiceFromPeer(ctx, accountID, peerID,
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateServiceFromPeer", reflect.TypeOf((*MockManager)(nil).CreateServiceFromPeer), ctx, accountID, peerID, req)
 }
 
-// DeleteAllServices mocks base method.
-func (m *MockManager) DeleteAllServices(ctx context.Context, accountID, userID string) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "DeleteAllServices", ctx, accountID, userID)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// DeleteAllServices indicates an expected call of DeleteAllServices.
-func (mr *MockManagerMockRecorder) DeleteAllServices(ctx, accountID, userID interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteAllServices", reflect.TypeOf((*MockManager)(nil).DeleteAllServices), ctx, accountID, userID)
-}
-
 // DeleteAccountCluster mocks base method.
 func (m *MockManager) DeleteAccountCluster(ctx context.Context, accountID, userID, clusterAddress string) error {
 	m.ctrl.T.Helper()
@@ -93,6 +79,20 @@ func (mr *MockManagerMockRecorder) DeleteAccountCluster(ctx, accountID, userID,
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteAccountCluster", reflect.TypeOf((*MockManager)(nil).DeleteAccountCluster), ctx, accountID, userID, clusterAddress)
 }
 
+// DeleteAllServices mocks base method.
+func (m *MockManager) DeleteAllServices(ctx context.Context, accountID, userID string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeleteAllServices", ctx, accountID, userID)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// DeleteAllServices indicates an expected call of DeleteAllServices.
+func (mr *MockManagerMockRecorder) DeleteAllServices(ctx, accountID, userID interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteAllServices", reflect.TypeOf((*MockManager)(nil).DeleteAllServices), ctx, accountID, userID)
+}
+
 // DeleteService mocks base method.
 func (m *MockManager) DeleteService(ctx context.Context, accountID, userID, serviceID string) error {
 	m.ctrl.T.Helper()
@@ -122,21 +122,6 @@ func (mr *MockManagerMockRecorder) GetAccountServices(ctx, accountID interface{}
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAccountServices", reflect.TypeOf((*MockManager)(nil).GetAccountServices), ctx, accountID)
 }
 
-// GetActiveClusters mocks base method.
-func (m *MockManager) GetActiveClusters(ctx context.Context, accountID, userID string) ([]proxy.Cluster, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetActiveClusters", ctx, accountID, userID)
-	ret0, _ := ret[0].([]proxy.Cluster)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// GetActiveClusters indicates an expected call of GetActiveClusters.
-func (mr *MockManagerMockRecorder) GetActiveClusters(ctx, accountID, userID interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetActiveClusters", reflect.TypeOf((*MockManager)(nil).GetActiveClusters), ctx, accountID, userID)
-}
-
 // GetAllServices mocks base method.
 func (m *MockManager) GetAllServices(ctx context.Context, accountID, userID string) ([]*Service, error) {
 	m.ctrl.T.Helper()
@@ -152,19 +137,19 @@ func (mr *MockManagerMockRecorder) GetAllServices(ctx, accountID, userID interfa
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllServices", reflect.TypeOf((*MockManager)(nil).GetAllServices), ctx, accountID, userID)
 }
 
-// GetServiceByDomain mocks base method.
-func (m *MockManager) GetServiceByDomain(ctx context.Context, domain string) (*Service, error) {
+// GetClusters mocks base method.
+func (m *MockManager) GetClusters(ctx context.Context, accountID, userID string) ([]proxy.Cluster, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetServiceByDomain", ctx, domain)
-	ret0, _ := ret[0].(*Service)
+	ret := m.ctrl.Call(m, "GetClusters", ctx, accountID, userID)
+	ret0, _ := ret[0].([]proxy.Cluster)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
-// GetServiceByDomain indicates an expected call of GetServiceByDomain.
-func (mr *MockManagerMockRecorder) GetServiceByDomain(ctx, domain interface{}) *gomock.Call {
+// GetClusters indicates an expected call of GetClusters.
+func (mr *MockManagerMockRecorder) GetClusters(ctx, accountID, userID interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetServiceByDomain", reflect.TypeOf((*MockManager)(nil).GetServiceByDomain), ctx, domain)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetClusters", reflect.TypeOf((*MockManager)(nil).GetClusters), ctx, accountID, userID)
 }
 
 // GetGlobalServices mocks base method.
@@ -197,6 +182,21 @@ func (mr *MockManagerMockRecorder) GetService(ctx, accountID, userID, serviceID
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetService", reflect.TypeOf((*MockManager)(nil).GetService), ctx, accountID, userID, serviceID)
 }
 
+// GetServiceByDomain mocks base method.
+func (m *MockManager) GetServiceByDomain(ctx context.Context, domain string) (*Service, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetServiceByDomain", ctx, domain)
+	ret0, _ := ret[0].(*Service)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetServiceByDomain indicates an expected call of GetServiceByDomain.
+func (mr *MockManagerMockRecorder) GetServiceByDomain(ctx, domain interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetServiceByDomain", reflect.TypeOf((*MockManager)(nil).GetServiceByDomain), ctx, domain)
+}
+
 // GetServiceByID mocks base method.
 func (m *MockManager) GetServiceByID(ctx context.Context, accountID, serviceID string) (*Service, error) {
 	m.ctrl.T.Helper()

management/internals/modules/reverseproxy/service/manager/api.go

@@ -187,7 +187,7 @@ func (h *handler) getClusters(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	clusters, err := h.manager.GetActiveClusters(r.Context(), userAuth.AccountId, userAuth.UserId)
+	clusters, err := h.manager.GetClusters(r.Context(), userAuth.AccountId, userAuth.UserId)
 	if err != nil {
 		util.WriteError(r.Context(), err, w)
 		return
@@ -196,10 +196,15 @@ func (h *handler) getClusters(w http.ResponseWriter, r *http.Request) {
 	apiClusters := make([]api.ProxyCluster, 0, len(clusters))
 	for _, c := range clusters {
 		apiClusters = append(apiClusters, api.ProxyCluster{
-			Id:               c.ID,
-			Address:          c.Address,
-			ConnectedProxies: c.ConnectedProxies,
-			SelfHosted:       c.SelfHosted,
+			Id:                  c.ID,
+			Address:             c.Address,
+			Type:                api.ProxyClusterType(c.Type),
+			Online:              c.Online,
+			ConnectedProxies:    c.ConnectedProxies,
+			SupportsCustomPorts: c.SupportsCustomPorts,
+			RequireSubdomain:    c.RequireSubdomain,
+			SupportsCrowdsec:    c.SupportsCrowdSec,
+			Private:             c.Private,
 		})
 	}
 

management/internals/modules/reverseproxy/service/manager/manager.go

@@ -81,6 +81,8 @@ type ClusterDeriver interface {
 type CapabilityProvider interface {
 	ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
 	ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
+	ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
+	ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
 }
 
 type Manager struct {
@@ -112,8 +114,12 @@ func (m *Manager) StartExposeReaper(ctx context.Context) {
 	m.exposeReaper.StartExposeReaper(ctx)
 }
 
-// GetActiveClusters returns all active proxy clusters with their connected proxy count.
-func (m *Manager) GetActiveClusters(ctx context.Context, accountID, userID string) ([]proxy.Cluster, error) {
+// GetClusters returns every proxy cluster visible to the account
+// (shared + its own BYOP), regardless of whether any proxy in the
+// cluster is currently heartbeating. Each cluster is enriched with the
+// capability flags reported by its active proxies so the dashboard can
+// render feature support without a second round-trip.
+func (m *Manager) GetClusters(ctx context.Context, accountID, userID string) ([]proxy.Cluster, error) {
 	ok, err := m.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Services, operations.Read)
 	if err != nil {
 		return nil, status.NewPermissionValidationError(err)
@@ -122,7 +128,19 @@ func (m *Manager) GetActiveClusters(ctx context.Context, accountID, userID strin
 		return nil, status.NewPermissionDeniedError()
 	}
 
-	return m.store.GetActiveProxyClusters(ctx, accountID)
+	clusters, err := m.store.GetProxyClusters(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	for i := range clusters {
+		clusters[i].SupportsCustomPorts = m.capabilities.ClusterSupportsCustomPorts(ctx, clusters[i].Address)
+		clusters[i].RequireSubdomain = m.capabilities.ClusterRequireSubdomain(ctx, clusters[i].Address)
+		clusters[i].SupportsCrowdSec = m.capabilities.ClusterSupportsCrowdSec(ctx, clusters[i].Address)
+		clusters[i].Private = m.capabilities.ClusterSupportsPrivate(ctx, clusters[i].Address)
+	}
+
+	return clusters, nil
 }
 
 // DeleteAccountCluster removes all proxy registrations for the given cluster address
@@ -192,6 +210,9 @@ func (m *Manager) replaceHostByLookup(ctx context.Context, accountID string, s *
 			target.Host = resource.Domain
 		case service.TargetTypeSubnet:
 			// For subnets we do not do any lookups on the resource
+		case service.TargetTypeCluster:
+			// Cluster targets carry the upstream address on target_id; the
+			// proxy resolves the destination at request time.
 		default:
 			return fmt.Errorf("unknown target type: %s", target.TargetType)
 		}
@@ -306,6 +327,10 @@ func (m *Manager) validateSubdomainRequirement(ctx context.Context, domain, clus
 func (m *Manager) persistNewService(ctx context.Context, accountID string, svc *service.Service) error {
 	customPorts := m.clusterCustomPorts(ctx, svc)
 
+	if err := validateTargetReferences(ctx, m.store, accountID, svc.Targets); err != nil {
+		return err
+	}
+
 	return m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
 		if svc.Domain != "" {
 			if err := m.checkDomainAvailable(ctx, transaction, svc.Domain, ""); err != nil {
@@ -321,10 +346,6 @@ func (m *Manager) persistNewService(ctx context.Context, accountID string, svc *
 			return err
 		}
 
-		if err := validateTargetReferences(ctx, transaction, accountID, svc.Targets); err != nil {
-			return err
-		}
-
 		if err := transaction.CreateService(ctx, svc); err != nil {
 			return fmt.Errorf("create service: %w", err)
 		}
@@ -435,6 +456,10 @@ func (m *Manager) assignPort(ctx context.Context, tx store.Store, cluster string
 func (m *Manager) persistNewEphemeralService(ctx context.Context, accountID, peerID string, svc *service.Service) error {
 	customPorts := m.clusterCustomPorts(ctx, svc)
 
+	if err := validateTargetReferences(ctx, m.store, accountID, svc.Targets); err != nil {
+		return err
+	}
+
 	return m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
 		if err := m.validateEphemeralPreconditions(ctx, transaction, accountID, peerID, svc); err != nil {
 			return err
@@ -448,10 +473,6 @@ func (m *Manager) persistNewEphemeralService(ctx context.Context, accountID, pee
 			return err
 		}
 
-		if err := validateTargetReferences(ctx, transaction, accountID, svc.Targets); err != nil {
-			return err
-		}
-
 		if err := transaction.CreateService(ctx, svc); err != nil {
 			return fmt.Errorf("create service: %w", err)
 		}
@@ -552,10 +573,22 @@ func (m *Manager) persistServiceUpdate(ctx context.Context, accountID string, se
 	svcForCaps.ProxyCluster = effectiveCluster
 	customPorts := m.clusterCustomPorts(ctx, &svcForCaps)
 
+	if err := validateTargetReferences(ctx, m.store, accountID, service.Targets); err != nil {
+		return nil, err
+	}
+
+	// Validate subdomain requirement *before* the transaction: the underlying
+	// capability lookup talks to the main DB pool, and SQLite's single-connection
+	// pool would self-deadlock if this ran while the tx already held the only
+	// connection.
+	if err := m.validateSubdomainRequirement(ctx, service.Domain, effectiveCluster); err != nil {
+		return nil, err
+	}
+
 	var updateInfo serviceUpdateInfo
 
 	err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
-		return m.executeServiceUpdate(ctx, transaction, accountID, service, &updateInfo, customPorts)
+		return m.executeServiceUpdate(ctx, transaction, accountID, service, &updateInfo, customPorts, effectiveCluster)
 	})
 
 	return &updateInfo, err
@@ -585,7 +618,7 @@ func (m *Manager) resolveEffectiveCluster(ctx context.Context, accountID string,
 	return existing.ProxyCluster, nil
 }
 
-func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.Store, accountID string, service *service.Service, updateInfo *serviceUpdateInfo, customPorts *bool) error {
+func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.Store, accountID string, service *service.Service, updateInfo *serviceUpdateInfo, customPorts *bool, effectiveCluster string) error {
 	existingService, err := transaction.GetServiceByID(ctx, store.LockingStrengthUpdate, accountID, service.ID)
 	if err != nil {
 		return err
@@ -603,17 +636,13 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St
 	updateInfo.domainChanged = existingService.Domain != service.Domain
 
 	if updateInfo.domainChanged {
-		if err := m.handleDomainChange(ctx, transaction, accountID, service); err != nil {
+		if err := m.handleDomainChange(ctx, transaction, service, effectiveCluster); err != nil {
 			return err
 		}
 	} else {
 		service.ProxyCluster = existingService.ProxyCluster
 	}
 
-	if err := m.validateSubdomainRequirement(ctx, service.Domain, service.ProxyCluster); err != nil {
-		return err
-	}
-
 	m.preserveExistingAuthSecrets(service, existingService)
 	if err := validateHeaderAuthValues(service.Auth.HeaderAuths); err != nil {
 		return err
@@ -628,9 +657,6 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St
 	if err := m.checkPortConflict(ctx, transaction, service); err != nil {
 		return err
 	}
-	if err := validateTargetReferences(ctx, transaction, accountID, service.Targets); err != nil {
-		return err
-	}
 	if err := transaction.UpdateService(ctx, service); err != nil {
 		return fmt.Errorf("update service: %w", err)
 	}
@@ -638,20 +664,18 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St
 	return nil
 }
 
-func (m *Manager) handleDomainChange(ctx context.Context, transaction store.Store, accountID string, svc *service.Service) error {
+// handleDomainChange validates the new domain is free inside the transaction
+// and applies the pre-resolved cluster (computed outside the tx by
+// resolveEffectiveCluster). It must NOT call clusterDeriver here: that talks
+// to the main DB pool and would self-deadlock under SQLite (max_open_conns=1)
+// because the transaction already holds the only connection.
+func (m *Manager) handleDomainChange(ctx context.Context, transaction store.Store, svc *service.Service, effectiveCluster string) error {
 	if err := m.checkDomainAvailable(ctx, transaction, svc.Domain, svc.ID); err != nil {
 		return err
 	}
-
-	if m.clusterDeriver != nil {
-		newCluster, err := m.clusterDeriver.DeriveClusterFromDomain(ctx, accountID, svc.Domain)
-		if err != nil {
-			log.WithError(err).Warnf("could not derive cluster from domain %s", svc.Domain)
-		} else {
-			svc.ProxyCluster = newCluster
-		}
+	if effectiveCluster != "" {
+		svc.ProxyCluster = effectiveCluster
 	}
-
 	return nil
 }
 
@@ -760,6 +784,10 @@ func validateTargetReferences(ctx context.Context, transaction store.Store, acco
 			if err := validateResourceTarget(ctx, transaction, accountID, target); err != nil {
 				return err
 			}
+		case service.TargetTypeCluster:
+			if err := validateClusterTarget(target); err != nil {
+				return err
+			}
 		default:
 			return status.Errorf(status.InvalidArgument, "unknown target type %q for target %q", target.TargetType, target.TargetId)
 		}
@@ -767,6 +795,13 @@ func validateTargetReferences(ctx context.Context, transaction store.Store, acco
 	return nil
 }
 
+func validateClusterTarget(target *service.Target) error {
+	if !target.Options.DirectUpstream {
+		return status.Errorf(status.InvalidArgument, "cluster target %s has direct upstream disabled", target.Host)
+	}
+	return nil
+}
+
 func validatePeerTarget(ctx context.Context, transaction store.Store, accountID string, target *service.Target) error {
 	if _, err := transaction.GetPeerByID(ctx, store.LockingStrengthShare, accountID, target.TargetId); err != nil {
 		if sErr, ok := status.FromError(err); ok && sErr.Type() == status.NotFound {
@@ -943,12 +978,14 @@ func (m *Manager) ReloadAllServicesForAccount(ctx context.Context, accountID str
 		return fmt.Errorf("failed to get services: %w", err)
 	}
 
+	oidcCfg := m.proxyController.GetOIDCValidationConfig()
+
 	for _, s := range services {
 		err = m.replaceHostByLookup(ctx, accountID, s)
 		if err != nil {
 			return fmt.Errorf("failed to replace host by lookup for service %s: %w", s.ID, err)
 		}
-		m.proxyController.SendServiceUpdateToCluster(ctx, accountID, s.ToProtoMapping(service.Update, "", m.proxyController.GetOIDCValidationConfig()), s.ProxyCluster)
+		m.proxyController.SendServiceUpdateToCluster(ctx, accountID, s.ToProtoMapping(service.Update, "", oidcCfg), s.ProxyCluster)
 	}
 
 	return nil

management/internals/modules/reverseproxy/service/manager/manager_test.go

@@ -1344,3 +1344,66 @@ func TestValidateSubdomainRequirement(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateTargetReferences_ClusterTargetSkipsLookup(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	mockStore := store.NewMockStore(ctrl)
+	accountID := "test-account"
+
+	// No peer or resource lookups must be issued for cluster targets.
+	targets := []*rpservice.Target{
+		{
+			TargetId:   "eu.proxy.netbird.io",
+			TargetType: rpservice.TargetTypeCluster,
+			Options:    rpservice.TargetOptions{DirectUpstream: true},
+		},
+	}
+	require.NoError(t, validateTargetReferences(ctx, mockStore, accountID, targets), "cluster target must validate without store lookups")
+}
+
+// TestValidateTargetReferences_ClusterTargetRequiresDirectUpstream pins the
+// store-side check that cluster targets must opt into the host-stack dial
+// path. Without DirectUpstream the proxy would route this target through
+// the embedded NetBird client and fail on every request.
+func TestValidateTargetReferences_ClusterTargetRequiresDirectUpstream(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	mockStore := store.NewMockStore(ctrl)
+	accountID := "test-account"
+
+	targets := []*rpservice.Target{
+		{
+			TargetId:   "eu.proxy.netbird.io",
+			TargetType: rpservice.TargetTypeCluster,
+			Host:       "backend.lan",
+		},
+	}
+	err := validateTargetReferences(ctx, mockStore, accountID, targets)
+	require.Error(t, err, "cluster target without direct_upstream must be rejected")
+	assert.ErrorContains(t, err, "direct upstream disabled")
+}
+
+func TestReplaceHostByLookup_SkipsClusterTarget(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	mockStore := store.NewMockStore(ctrl)
+	accountID := "test-account"
+
+	mgr := &Manager{store: mockStore}
+
+	svc := &rpservice.Service{
+		ID:        "svc-1",
+		AccountID: accountID,
+		Targets: []*rpservice.Target{
+			{
+				TargetId:   "eu.proxy.netbird.io",
+				TargetType: rpservice.TargetTypeCluster,
+				Host:       "127.0.0.1",
+			},
+		},
+	}
+
+	require.NoError(t, mgr.replaceHostByLookup(ctx, accountID, svc), "cluster target must not trigger peer/resource lookup")
+	assert.Equal(t, "127.0.0.1", svc.Targets[0].Host, "operator-supplied host must be preserved for cluster target")
+}

management/internals/modules/reverseproxy/service/service.go

@@ -45,10 +45,11 @@ const (
 	StatusCertificateFailed  Status = "certificate_failed"
 	StatusError              Status = "error"
 
-	TargetTypePeer   TargetType = "peer"
-	TargetTypeHost   TargetType = "host"
-	TargetTypeDomain TargetType = "domain"
-	TargetTypeSubnet TargetType = "subnet"
+	TargetTypePeer    TargetType = "peer"
+	TargetTypeHost    TargetType = "host"
+	TargetTypeDomain  TargetType = "domain"
+	TargetTypeSubnet  TargetType = "subnet"
+	TargetTypeCluster TargetType = "cluster"
 
 	SourcePermanent = "permanent"
 	SourceEphemeral = "ephemeral"
@@ -60,6 +61,11 @@ type TargetOptions struct {
 	SessionIdleTimeout time.Duration     `json:"session_idle_timeout,omitempty"`
 	PathRewrite        PathRewriteMode   `json:"path_rewrite,omitempty"`
 	CustomHeaders      map[string]string `gorm:"serializer:json" json:"custom_headers,omitempty"`
+	// DirectUpstream bypasses the proxy's embedded NetBird client and dials
+	// the target via the proxy host's network stack. Useful for upstreams
+	// reachable without WireGuard (public APIs, LAN services, localhost
+	// sidecars). Default false.
+	DirectUpstream bool `json:"direct_upstream,omitempty"`
 }
 
 type Target struct {
@@ -67,7 +73,7 @@ type Target struct {
 	AccountID     string        `gorm:"index:idx_target_account;not null" json:"-"`
 	ServiceID     string        `gorm:"index:idx_service_targets;not null" json:"-"`
 	Path          *string       `json:"path,omitempty"`
-	Host          string        `json:"host"` // the Host field is only used for subnet targets, otherwise ignored
+	Host          string        `json:"host"`
 	Port          uint16        `gorm:"index:idx_target_port" json:"port"`
 	Protocol      string        `gorm:"index:idx_target_protocol" json:"protocol"`
 	TargetId      string        `gorm:"index:idx_target_id" json:"target_id"`
@@ -200,6 +206,10 @@ type Service struct {
 	Mode             string `gorm:"default:'http'"`
 	ListenPort       uint16
 	PortAutoAssigned bool
+	// Private marks the service as NetBird-only: auth via ValidateTunnelPeer against AccessGroups instead of SSO. HTTP-only.
+	Private bool
+	// AccessGroups is the group ID allowlist for inbound peers on private services. Mutually exclusive with bearer SSO.
+	AccessGroups []string `json:"access_groups,omitempty" gorm:"serializer:json"`
 }
 
 // InitNewRecord generates a new unique ID and resets metadata for a newly created
@@ -299,6 +309,12 @@ func (s *Service) ToAPIResponse() *api.Service {
 		Mode:               &mode,
 		ListenPort:         &listenPort,
 		PortAutoAssigned:   &s.PortAutoAssigned,
+		Private:            &s.Private,
+	}
+
+	if len(s.AccessGroups) > 0 {
+		groups := append([]string(nil), s.AccessGroups...)
+		resp.AccessGroups = &groups
 	}
 
 	if s.ProxyCluster != "" {
@@ -308,6 +324,7 @@ func (s *Service) ToAPIResponse() *api.Service {
 	return resp
 }
 
+// ToProtoMapping converts the service into the wire format the proxy consumes.
 func (s *Service) ToProtoMapping(operation Operation, authToken string, oidcConfig proxy.OIDCValidationConfig) *proto.ProxyMapping {
 	pathMappings := s.buildPathMappings()
 
@@ -349,6 +366,7 @@ func (s *Service) ToProtoMapping(operation Operation, authToken string, oidcConf
 		RewriteRedirects: s.RewriteRedirects,
 		Mode:             s.Mode,
 		ListenPort:       int32(s.ListenPort), //nolint:gosec
+		Private:          s.Private,
 	}
 
 	if r := restrictionsToProto(s.Restrictions); r != nil {
@@ -381,13 +399,14 @@ func (s *Service) buildPathMappings() []*proto.PathMapping {
 		}
 
 		// HTTP/HTTPS: build full URL
+		hostNoBrackets := strings.TrimSuffix(strings.TrimPrefix(target.Host, "["), "]")
 		targetURL := url.URL{
 			Scheme: target.Protocol,
-			Host:   target.Host,
+			Host:   bracketIPv6Host(hostNoBrackets),
 			Path:   "/",
 		}
 		if target.Port > 0 && !isDefaultPort(target.Protocol, target.Port) {
-			targetURL.Host = net.JoinHostPort(targetURL.Host, strconv.FormatUint(uint64(target.Port), 10))
+			targetURL.Host = net.JoinHostPort(hostNoBrackets, strconv.FormatUint(uint64(target.Port), 10))
 		}
 
 		path := "/"
@@ -405,6 +424,19 @@ func (s *Service) buildPathMappings() []*proto.PathMapping {
 	return pathMappings
 }
 
+// bracketIPv6Host wraps host in square brackets when it is an IPv6 literal, as
+// required for the Host field of net/url.URL (RFC 3986 §3.2.2). v4-mapped IPv6
+// addresses are bracketed too since their textual form contains colons.
+func bracketIPv6Host(host string) string {
+	if strings.HasPrefix(host, "[") {
+		return host
+	}
+	if addr, err := netip.ParseAddr(host); err == nil && addr.Is6() {
+		return "[" + host + "]"
+	}
+	return host
+}
+
 func operationToProtoType(op Operation) proto.ProxyMappingUpdateType {
 	switch op {
 	case Create:
@@ -441,7 +473,8 @@ func pathRewriteToProto(mode PathRewriteMode) proto.PathRewriteMode {
 }
 
 func targetOptionsToAPI(opts TargetOptions) *api.ServiceTargetOptions {
-	if !opts.SkipTLSVerify && opts.RequestTimeout == 0 && opts.SessionIdleTimeout == 0 && opts.PathRewrite == "" && len(opts.CustomHeaders) == 0 {
+	if !opts.SkipTLSVerify && opts.RequestTimeout == 0 && opts.SessionIdleTimeout == 0 &&
+		opts.PathRewrite == "" && len(opts.CustomHeaders) == 0 && !opts.DirectUpstream {
 		return nil
 	}
 	apiOpts := &api.ServiceTargetOptions{}
@@ -463,17 +496,22 @@ func targetOptionsToAPI(opts TargetOptions) *api.ServiceTargetOptions {
 	if len(opts.CustomHeaders) > 0 {
 		apiOpts.CustomHeaders = &opts.CustomHeaders
 	}
+	if opts.DirectUpstream {
+		apiOpts.DirectUpstream = &opts.DirectUpstream
+	}
 	return apiOpts
 }
 
 func targetOptionsToProto(opts TargetOptions) *proto.PathTargetOptions {
-	if !opts.SkipTLSVerify && opts.PathRewrite == "" && opts.RequestTimeout == 0 && len(opts.CustomHeaders) == 0 {
+	if !opts.SkipTLSVerify && opts.PathRewrite == "" && opts.RequestTimeout == 0 &&
+		len(opts.CustomHeaders) == 0 && !opts.DirectUpstream {
 		return nil
 	}
 	popts := &proto.PathTargetOptions{
-		SkipTlsVerify: opts.SkipTLSVerify,
-		PathRewrite:   pathRewriteToProto(opts.PathRewrite),
-		CustomHeaders: opts.CustomHeaders,
+		SkipTlsVerify:  opts.SkipTLSVerify,
+		PathRewrite:    pathRewriteToProto(opts.PathRewrite),
+		CustomHeaders:  opts.CustomHeaders,
+		DirectUpstream: opts.DirectUpstream,
 	}
 	if opts.RequestTimeout != 0 {
 		popts.RequestTimeout = durationpb.New(opts.RequestTimeout)
@@ -523,6 +561,9 @@ func targetOptionsFromAPI(idx int, o *api.ServiceTargetOptions) (TargetOptions,
 	if o.CustomHeaders != nil {
 		opts.CustomHeaders = *o.CustomHeaders
 	}
+	if o.DirectUpstream != nil {
+		opts.DirectUpstream = *o.DirectUpstream
+	}
 	return opts, nil
 }
 
@@ -537,6 +578,14 @@ func (s *Service) FromAPIRequest(req *api.ServiceRequest, accountID string) erro
 	if req.ListenPort != nil {
 		s.ListenPort = uint16(*req.ListenPort) //nolint:gosec
 	}
+	if req.Private != nil {
+		s.Private = *req.Private
+	}
+	if req.AccessGroups != nil {
+		s.AccessGroups = append([]string(nil), *req.AccessGroups...)
+	} else {
+		s.AccessGroups = nil
+	}
 
 	targets, err := targetsFromAPI(accountID, req.Targets)
 	if err != nil {
@@ -726,6 +775,9 @@ func (s *Service) Validate() error {
 	if err := validateAccessRestrictions(&s.Restrictions); err != nil {
 		return err
 	}
+	if err := s.validatePrivateRequirements(); err != nil {
+		return err
+	}
 
 	switch s.Mode {
 	case ModeHTTP:
@@ -739,6 +791,23 @@ func (s *Service) Validate() error {
 	}
 }
 
+// validatePrivateRequirements enforces the private-service contract: HTTP mode, ≥1 access group, no bearer auth.
+func (s *Service) validatePrivateRequirements() error {
+	if !s.Private {
+		return nil
+	}
+	if s.Mode != "" && s.Mode != ModeHTTP {
+		return fmt.Errorf("private services only support HTTP mode, got %q", s.Mode)
+	}
+	if len(s.AccessGroups) == 0 {
+		return errors.New("private services require at least one access group")
+	}
+	if s.Auth.BearerAuth != nil && s.Auth.BearerAuth.Enabled {
+		return errors.New("private services cannot enable bearer auth (SSO): NetBird-only access and SSO are mutually exclusive")
+	}
+	return nil
+}
+
 func (s *Service) validateHTTPMode() error {
 	if s.Domain == "" {
 		return errors.New("service domain is required")
@@ -785,11 +854,21 @@ func (s *Service) validateHTTPTargets() error {
 	for i, target := range s.Targets {
 		switch target.TargetType {
 		case TargetTypePeer, TargetTypeHost, TargetTypeDomain:
-			// host field will be ignored
+			// Host is normally overwritten by replaceHostByLookup with the
+			// resolved peer IP / resource address; operator-supplied values
+			// are honored only when DirectUpstream is set. Validate the
+			// override here so misconfigured hosts fail fast at API time.
+			if err := validateDirectUpstreamHost(i, target); err != nil {
+				return err
+			}
 		case TargetTypeSubnet:
 			if target.Host == "" {
 				return fmt.Errorf("target %d has empty host but target_type is %q", i, target.TargetType)
 			}
+		case TargetTypeCluster:
+			if err := validateClusterTarget(i, target); err != nil {
+				return err
+			}
 		default:
 			return fmt.Errorf("target %d has invalid target_type %q", i, target.TargetType)
 		}
@@ -807,25 +886,67 @@ func (s *Service) validateHTTPTargets() error {
 	return nil
 }
 
+// validateClusterTarget cluster targets should not have empty hosts and should have direct upstream enabled.
+func validateClusterTarget(idx int, target *Target) error {
+	host := strings.TrimSpace(target.Host)
+	if host == "" {
+		return fmt.Errorf("target %d: has empty host", idx)
+	}
+	if !target.Options.DirectUpstream {
+		return fmt.Errorf("target %d: %s has direct upstream disabled", idx, target.Host)
+	}
+	return validateDirectUpstreamHost(idx, target)
+}
+
+// validateDirectUpstreamHost validates the operator-supplied Host on a
+// peer/host/domain target when DirectUpstream is set. Empty Host is
+// allowed — the lookup fills in the default peer IP / resource address.
+// Without DirectUpstream the Host value is silently overwritten by
+// replaceHostByLookup, so we don't validate it (preserves the historical
+// behaviour where APIs accepted any value and dropped it). Non-empty
+// Host with DirectUpstream must look like a hostname or IP and must
+// not carry a port (port lives on Target.Port).
+func validateDirectUpstreamHost(idx int, target *Target) error {
+	if !target.Options.DirectUpstream {
+		return nil
+	}
+	host := strings.TrimSpace(target.Host)
+	if host == "" {
+		return nil
+	}
+	if strings.ContainsAny(host, " \t/") {
+		return fmt.Errorf("target %d: host %q contains invalid characters", idx, host)
+	}
+	if _, _, err := net.SplitHostPort(host); err == nil {
+		return fmt.Errorf("target %d: host %q must not include a port (set target.port instead)", idx, host)
+	}
+	return nil
+}
+
 func (s *Service) validateL4Target(target *Target) error {
 	// L4 services have a single target; per-target disable is meaningless
 	// (use the service-level Enabled flag instead). Force it on so that
 	// buildPathMappings always includes the target in the proto.
 	target.Enabled = true
 
-	if target.Port == 0 {
-		return errors.New("target port is required for L4 services")
-	}
 	if target.TargetId == "" {
 		return errors.New("target_id is required for L4 services")
 	}
+	if target.TargetType != TargetTypeCluster && target.Port == 0 {
+		return errors.New("target port is required for L4 services")
+	}
 	switch target.TargetType {
 	case TargetTypePeer, TargetTypeHost, TargetTypeDomain:
-		// OK
+		if err := validateDirectUpstreamHost(0, target); err != nil {
+			return err
+		}
 	case TargetTypeSubnet:
 		if target.Host == "" {
 			return errors.New("target host is required for subnet targets")
 		}
+	case TargetTypeCluster:
+		// target_id carries the cluster address; the proxy resolves
+		// the upstream at request time.
 	default:
 		return fmt.Errorf("invalid target_type %q for L4 service", target.TargetType)
 	}
@@ -1160,6 +1281,11 @@ func (s *Service) Copy() *Service {
 		}
 	}
 
+	var accessGroups []string
+	if len(s.AccessGroups) > 0 {
+		accessGroups = append([]string(nil), s.AccessGroups...)
+	}
+
 	return &Service{
 		ID:                s.ID,
 		AccountID:         s.AccountID,
@@ -1181,6 +1307,8 @@ func (s *Service) Copy() *Service {
 		Mode:              s.Mode,
 		ListenPort:        s.ListenPort,
 		PortAutoAssigned:  s.PortAutoAssigned,
+		Private:           s.Private,
+		AccessGroups:      accessGroups,
 	}
 }
 

management/internals/modules/reverseproxy/service/service_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
 	"github.com/netbirdio/netbird/shared/hash/argon2id"
+	"github.com/netbirdio/netbird/shared/management/http/api"
 	"github.com/netbirdio/netbird/shared/management/proto"
 )
 
@@ -351,6 +352,83 @@ func TestToProtoMapping_PortInTargetURL(t *testing.T) {
 			port:       80,
 			wantTarget: "https://10.0.0.1:80/",
 		},
+		{
+			name:       "domain host without port is unchanged",
+			protocol:   "http",
+			host:       "example.com",
+			port:       0,
+			wantTarget: "http://example.com/",
+		},
+		{
+			name:       "domain host with non-default port is unchanged",
+			protocol:   "http",
+			host:       "example.com",
+			port:       8080,
+			wantTarget: "http://example.com:8080/",
+		},
+		{
+			name:       "ipv6 host without port is bracketed",
+			protocol:   "http",
+			host:       "fb00:cafe:1::3",
+			port:       0,
+			wantTarget: "http://[fb00:cafe:1::3]/",
+		},
+		{
+			name:       "ipv6 host with default port omits port and brackets host",
+			protocol:   "http",
+			host:       "fb00:cafe:1::3",
+			port:       80,
+			wantTarget: "http://[fb00:cafe:1::3]/",
+		},
+		{
+			name:       "ipv6 host with non-default port is bracketed",
+			protocol:   "http",
+			host:       "fb00:cafe:1::3",
+			port:       8080,
+			wantTarget: "http://[fb00:cafe:1::3]:8080/",
+		},
+		{
+			name:       "ipv6 loopback without port is bracketed",
+			protocol:   "http",
+			host:       "::1",
+			port:       0,
+			wantTarget: "http://[::1]/",
+		},
+		{
+			name:       "ipv6 host with 5-digit port is bracketed",
+			protocol:   "http",
+			host:       "fb00:cafe::1",
+			port:       18080,
+			wantTarget: "http://[fb00:cafe::1]:18080/",
+		},
+		{
+			name:       "pre-bracketed ipv6 without port stays single-bracketed",
+			protocol:   "http",
+			host:       "[fb00:cafe::1]",
+			port:       0,
+			wantTarget: "http://[fb00:cafe::1]/",
+		},
+		{
+			name:       "pre-bracketed ipv6 with port is not double-bracketed",
+			protocol:   "http",
+			host:       "[fb00:cafe::1]",
+			port:       8080,
+			wantTarget: "http://[fb00:cafe::1]:8080/",
+		},
+		{
+			name:       "v4-mapped ipv6 host without port is bracketed",
+			protocol:   "http",
+			host:       "::ffff:10.0.0.1",
+			port:       0,
+			wantTarget: "http://[::ffff:10.0.0.1]/",
+		},
+		{
+			name:       "full-form 8-group ipv6 without port is bracketed",
+			protocol:   "http",
+			host:       "fb00:cafe:1:0:0:0:0:3",
+			port:       0,
+			wantTarget: "http://[fb00:cafe:1:0:0:0:0:3]/",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -1039,3 +1117,191 @@ func TestValidate_HeaderAuths(t *testing.T) {
 		assert.Contains(t, err.Error(), "exceeds maximum length")
 	})
 }
+
+func TestValidate_HTTPClusterTarget(t *testing.T) {
+	rp := validProxy()
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "http",
+		Host:       "backend.lan",
+		Options:    TargetOptions{DirectUpstream: true},
+		Enabled:    true,
+	}}
+	require.NoError(t, rp.Validate(), "HTTP cluster target with target_id, host, and direct_upstream must validate")
+}
+
+func TestValidate_HTTPClusterTarget_RequiresTargetId(t *testing.T) {
+	rp := validProxy()
+	rp.Targets = []*Target{{
+		TargetType: TargetTypeCluster,
+		Protocol:   "http",
+		Host:       "backend.lan",
+		Options:    TargetOptions{DirectUpstream: true},
+		Enabled:    true,
+	}}
+	assert.ErrorContains(t, rp.Validate(), "empty target_id", "cluster target must reject empty target_id")
+}
+
+// TestValidate_HTTPClusterTarget_RequiresHost pins the new cluster-target
+// rule that operator-supplied Host is mandatory: cluster targets dial the
+// upstream via the host network stack (direct_upstream is implied), so an
+// empty Host leaves the proxy with nothing to dial.
+func TestValidate_HTTPClusterTarget_RequiresHost(t *testing.T) {
+	rp := validProxy()
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "http",
+		Options:    TargetOptions{DirectUpstream: true},
+		Enabled:    true,
+	}}
+	assert.ErrorContains(t, rp.Validate(), "empty host", "cluster target must reject empty host")
+}
+
+// TestValidate_HTTPClusterTarget_RequiresDirectUpstream pins the second
+// half of the cluster-target rule: DirectUpstream must be true so the
+// stdlib transport branch in MultiTransport is taken. Without it the
+// embedded NetBird client would try to dial the cluster address through
+// the WG tunnel, which is the wrong network for a cluster upstream.
+func TestValidate_HTTPClusterTarget_RequiresDirectUpstream(t *testing.T) {
+	rp := validProxy()
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "http",
+		Host:       "backend.lan",
+		Enabled:    true,
+	}}
+	assert.ErrorContains(t, rp.Validate(), "direct upstream disabled", "cluster target must reject direct_upstream=false")
+}
+
+func TestValidate_L4ClusterTarget(t *testing.T) {
+	rp := validProxy()
+	rp.Mode = ModeTCP
+	rp.ListenPort = 9000
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "tcp",
+		Enabled:    true,
+	}}
+	require.NoError(t, rp.Validate(), "L4 cluster target must validate without an explicit port")
+}
+
+func TestService_Copy_RoundtripsPrivate(t *testing.T) {
+	svc := validProxy()
+	svc.Private = true
+	svc.AccessGroups = []string{"grp-admins", "grp-ops"}
+	cp := svc.Copy()
+	require.NotNil(t, cp)
+	assert.True(t, cp.Private)
+	assert.Equal(t, []string{"grp-admins", "grp-ops"}, cp.AccessGroups)
+
+	cp.Private = false
+	assert.True(t, svc.Private)
+
+	cp.AccessGroups[0] = "grp-other"
+	assert.Equal(t, []string{"grp-admins", "grp-ops"}, svc.AccessGroups)
+}
+
+func TestService_APIRoundtrip_Private(t *testing.T) {
+	enabled := true
+	private := true
+	accessGroups := []string{"grp-admins"}
+	targets := []api.ServiceTarget{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: api.ServiceTargetTargetType("cluster"),
+		Protocol:   "http",
+		Port:       80,
+		Enabled:    true,
+	}}
+	req := &api.ServiceRequest{
+		Name:         "svc-private",
+		Domain:       "myapp.eu.proxy.netbird.io",
+		Enabled:      enabled,
+		Private:      &private,
+		AccessGroups: &accessGroups,
+		Targets:      &targets,
+	}
+
+	svc := &Service{}
+	require.NoError(t, svc.FromAPIRequest(req, "acc-1"))
+	assert.True(t, svc.Private)
+	assert.Equal(t, []string{"grp-admins"}, svc.AccessGroups)
+
+	resp := svc.ToAPIResponse()
+	require.NotNil(t, resp.Private)
+	assert.True(t, *resp.Private)
+	require.NotNil(t, resp.AccessGroups)
+	assert.Equal(t, []string{"grp-admins"}, *resp.AccessGroups)
+}
+
+func TestValidate_Private_RequiresAccessGroups(t *testing.T) {
+	rp := validProxy()
+	rp.Private = true
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "http",
+		Host:       "backend.lan",
+		Options:    TargetOptions{DirectUpstream: true},
+		Enabled:    true,
+	}}
+	assert.ErrorContains(t, rp.Validate(), "access group")
+}
+
+func TestValidate_Private_RejectsBearerAuth(t *testing.T) {
+	rp := validProxy()
+	rp.Private = true
+	rp.AccessGroups = []string{"grp-admins"}
+	rp.Auth.BearerAuth = &BearerAuthConfig{
+		Enabled:            true,
+		DistributionGroups: []string{"grp-sso"},
+	}
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "http",
+		Host:       "backend.lan",
+		Options:    TargetOptions{DirectUpstream: true},
+		Enabled:    true,
+	}}
+	assert.ErrorContains(t, rp.Validate(), "mutually exclusive")
+}
+
+func TestValidate_Private_AcceptsNonClusterTargets(t *testing.T) {
+	rp := validProxy()
+	rp.Private = true
+	rp.AccessGroups = []string{"grp-admins"}
+	require.NoError(t, rp.Validate())
+}
+
+func TestValidate_Private_AcceptsClusterTargetWithAccessGroups(t *testing.T) {
+	rp := validProxy()
+	rp.Private = true
+	rp.AccessGroups = []string{"grp-admins"}
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "http",
+		Host:       "backend.lan",
+		Options:    TargetOptions{DirectUpstream: true},
+		Enabled:    true,
+	}}
+	require.NoError(t, rp.Validate())
+}
+
+func TestValidate_Private_RejectsNonHTTPMode(t *testing.T) {
+	rp := validProxy()
+	rp.Private = true
+	rp.AccessGroups = []string{"grp-admins"}
+	rp.Mode = ModeTCP
+	rp.Targets = []*Target{{
+		TargetId:   "eu.proxy.netbird.io",
+		TargetType: TargetTypeCluster,
+		Protocol:   "tcp",
+		Enabled:    true,
+	}}
+	assert.ErrorContains(t, rp.Validate(), "HTTP")
+}

management/internals/modules/reverseproxy/sessionkey/sessionkey.go

@@ -20,6 +20,20 @@ type KeyPair struct {
 type Claims struct {
 	jwt.RegisteredClaims
 	Method auth.Method `json:"method"`
+	// Email is the calling user's email address. Carried so the
+	// proxy can stamp identity on upstream requests (e.g.
+	// x-litellm-end-user-id) without an extra management
+	// round-trip on every cookie-bearing request.
+	Email string `json:"email,omitempty"`
+	// Groups carries the user's group IDs so the proxy can stamp them
+	// onto upstream requests (X-NetBird-Groups) from the cookie path
+	// without an extra management round-trip.
+	Groups []string `json:"groups,omitempty"`
+	// GroupNames carries the human-readable display names for the ids
+	// in Groups, ordered identically (positional pairing). Slice may be
+	// shorter than Groups for tokens minted before names were
+	// resolvable; the consumer falls back to ids for missing positions.
+	GroupNames []string `json:"group_names,omitempty"`
 }
 
 func GenerateKeyPair() (*KeyPair, error) {
@@ -34,7 +48,13 @@ func GenerateKeyPair() (*KeyPair, error) {
 	}, nil
 }
 
-func SignToken(privKeyB64, userID, domain string, method auth.Method, expiration time.Duration) (string, error) {
+// SignToken mints a session JWT for the given user and domain. email,
+// groups, and groupNames, when non-empty, are embedded so the proxy can
+// authorise and stamp identity for policy-aware middlewares without a
+// management round-trip on every cookie-bearing request. groupNames
+// pairs positionally with groups; pass nil when names couldn't be
+// resolved.
+func SignToken(privKeyB64, userID, email, domain string, method auth.Method, groups, groupNames []string, expiration time.Duration) (string, error) {
 	privKeyBytes, err := base64.StdEncoding.DecodeString(privKeyB64)
 	if err != nil {
 		return "", fmt.Errorf("decode private key: %w", err)
@@ -56,7 +76,10 @@ func SignToken(privKeyB64, userID, domain string, method auth.Method, expiration
 			IssuedAt:  jwt.NewNumericDate(now),
 			NotBefore: jwt.NewNumericDate(now),
 		},
-		Method: method,
+		Method:     method,
+		Email:      email,
+		Groups:     append([]string(nil), groups...),
+		GroupNames: append([]string(nil), groupNames...),
 	}
 
 	token := jwt.NewWithClaims(jwt.SigningMethodEdDSA, claims)

management/internals/server/boot.go

@@ -10,8 +10,10 @@ import (
 	"slices"
 	"time"
 
+	"github.com/gorilla/mux"
 	grpcMiddleware "github.com/grpc-ecosystem/go-grpc-middleware/v2"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
+	"github.com/rs/cors"
 	"github.com/rs/xid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
@@ -19,7 +21,6 @@ import (
 	"google.golang.org/grpc/keepalive"
 
 	cachestore "github.com/eko/gocache/lib/v4/store"
-	"github.com/netbirdio/management-integrations/integrations"
 
 	"github.com/netbirdio/netbird/encryption"
 	"github.com/netbirdio/netbird/formatter/hook"
@@ -27,16 +28,20 @@ import (
 	accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager"
 	nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
 	"github.com/netbirdio/netbird/management/server/activity"
+	activitystore "github.com/netbirdio/netbird/management/server/activity/store"
 	nbcache "github.com/netbirdio/netbird/management/server/cache"
 	nbContext "github.com/netbirdio/netbird/management/server/context"
 	nbhttp "github.com/netbirdio/netbird/management/server/http"
 	"github.com/netbirdio/netbird/management/server/http/middleware"
+	"github.com/netbirdio/netbird/management/server/idp"
 	"github.com/netbirdio/netbird/management/server/store"
 	"github.com/netbirdio/netbird/management/server/telemetry"
 	mgmtProto "github.com/netbirdio/netbird/shared/management/proto"
 	"github.com/netbirdio/netbird/util/crypt"
 )
 
+const apiPrefix = "/api"
+
 var (
 	kaep = keepalive.EnforcementPolicy{
 		MinTime:             15 * time.Second,
@@ -94,12 +99,17 @@ func (s *BaseServer) Store() store.Store {
 
 func (s *BaseServer) EventStore() activity.Store {
 	return Create(s, func() activity.Store {
-		integrationMetrics, err := integrations.InitIntegrationMetrics(context.Background(), s.Metrics())
-		if err != nil {
-			log.Fatalf("failed to initialize integration metrics: %v", err)
+		var err error
+		key := s.Config.DataStoreEncryptionKey
+		if key == "" {
+			log.Debugf("generate new activity store encryption key")
+			key, err = crypt.GenerateKey()
+			if err != nil {
+				log.Fatalf("failed to generate event store encryption key: %v", err)
+			}
 		}
 
-		eventStore, _, err := integrations.InitEventStore(context.Background(), s.Config.Datadir, s.Config.DataStoreEncryptionKey, integrationMetrics)
+		eventStore, err := activitystore.NewSqlStore(context.Background(), s.Config.Datadir, key)
 		if err != nil {
 			log.Fatalf("failed to initialize event store: %v", err)
 		}
@@ -110,7 +120,7 @@ func (s *BaseServer) EventStore() activity.Store {
 
 func (s *BaseServer) APIHandler() http.Handler {
 	return Create(s, func() http.Handler {
-		httpAPIHandler, err := nbhttp.NewAPIHandler(context.Background(), s.AccountManager(), s.NetworksManager(), s.ResourcesManager(), s.RoutesManager(), s.GroupsManager(), s.GeoLocationManager(), s.AuthManager(), s.Metrics(), s.IntegratedValidator(), s.ProxyController(), s.PermissionsManager(), s.PeersManager(), s.SettingsManager(), s.ZonesManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, s.RateLimiter())
+		httpAPIHandler, err := nbhttp.NewAPIHandler(context.Background(), s.Router(), s.AccountManager(), s.NetworksManager(), s.ResourcesManager(), s.RoutesManager(), s.GroupsManager(), s.GeoLocationManager(), s.AuthManager(), s.Metrics(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, s.RateLimiter(), s.IsValidChildAccount)
 		if err != nil {
 			log.Fatalf("failed to create API handler: %v", err)
 		}
@@ -118,6 +128,22 @@ func (s *BaseServer) APIHandler() http.Handler {
 	})
 }
 
+// IDPHandler returns the HTTP handler for the embedded IdP (Dex), or nil if
+// the deployment isn't using the embedded variant.
+func (s *BaseServer) IDPHandler() http.Handler {
+	embeddedIdP, ok := s.IdpManager().(*idp.EmbeddedIdPManager)
+	if !ok || embeddedIdP == nil {
+		return nil
+	}
+	return cors.AllowAll().Handler(embeddedIdP.Handler())
+}
+
+func (s *BaseServer) Router() *mux.Router {
+	return Create(s, func() *mux.Router {
+		return mux.NewRouter().PathPrefix(apiPrefix).Subrouter()
+	})
+}
+
 func (s *BaseServer) RateLimiter() *middleware.APIRateLimiter {
 	return Create(s, func() *middleware.APIRateLimiter {
 		cfg, enabled := middleware.RateLimiterConfigFromEnv()

management/internals/server/controllers.go

@@ -19,6 +19,7 @@ import (
 	"github.com/netbirdio/netbird/management/server"
 	"github.com/netbirdio/netbird/management/server/auth"
 	"github.com/netbirdio/netbird/management/server/integrations/integrated_validator"
+	"github.com/netbirdio/netbird/management/server/integrations/integrated_validator/validator"
 	"github.com/netbirdio/netbird/management/server/integrations/port_forwarding"
 	"github.com/netbirdio/netbird/management/server/job"
 	nbjwt "github.com/netbirdio/netbird/shared/auth/jwt"
@@ -38,7 +39,7 @@ func (s *BaseServer) JobManager() *job.Manager {
 
 func (s *BaseServer) IntegratedValidator() integrated_validator.IntegratedValidator {
 	return Create(s, func() integrated_validator.IntegratedValidator {
-		integratedPeerValidator, err := integrations.NewIntegratedValidator(
+		integratedPeerValidator, err := validator.NewIntegratedValidator(
 			context.Background(),
 			s.PeersManager(),
 			s.SettingsManager(),
@@ -112,7 +113,11 @@ func (s *BaseServer) AuthManager() auth.Manager {
 
 func (s *BaseServer) EphemeralManager() ephemeral.Manager {
 	return Create(s, func() ephemeral.Manager {
-		return manager.NewEphemeralManager(s.Store(), s.PeersManager())
+		em := manager.NewEphemeralManager(s.Store(), s.PeersManager())
+		if metrics := s.Metrics(); metrics != nil {
+			em.SetMetrics(metrics.EphemeralPeersMetrics())
+		}
+		return em
 	})
 }
 

management/internals/server/modules.go

@@ -57,13 +57,7 @@ func (s *BaseServer) GeoLocationManager() geolocation.Geolocation {
 
 func (s *BaseServer) PermissionsManager() permissions.Manager {
 	return Create(s, func() permissions.Manager {
-		manager := integrations.InitPermissionsManager(s.Store(), s.Metrics().GetMeter())
-
-		s.AfterInit(func(s *BaseServer) {
-			manager.SetAccountManager(s.AccountManager())
-		})
-
-		return manager
+		return permissions.NewManager(s.Store())
 	})
 }
 
@@ -153,7 +147,6 @@ func (s *BaseServer) IdpManager() idp.Manager {
 			return idpManager
 		}
 
-
 		return nil
 	})
 }
@@ -235,3 +228,7 @@ func (s *BaseServer) ReverseProxyDomainManager() *manager.Manager {
 		return &m
 	})
 }
+
+func (s *BaseServer) IsValidChildAccount(_ context.Context, _, _, _ string) bool {
+	return false
+}

management/internals/server/server.go

@@ -193,7 +193,7 @@ func (s *BaseServer) Start(ctx context.Context) error {
 		log.WithContext(srvCtx).Infof("running gRPC backward compatibility server: %s", compatListener.Addr().String())
 	}
 
-	rootHandler := s.handlerFunc(srvCtx, s.GRPCServer(), s.APIHandler(), s.Metrics().GetMeter())
+	rootHandler := s.handlerFunc(srvCtx, s.GRPCServer(), s.APIHandler(), s.IDPHandler(), s.Metrics().GetMeter())
 	switch {
 	case s.certManager != nil:
 		// a call to certManager.Listener() always creates a new listener so we do it once
@@ -304,7 +304,7 @@ func (s *BaseServer) SetHandlerFunc(handler http.Handler) {
 	log.Tracef("custom handler set successfully")
 }
 
-func (s *BaseServer) handlerFunc(_ context.Context, gRPCHandler *grpc.Server, httpHandler http.Handler, meter metric.Meter) http.Handler {
+func (s *BaseServer) handlerFunc(_ context.Context, gRPCHandler *grpc.Server, httpHandler http.Handler, idpHandler http.Handler, meter metric.Meter) http.Handler {
 	// Check if a custom handler was set (for multiplexing additional services)
 	if customHandler, ok := s.GetContainer("customHandler"); ok {
 		if handler, ok := customHandler.(http.Handler); ok {
@@ -323,6 +323,8 @@ func (s *BaseServer) handlerFunc(_ context.Context, gRPCHandler *grpc.Server, ht
 			gRPCHandler.ServeHTTP(writer, request)
 		case request.URL.Path == wsproxy.ProxyPath+wsproxy.ManagementComponent:
 			wsProxy.Handler().ServeHTTP(writer, request)
+		case idpHandler != nil && strings.HasPrefix(request.URL.Path, "/oauth2"):
+			idpHandler.ServeHTTP(writer, request)
 		default:
 			httpHandler.ServeHTTP(writer, request)
 		}

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 {
@@ -394,6 +579,13 @@ func (s *ProxyServiceServer) sendSnapshot(ctx context.Context, conn *proxyConnec
 		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 := conn.stream.Send(&proto.GetMappingUpdateResponse{
 			Mapping:             mappings[i:end],
 			InitialSyncComplete: end == len(mappings),
@@ -425,18 +617,14 @@ func (s *ProxyServiceServer) snapshotServiceMappings(ctx context.Context, conn *
 		return nil, fmt.Errorf("get services from store: %w", err)
 	}
 
+	oidcCfg := s.GetOIDCValidationConfig()
 	var mappings []*proto.ProxyMapping
 	for _, service := range services {
 		if !service.Enabled || service.ProxyCluster == "" || service.ProxyCluster != conn.address {
 			continue
 		}
 
-		token, err := s.tokenStore.GenerateToken(service.AccountID, service.ID, s.proxyTokenTTL())
-		if err != nil {
-			return nil, fmt.Errorf("generate auth token for service %s: %w", service.ID, err)
-		}
-
-		m := service.ToProtoMapping(rpservice.Create, token, s.GetOIDCValidationConfig())
+		m := service.ToProtoMapping(rpservice.Create, "", oidcCfg)
 		if !proxyAcceptsMapping(conn, m) {
 			continue
 		}
@@ -457,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
 			}
@@ -472,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()
@@ -538,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)
@@ -670,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
 	}
@@ -688,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.
@@ -749,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
 	}
@@ -838,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
 	}
@@ -846,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 {
@@ -858,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 {
@@ -1122,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 {
@@ -1133,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,
 	)
 }
@@ -1241,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,
@@ -1254,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,
@@ -1288,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
 	}
 
@@ -1303,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
 }
 
@@ -1339,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/proxy_snapshot_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -172,3 +173,55 @@ func TestSendSnapshot_EmptySnapshot(t *testing.T) {
 	assert.Empty(t, stream.messages[0].Mapping)
 	assert.True(t, stream.messages[0].InitialSyncComplete)
 }
+
+type hookingStream struct {
+	grpc.ServerStream
+	onSend func(*proto.GetMappingUpdateResponse)
+}
+
+func (s *hookingStream) Send(m *proto.GetMappingUpdateResponse) error {
+	if s.onSend != nil {
+		s.onSend(m)
+	}
+	return nil
+}
+
+func (s *hookingStream) Context() context.Context     { return context.Background() }
+func (s *hookingStream) SetHeader(metadata.MD) error  { return nil }
+func (s *hookingStream) SendHeader(metadata.MD) error { return nil }
+func (s *hookingStream) SetTrailer(metadata.MD)       {}
+func (s *hookingStream) SendMsg(any) error            { return nil }
+func (s *hookingStream) RecvMsg(any) error            { return nil }
+
+func TestSendSnapshot_TokensRemainValidUnderSlowSend(t *testing.T) {
+	const cluster = "cluster.example.com"
+	const batchSize = 2
+	const totalServices = 6
+	const ttl = 100 * time.Millisecond
+	const sendDelay = 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
+
+	var validateErrs []error
+	stream := &hookingStream{
+		onSend: func(resp *proto.GetMappingUpdateResponse) {
+			for _, m := range resp.Mapping {
+				if err := s.tokenStore.ValidateAndConsume(m.AuthToken, m.AccountId, m.Id); err != nil {
+					validateErrs = append(validateErrs, fmt.Errorf("svc %s: %w", m.Id, err))
+				}
+			}
+			time.Sleep(sendDelay)
+		},
+	}
+	conn := &proxyConnection{proxyID: "proxy-a", address: cluster, stream: stream}
+
+	require.NoError(t, s.sendSnapshot(context.Background(), conn))
+	require.Empty(t, validateErrs,
+		"tokens must remain valid even when batches are sent slowly: lazy per-batch generation guarantees freshness")
+}

management/internals/shared/grpc/server.go

@@ -522,10 +522,11 @@ func (s *Server) sendJob(ctx context.Context, peerKey wgtypes.Key, job *job.Even
 }
 
 func (s *Server) cancelPeerRoutines(ctx context.Context, accountID string, peer *nbpeer.Peer, streamStartTime time.Time) {
-	unlock := s.acquirePeerLockByUID(ctx, peer.Key)
+	uncanceledCTX := context.WithoutCancel(ctx)
+	unlock := s.acquirePeerLockByUID(uncanceledCTX, peer.Key)
 	defer unlock()
 
-	s.cancelPeerRoutinesWithoutLock(ctx, accountID, peer, streamStartTime)
+	s.cancelPeerRoutinesWithoutLock(uncanceledCTX, accountID, peer, streamStartTime)
 }
 
 func (s *Server) cancelPeerRoutinesWithoutLock(ctx context.Context, accountID string, peer *nbpeer.Peer, streamStartTime time.Time) {

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,21 +324,29 @@ 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
 }
 
+func (m *testValidateSessionServiceManager) DeleteAccountCluster(_ context.Context, _, _, _ string) error {
+	return nil
+}
+
 type testValidateSessionProxyManager struct{}
 
-func (m *testValidateSessionProxyManager) Connect(_ context.Context, _, _, _ string, _ *string, _ *proxy.Capabilities) error {
+func (m *testValidateSessionProxyManager) Connect(_ context.Context, _, _, _, _ string, _ *string, _ *proxy.Capabilities) (*proxy.Proxy, error) {
+	return nil, nil
+}
+
+func (m *testValidateSessionProxyManager) Disconnect(_ context.Context, _, _ string) error {
 	return nil
 }
 
-func (m *testValidateSessionProxyManager) Disconnect(_ context.Context, _ string) error {
+func (m *testValidateSessionProxyManager) Heartbeat(_ context.Context, _ *proxy.Proxy) error {
 	return nil
 }
 
-func (m *testValidateSessionProxyManager) Heartbeat(_ context.Context, _, _, _ string) error {
+func (m *testValidateSessionProxyManager) DeleteAccountCluster(_ context.Context, _, _ string) error {
 	return nil
 }