[management] Affected peers for user updates (#7099)

management/server/account.go

@@ -33,6 +33,7 @@ import (
 	nbconfig "github.com/netbirdio/netbird/management/internals/server/config"
 	"github.com/netbirdio/netbird/management/server/account"
 	"github.com/netbirdio/netbird/management/server/activity"
+	"github.com/netbirdio/netbird/management/server/affectedpeers"
 	nbcache "github.com/netbirdio/netbird/management/server/cache"
 	nbcontext "github.com/netbirdio/netbird/management/server/context"
 	"github.com/netbirdio/netbird/management/server/geolocation"
@@ -1626,6 +1627,8 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 	var removeOldGroups []string
 	var hasChanges bool
 	var user *types.User
+	var change affectedpeers.Change
+	var snap *affectedpeers.Snapshot
 	err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
 		user, err = transaction.GetUserByUserID(ctx, store.LockingStrengthNone, userAuth.UserId)
 		if err != nil {
@@ -1664,14 +1667,25 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 			return fmt.Errorf("error saving user: %w", err)
 		}
 
+		allGroupChanges := slices.Concat(addNewGroups, removeOldGroups)
+		// The user's auto-groups changed, so the SSH rules authorizing them ship a new
+		// group -> user mapping even when no peer moves between groups.
+		change.UserGroupIDs = allGroupChanges
+
+		// The user's peers are the changed entity in every scenario the sync can
+		// produce — group membership, IPv6 assignment, SSH mappings — so they refresh
+		// together with every peer they can connect to, like on a regular peer update.
+		userPeers, err := transaction.GetUserPeers(ctx, store.LockingStrengthNone, userAuth.AccountId, userAuth.UserId)
+		if err != nil {
+			return fmt.Errorf("error getting user peers: %w", err)
+		}
+		for _, peer := range userPeers {
+			change.ChangedPeerIDs = append(change.ChangedPeerIDs, peer.ID)
+		}
+
 		// Propagate changes to peers if group propagation is enabled
 		if settings.GroupsPropagationEnabled {
-			peers, err := transaction.GetUserPeers(ctx, store.LockingStrengthNone, userAuth.AccountId, userAuth.UserId)
-			if err != nil {
-				return fmt.Errorf("error getting user peers: %w", err)
-			}
-
-			for _, peer := range peers {
+			for _, peer := range userPeers {
 				for _, g := range addNewGroups {
 					if err := transaction.AddPeerToGroup(ctx, userAuth.AccountId, peer.ID, g); err != nil {
 						return fmt.Errorf("error adding peer %s to group %s: %w", peer.ID, g, err)
@@ -1684,7 +1698,8 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 				}
 			}
 
-			allGroupChanges := slices.Concat(addNewGroups, removeOldGroups)
+			change.LinkGroups = allGroupChanges
+
 			if err = am.reconcileIPv6ForGroupChanges(ctx, transaction, userAuth.AccountId, allGroupChanges); err != nil {
 				return fmt.Errorf("reconcile IPv6 for group changes: %w", err)
 			}
@@ -1694,6 +1709,10 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 			}
 		}
 
+		if snap, err = affectedpeers.Load(ctx, transaction, userAuth.AccountId, change); err != nil {
+			return err
+		}
+
 		return nil
 	})
 	if err != nil {
@@ -1730,20 +1749,17 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 		}
 	}
 
-	removedGroupAffectsPeers, err := areGroupChangesAffectPeers(ctx, am.Store, userAuth.AccountId, removeOldGroups)
-	if err != nil {
-		return err
-	}
-
-	newGroupsAffectsPeers, err := areGroupChangesAffectPeers(ctx, am.Store, userAuth.AccountId, addNewGroups)
-	if err != nil {
-		return err
-	}
-
-	if removedGroupAffectsPeers || newGroupsAffectsPeers {
-		log.WithContext(ctx).Tracef("user %s: JWT group membership changed, updating account peers", userAuth.UserId)
-		am.BufferUpdateAccountPeers(ctx, userAuth.AccountId, types.UpdateReason{Resource: types.UpdateResourceUser, Operation: types.UpdateOperationUpdate})
-	}
+	log.WithContext(ctx).Tracef("user %s: JWT group membership changed, updating affected peers", userAuth.UserId)
+	bgCtx := context.WithoutCancel(ctx)
+	go func() {
+		affectedPeerIDs := snap.Expand(bgCtx, userAuth.AccountId, change)
+		if len(affectedPeerIDs) == 0 {
+			return
+		}
+		if err := am.networkMapController.BufferUpdateAffectedPeers(bgCtx, userAuth.AccountId, affectedPeerIDs, types.UpdateReason{Resource: types.UpdateResourceUser, Operation: types.UpdateOperationUpdate}); err != nil {
+			log.WithContext(bgCtx).Errorf("failed to update affected peers after JWT group sync for account %s: %v", userAuth.AccountId, err)
+		}
+	}()
 
 	return nil
 }
@@ -2426,30 +2442,24 @@ func (am *DefaultAccountManager) reconcileIPv6ForGroupChanges(ctx context.Contex
 		return fmt.Errorf("get account settings: %w", err)
 	}
 
-	if len(settings.IPv6EnabledGroups) == 0 {
-		return nil
-	}
-
-	enabledSet := make(map[string]struct{}, len(settings.IPv6EnabledGroups))
-	for _, gid := range settings.IPv6EnabledGroups {
-		enabledSet[gid] = struct{}{}
-	}
-
-	affected := false
-	for _, gid := range groupIDs {
-		if _, ok := enabledSet[gid]; ok {
-			affected = true
-			break
-		}
-	}
-
-	if !affected {
+	if !ipv6ReconcileNeeded(settings, groupIDs) {
 		return nil
 	}
 
 	return am.updatePeerIPv6Addresses(ctx, transaction, accountID, settings)
 }
 
+// ipv6ReconcileNeeded reports whether changes to the given groups trigger an IPv6
+// reconciliation.
+func ipv6ReconcileNeeded(settings *types.Settings, groupIDs []string) bool {
+	for _, groupID := range groupIDs {
+		if slices.Contains(settings.IPv6EnabledGroups, groupID) {
+			return true
+		}
+	}
+	return false
+}
+
 func (am *DefaultAccountManager) ensureIPv6Subnet(ctx context.Context, transaction store.Store, accountID string, settings *types.Settings, network *types.Network) error {
 	if settings.NetworkRangeV6.IsValid() {
 		network.NetV6 = net.IPNet{

management/server/affected_peers_jwt_test.go

@@ -0,0 +1,179 @@
+package server
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+
+	"github.com/netbirdio/netbird/management/server/affectedpeers"
+	nbpeer "github.com/netbirdio/netbird/management/server/peer"
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
+	"github.com/netbirdio/netbird/shared/auth"
+)
+
+// A user's auto-group change refreshes the destinations of the SSH rules authorizing
+// that group — they carry the group -> user mapping — even though no peer moved
+// between groups.
+func TestAffectedPeers_UserGroupChange_RefreshesSSHAuthorizedDestinations(t *testing.T) {
+	manager, s, accountID, peerIDs, groupIDs := setupAffectedPeersTest(t)
+	ctx := context.Background()
+
+	_, err := manager.SavePolicy(ctx, accountID, userID, &types.Policy{
+		Enabled: true,
+		Rules: []*types.PolicyRule{
+			{
+				Enabled:          true,
+				Sources:          []string{groupIDs[0]},
+				Destinations:     []string{groupIDs[1]},
+				Protocol:         types.PolicyRuleProtocolNetbirdSSH,
+				Action:           types.PolicyTrafficActionAccept,
+				AuthorizedGroups: map[string][]string{groupIDs[3]: {"root"}},
+			},
+		},
+	}, true)
+	require.NoError(t, err)
+
+	result := resolveAffected(t, s, accountID, affectedpeers.Change{UserGroupIDs: []string{groupIDs[3]}})
+	assert.ElementsMatch(t, []string{peerIDs[1]}, result,
+		"only the SSH rule's destination peers carry the changed group -> user mapping")
+
+	result = resolveAffected(t, s, accountID, affectedpeers.Change{UserGroupIDs: []string{groupIDs[4]}})
+	assert.Empty(t, result, "a group no SSH rule authorizes affects nobody")
+}
+
+// Creating, blocking or unblocking a user changes the account's allowed-user set, which
+// reaches only the destinations of the SSH rules that ship it.
+func TestAffectedPeers_AllowedUsersChange_RefreshesSSHDestinations(t *testing.T) {
+	manager, s, accountID, peerIDs, groupIDs := setupAffectedPeersTest(t)
+	ctx := context.Background()
+
+	// Ships the allowed-user set: an SSH rule naming no groups and no user.
+	_, err := manager.SavePolicy(ctx, accountID, userID, &types.Policy{
+		Enabled: true,
+		Rules: []*types.PolicyRule{{
+			Enabled:      true,
+			Sources:      []string{groupIDs[0]},
+			Destinations: []string{groupIDs[1]},
+			Protocol:     types.PolicyRuleProtocolNetbirdSSH,
+			Action:       types.PolicyTrafficActionAccept,
+		}},
+	}, true)
+	require.NoError(t, err)
+
+	// Does not ship it: an SSH rule that authorizes a specific group.
+	_, err = manager.SavePolicy(ctx, accountID, userID, &types.Policy{
+		Enabled: true,
+		Rules: []*types.PolicyRule{{
+			Enabled:          true,
+			Sources:          []string{groupIDs[2]},
+			Destinations:     []string{groupIDs[3]},
+			Protocol:         types.PolicyRuleProtocolNetbirdSSH,
+			Action:           types.PolicyTrafficActionAccept,
+			AuthorizedGroups: map[string][]string{groupIDs[0]: {"root"}},
+		}},
+	}, true)
+	require.NoError(t, err)
+
+	result := resolveAffected(t, s, accountID, affectedpeers.Change{AllowedUsersChanged: true})
+	assert.ElementsMatch(t, []string{peerIDs[1]}, result,
+		"only the destinations of the rule shipping the allowed-user set refresh")
+}
+
+// TestAffectedPeers_SyncUserJWTGroups_OnlyAffectedPeersUpdated verifies that a JWT
+// auto-group change updates only the user's peers and the peers linked to the changed
+// group through policies, instead of fanning out to the whole account.
+func TestAffectedPeers_SyncUserJWTGroups_OnlyAffectedPeersUpdated(t *testing.T) {
+	manager, updateManager, account, _, peer2, peer3 := setupNetworkMapTest(t)
+	ctx := context.Background()
+	accountID := account.Id
+
+	key, err := wgtypes.GeneratePrivateKey()
+	require.NoError(t, err)
+	userPeer, _, _, _, err := manager.AddPeer(ctx, accountID, "", userID, &nbpeer.Peer{
+		Key:  key.PublicKey().String(),
+		Meta: nbpeer.PeerSystemMeta{Hostname: "user-peer"},
+	}, false)
+	require.NoError(t, err)
+
+	policies, err := manager.Store.GetAccountPolicies(ctx, store.LockingStrengthNone, accountID)
+	require.NoError(t, err)
+	for _, p := range policies {
+		require.NoError(t, manager.Store.DeletePolicy(ctx, accountID, p.ID))
+	}
+
+	account, err = manager.Store.GetAccount(ctx, accountID)
+	require.NoError(t, err)
+	account.Settings.JWTGroupsEnabled = true
+	account.Settings.JWTGroupsClaimName = "groups"
+	account.Settings.GroupsPropagationEnabled = true
+	require.NoError(t, manager.Store.SaveAccount(ctx, account))
+
+	require.NoError(t, manager.CreateGroup(ctx, accountID, userID, &types.Group{ID: "jwt-grp", Name: "jwt-linked", Issued: types.GroupIssuedJWT, Peers: []string{}}))
+	require.NoError(t, manager.CreateGroup(ctx, accountID, userID, &types.Group{ID: "jwt-dest", Name: "jwt-dest", Peers: []string{peer2.ID}}))
+
+	_, err = manager.SavePolicy(ctx, accountID, userID, &types.Policy{
+		Enabled: true,
+		Rules: []*types.PolicyRule{
+			{
+				Enabled:       true,
+				Sources:       []string{"jwt-grp"},
+				Destinations:  []string{"jwt-dest"},
+				Bidirectional: true,
+				Action:        types.PolicyTrafficActionAccept,
+			},
+		},
+	}, true)
+	require.NoError(t, err)
+
+	updUser := updateManager.CreateChannel(ctx, userPeer.ID)
+	upd2 := updateManager.CreateChannel(ctx, peer2.ID)
+	upd3 := updateManager.CreateChannel(ctx, peer3.ID)
+	t.Cleanup(func() {
+		updateManager.CloseChannel(ctx, userPeer.ID)
+		updateManager.CloseChannel(ctx, peer2.ID)
+		updateManager.CloseChannel(ctx, peer3.ID)
+	})
+
+	userAuth := auth.UserAuth{
+		AccountId: accountID,
+		UserId:    userID,
+		Groups:    []string{"jwt-linked"},
+	}
+
+	t.Run("adding JWT group updates only linked peers", func(t *testing.T) {
+		drainPeerUpdates(updUser)
+		drainPeerUpdates(upd2)
+		drainPeerUpdates(upd3)
+
+		require.NoError(t, manager.SyncUserJWTGroups(ctx, userAuth))
+
+		peerShouldReceiveUpdate(t, updUser)
+		peerShouldReceiveUpdate(t, upd2)
+		peerShouldNotReceiveUpdate(t, upd3)
+
+		user, err := manager.Store.GetUserByUserID(ctx, store.LockingStrengthNone, userID)
+		require.NoError(t, err)
+		assert.Contains(t, user.AutoGroups, "jwt-grp")
+	})
+
+	t.Run("removing JWT group updates only linked peers", func(t *testing.T) {
+		drainPeerUpdates(updUser)
+		drainPeerUpdates(upd2)
+		drainPeerUpdates(upd3)
+
+		userAuth.Groups = nil
+		require.NoError(t, manager.SyncUserJWTGroups(ctx, userAuth))
+
+		peerShouldReceiveUpdate(t, updUser)
+		peerShouldReceiveUpdate(t, upd2)
+		peerShouldNotReceiveUpdate(t, upd3)
+
+		user, err := manager.Store.GetUserByUserID(ctx, store.LockingStrengthNone, userID)
+		require.NoError(t, err)
+		assert.NotContains(t, user.AutoGroups, "jwt-grp")
+	})
+}

management/server/affected_peers_user_test.go

@@ -0,0 +1,170 @@
+package server
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+
+	"github.com/netbirdio/netbird/management/server/activity"
+	nbpeer "github.com/netbirdio/netbird/management/server/peer"
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
+)
+
+// A user update refreshes only the peers its auto-group change reaches, and a user
+// update that changes no group membership refreshes nobody.
+func TestAffectedPeers_SaveUser_OnlyAffectedPeersUpdated(t *testing.T) {
+	manager, updateManager, account, _, peer2, peer3 := setupNetworkMapTest(t)
+	ctx := context.Background()
+	accountID := account.Id
+
+	const targetUserID = "target-user"
+	require.NoError(t, manager.Store.SaveUser(ctx, &types.User{
+		Id: targetUserID, AccountID: accountID, Role: types.UserRoleUser,
+	}))
+
+	key, err := wgtypes.GeneratePrivateKey()
+	require.NoError(t, err)
+	targetPeer, _, _, _, err := manager.AddPeer(ctx, accountID, "", targetUserID, &nbpeer.Peer{
+		Key:  key.PublicKey().String(),
+		Meta: nbpeer.PeerSystemMeta{Hostname: "target-peer"},
+	}, false)
+	require.NoError(t, err)
+
+	policies, err := manager.Store.GetAccountPolicies(ctx, store.LockingStrengthNone, accountID)
+	require.NoError(t, err)
+	for _, p := range policies {
+		require.NoError(t, manager.Store.DeletePolicy(ctx, accountID, p.ID))
+	}
+
+	account, err = manager.Store.GetAccount(ctx, accountID)
+	require.NoError(t, err)
+	account.Settings.GroupsPropagationEnabled = true
+	require.NoError(t, manager.Store.SaveAccount(ctx, account))
+
+	require.NoError(t, manager.CreateGroup(ctx, accountID, userID, &types.Group{ID: "ug-linked", Name: "ug-linked"}))
+	require.NoError(t, manager.CreateGroup(ctx, accountID, userID, &types.Group{ID: "ug-dest", Name: "ug-dest", Peers: []string{peer2.ID}}))
+
+	_, err = manager.SavePolicy(ctx, accountID, userID, &types.Policy{
+		Enabled: true,
+		Rules: []*types.PolicyRule{
+			{
+				Enabled:       true,
+				Sources:       []string{"ug-linked"},
+				Destinations:  []string{"ug-dest"},
+				Bidirectional: true,
+				Action:        types.PolicyTrafficActionAccept,
+			},
+		},
+	}, true)
+	require.NoError(t, err)
+
+	updTarget := updateManager.CreateChannel(ctx, targetPeer.ID)
+	upd2 := updateManager.CreateChannel(ctx, peer2.ID)
+	upd3 := updateManager.CreateChannel(ctx, peer3.ID)
+	t.Cleanup(func() {
+		updateManager.CloseChannel(ctx, targetPeer.ID)
+		updateManager.CloseChannel(ctx, peer2.ID)
+		updateManager.CloseChannel(ctx, peer3.ID)
+	})
+
+	t.Run("auto group change updates only linked peers", func(t *testing.T) {
+		drainPeerUpdates(updTarget)
+		drainPeerUpdates(upd2)
+		drainPeerUpdates(upd3)
+
+		_, err := manager.SaveUser(ctx, accountID, activity.SystemInitiator, &types.User{
+			Id: targetUserID, AccountID: accountID, Role: types.UserRoleUser,
+			AutoGroups: []string{"ug-linked"},
+		})
+		require.NoError(t, err)
+
+		peerShouldReceiveUpdate(t, updTarget)
+		peerShouldReceiveUpdate(t, upd2)
+		peerShouldNotReceiveUpdate(t, upd3)
+	})
+
+	t.Run("update without group changes refreshes nobody", func(t *testing.T) {
+		drainPeerUpdates(updTarget)
+		drainPeerUpdates(upd2)
+		drainPeerUpdates(upd3)
+
+		_, err := manager.SaveUser(ctx, accountID, activity.SystemInitiator, &types.User{
+			Id: targetUserID, AccountID: accountID, Role: types.UserRoleUser,
+			AutoGroups: []string{"ug-linked"}, Name: "renamed",
+		})
+		require.NoError(t, err)
+
+		peerShouldNotReceiveUpdate(t, updTarget)
+		peerShouldNotReceiveUpdate(t, upd2)
+		peerShouldNotReceiveUpdate(t, upd3)
+
+		user, err := manager.Store.GetUserByUserID(ctx, store.LockingStrengthNone, targetUserID)
+		require.NoError(t, err)
+		assert.Equal(t, "renamed", user.Name)
+	})
+
+	t.Run("auto group change reassigning IPv6 refreshes the changed peers and their observers", func(t *testing.T) {
+		account, err := manager.Store.GetAccount(ctx, accountID)
+		require.NoError(t, err)
+		account.Settings.IPv6EnabledGroups = []string{"ug-v6"}
+		require.NoError(t, manager.Store.SaveAccount(ctx, account))
+		require.NoError(t, manager.CreateGroup(ctx, accountID, userID, &types.Group{ID: "ug-v6", Name: "ug-v6"}))
+
+		drainPeerUpdates(updTarget)
+		drainPeerUpdates(upd2)
+		drainPeerUpdates(upd3)
+
+		_, err = manager.SaveUser(ctx, accountID, activity.SystemInitiator, &types.User{
+			Id: targetUserID, AccountID: accountID, Role: types.UserRoleUser,
+			AutoGroups: []string{"ug-linked", "ug-v6"}, Name: "renamed",
+		})
+		require.NoError(t, err)
+
+		// The reassigned peer refreshes with everyone it can reach: peer2 via the
+		// policy, but not peer3, which shares no group or policy with it.
+		peerShouldReceiveUpdate(t, updTarget)
+		peerShouldReceiveUpdate(t, upd2)
+		peerShouldNotReceiveUpdate(t, upd3)
+	})
+
+	t.Run("unblocking a user refreshes only the SSH rule destinations", func(t *testing.T) {
+		// An SSH rule that authorizes no group of its own ships the account's
+		// allowed-user set to its destinations, so those are the peers an unblock
+		// reaches — not the whole account.
+		_, err := manager.SavePolicy(ctx, accountID, userID, &types.Policy{
+			Enabled: true,
+			Rules: []*types.PolicyRule{{
+				Enabled:      true,
+				Sources:      []string{"ug-linked"},
+				Destinations: []string{"ug-dest"},
+				Protocol:     types.PolicyRuleProtocolNetbirdSSH,
+				Action:       types.PolicyTrafficActionAccept,
+			}},
+		}, true)
+		require.NoError(t, err)
+
+		blocked, err := manager.Store.GetUserByUserID(ctx, store.LockingStrengthNone, targetUserID)
+		require.NoError(t, err)
+		blocked.Blocked = true
+		require.NoError(t, manager.Store.SaveUser(ctx, blocked))
+
+		drainPeerUpdates(updTarget)
+		drainPeerUpdates(upd2)
+		drainPeerUpdates(upd3)
+
+		// Same auto-groups as the previous subtest left them, so no group change and
+		// no IPv6 reconciliation interferes: the unblock alone drives the refresh.
+		_, err = manager.SaveUser(ctx, accountID, activity.SystemInitiator, &types.User{
+			Id: targetUserID, AccountID: accountID, Role: types.UserRoleUser,
+			AutoGroups: []string{"ug-linked", "ug-v6"}, Name: "renamed",
+		})
+		require.NoError(t, err)
+
+		peerShouldReceiveUpdate(t, upd2)
+		peerShouldNotReceiveUpdate(t, upd3)
+	})
+}

management/server/affectedpeers/resolver.go

@@ -18,6 +18,7 @@ import (
 	"context"
 
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/exp/maps"
 
 	nbdns "github.com/netbirdio/netbird/dns"
 	rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
@@ -83,7 +84,7 @@ func (snap *Snapshot) loadCollections(ctx context.Context, s store.Store, accoun
 	hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.LinkGroups) > 0 || len(c.Resources) > 0
 	hasNetworkObject := len(c.Routers) > 0 || len(c.Resources) > 0 || len(c.Networks) > 0
 	// the resource<->router bridge can fire for any of these
-	needsRoutersResources := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || hasNetworkObject
+	needsRoutersResources := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || hasNetworkObject || len(c.UserGroupIDs) > 0 || c.AllowedUsersChanged
 
 	if needsRoutersResources {
 		if err := snap.loadPolicyRoutersResources(ctx, s, accountID); err != nil {
@@ -219,6 +220,18 @@ type Change struct {
 	// (correct when the peer's own attributes changed, e.g. IP/status).
 	OutputPeerIDs []string
 
+	// UserGroupIDs are groups whose USER membership changed (a user's auto-groups),
+	// as opposed to their peer membership. Peers ship the group -> user mapping only
+	// for the groups an SSH rule authorizes, so these refresh the destinations of the
+	// SSH rules authorizing them — independently of any peer moving between groups.
+	UserGroupIDs []string
+
+	// AllowedUsersChanged marks a change to the set of users allowed to open SSH
+	// sessions — a user was created, blocked or unblocked. That set is account-wide,
+	// and peers receive it through the SSH rules that name no group or user of their
+	// own, so those rules' destinations refresh.
+	AllowedUsersChanged bool
+
 	// LinkGroups are groups used ONLY to match policies/routes/routers and walk to the
 	// OPPOSITE side — they are never expanded to their own members. Use this when a
 	// peer's group membership changed: pass the peer in ChangedPeerIDs and its
@@ -240,6 +253,8 @@ func (c Change) isEmpty() bool {
 		len(c.Resources) == 0 &&
 		len(c.Networks) == 0 &&
 		len(c.PostureCheckIDs) == 0 &&
+		len(c.UserGroupIDs) == 0 &&
+		!c.AllowedUsersChanged &&
 		len(c.DistributionGroupIDs) == 0 &&
 		len(c.RemovedPeersByGroup) == 0 &&
 		len(c.LinkGroups) == 0 &&
@@ -359,6 +374,9 @@ func (r *resolver) walk() {
 		r.collectFromProxyServices()
 	}
 
+	r.collectFromSSHAuthorizedGroups()
+	r.collectFromAllowedUsers()
+
 	r.collectFromChangedRoutes(r.change.Routes)
 	r.collectFromChangedRouters(r.change.Routers)
 	r.collectFromChangedResources(r.change.Resources)
@@ -811,6 +829,59 @@ func (r *resolver) collectFromNameServers() {
 	}
 }
 
+// collectFromSSHAuthorizedGroups folds the destinations of the enabled SSH rules that
+// authorize a group whose user membership changed. Those destination peers carry the
+// group -> user mapping for the groups they authorize, so they refresh even when no
+// peer moved between groups.
+func (r *resolver) collectFromSSHAuthorizedGroups() {
+	if len(r.change.UserGroupIDs) == 0 {
+		return
+	}
+
+	changed := toSet(r.change.UserGroupIDs)
+	for _, policy := range r.policies() {
+		for _, rule := range policy.Rules {
+			if !rule.Enabled || rule.Protocol != types.PolicyRuleProtocolNetbirdSSH {
+				continue
+			}
+			if !anyInSet(maps.Keys(rule.AuthorizedGroups), changed) {
+				continue
+			}
+			log.WithContext(r.ctx).Tracef("collectFromSSHAuthorizedGroups: rule %s authorizes a changed user group -> folding its destinations", rule.ID)
+			r.foldPolicySideForRule(policy, rule, sideDestination)
+		}
+	}
+}
+
+// collectFromAllowedUsers folds the destinations of the rules that make a peer carry
+// the account's allowed-user set, for a change to who is in that set.
+func (r *resolver) collectFromAllowedUsers() {
+	if !r.change.AllowedUsersChanged {
+		return
+	}
+
+	for _, policy := range r.policies() {
+		for _, rule := range policy.Rules {
+			if !rule.Enabled || !ruleShipsAllowedUsers(rule) {
+				continue
+			}
+			log.WithContext(r.ctx).Tracef("collectFromAllowedUsers: rule %s ships the allowed-user set -> folding its destinations", rule.ID)
+			r.foldPolicySideForRule(policy, rule, sideDestination)
+		}
+	}
+}
+
+// ruleShipsAllowedUsers reports whether a rule makes its destination peers carry the
+// account's allowed-user set. It mirrors the network map's SSH requirements except for
+// the destination peer's own SSH flag, which the snapshot does not hold — so it folds a
+// superset and never misses a peer.
+func ruleShipsAllowedUsers(rule *types.PolicyRule) bool {
+	if rule.Protocol == types.PolicyRuleProtocolNetbirdSSH {
+		return len(rule.AuthorizedGroups) == 0 && rule.AuthorizedUser == ""
+	}
+	return types.PolicyRuleImpliesLegacySSH(rule)
+}
+
 func (r *resolver) collectFromDNSSettings() {
 	if len(r.linkGroups) == 0 || r.snap.dnsSettings == nil {
 		return

management/server/affectedpeers/resolver_test.go

@@ -85,6 +85,8 @@ func TestChangeIsEmpty(t *testing.T) {
 	assert.False(t, Change{Resources: []*resourceTypes.NetworkResource{{ID: "r"}}}.isEmpty())
 	assert.False(t, Change{Networks: []*networkTypes.Network{{ID: "n"}}}.isEmpty())
 	assert.False(t, Change{PostureCheckIDs: []string{"pc"}}.isEmpty())
+	assert.False(t, Change{UserGroupIDs: []string{"g"}}.isEmpty())
+	assert.False(t, Change{AllowedUsersChanged: true}.isEmpty())
 }
 
 func TestPolicyReferencesPostureChecks(t *testing.T) {

management/server/user.go

@@ -593,7 +593,8 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
 		return nil, err
 	}
 
-	var updateAccountPeers bool
+	var snaps []*affectedpeers.Snapshot
+	var changes []affectedpeers.Change
 	var peersToExpire []*nbpeer.Peer
 	var addUserEvents []func()
 	var usersToSave = make([]*types.User, 0, len(updates))
@@ -629,20 +630,25 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
 		}
 
 		err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
-			_, updatedUser, userPeersToExpire, userEvents, err := am.processUserUpdate(
+			change, updatedUser, userPeersToExpire, userEvents, err := am.processUserUpdate(
 				ctx, transaction, groupsMap, accountID, initiatorUserID, initiatorUser, update, addIfNotExists, settings,
 			)
 			if err != nil {
 				return fmt.Errorf("failed to process update for user %s: %w", update.Id, err)
 			}
 
-			updateAccountPeers = true
-
 			err = transaction.SaveUser(ctx, updatedUser)
 			if err != nil {
 				return fmt.Errorf("failed to save updated user %s: %w", update.Id, err)
 			}
 
+			snap, err := affectedpeers.Load(ctx, transaction, accountID, change)
+			if err != nil {
+				return err
+			}
+
+			snaps = append(snaps, snap)
+			changes = append(changes, change)
 			usersToSave = append(usersToSave, updatedUser)
 			addUserEvents = append(addUserEvents, userEvents...)
 			peersToExpire = append(peersToExpire, userPeersToExpire...)
@@ -683,11 +689,11 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
 			log.WithContext(ctx).Errorf("failed update expired peers: %s", err)
 			return nil, err
 		}
-	} else if updateAccountPeers {
+	} else if len(usersToSave) > 0 {
 		if err = am.Store.IncrementNetworkSerial(ctx, accountID); err != nil {
 			return nil, fmt.Errorf("failed to increment network serial: %w", err)
 		}
-		am.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceUser, Operation: types.UpdateOperationUpdate})
+		go am.dispatchAffected(ctx, accountID, snaps, changes)
 	}
 
 	return updatedUsersInfo, globalErr
@@ -759,19 +765,21 @@ func (am *DefaultAccountManager) prepareUserUpdateEvents(ctx context.Context, ac
 }
 
 func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transaction store.Store, groupsMap map[string]*types.Group,
-	accountID, initiatorUserId string, initiatorUser, update *types.User, addIfNotExists bool, settings *types.Settings) (bool, *types.User, []*nbpeer.Peer, []func(), error) {
+	accountID, initiatorUserId string, initiatorUser, update *types.User, addIfNotExists bool, settings *types.Settings) (affectedpeers.Change, *types.User, []*nbpeer.Peer, []func(), error) {
+
+	var change affectedpeers.Change
 
 	if update == nil {
-		return false, nil, nil, nil, status.Errorf(status.InvalidArgument, "provided user update is nil")
+		return change, nil, nil, nil, status.Errorf(status.InvalidArgument, "provided user update is nil")
 	}
 
 	oldUser, isNewUser, err := getUserOrCreateIfNotExists(ctx, transaction, accountID, update, addIfNotExists)
 	if err != nil {
-		return false, nil, nil, nil, err
+		return change, nil, nil, nil, err
 	}
 
 	if err := validateUserUpdate(groupsMap, initiatorUser, oldUser, update); err != nil {
-		return false, nil, nil, nil, err
+		return change, nil, nil, nil, err
 	}
 
 	// only auto groups, revoked status, and integration reference can be updated for now
@@ -792,13 +800,13 @@ func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transact
 	var transferredOwnerRole bool
 	result, err := handleOwnerRoleTransfer(ctx, transaction, initiatorUser, update)
 	if err != nil {
-		return false, nil, nil, nil, err
+		return change, nil, nil, nil, err
 	}
 	transferredOwnerRole = result
 
 	userPeers, err := transaction.GetUserPeers(ctx, store.LockingStrengthNone, updatedUser.AccountID, update.Id)
 	if err != nil {
-		return false, nil, nil, nil, err
+		return change, nil, nil, nil, err
 	}
 
 	var peersToExpire []*nbpeer.Peer
@@ -807,6 +815,32 @@ func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transact
 		peersToExpire = userPeers
 	}
 
+	// A user reaches a peer's network map only through the SSH rules: as part of a
+	// group -> user mapping, and as part of the account's allowed-user set. Creating,
+	// blocking or unblocking a user adds it to or removes it from both, so every group
+	// it maps into changes — including the All group that holds every active user.
+	// Otherwise only the auto-groups it joined or left do.
+	if isNewUser || oldUser.IsBlocked() != updatedUser.IsBlocked() {
+		change.AllowedUsersChanged = true
+		change.UserGroupIDs = slices.Concat(oldUser.AutoGroups, updatedUser.AutoGroups, allGroupIDs(groupsMap))
+	} else {
+		change.UserGroupIDs = slices.Concat(
+			util.Difference(oldUser.AutoGroups, updatedUser.AutoGroups),
+			util.Difference(updatedUser.AutoGroups, oldUser.AutoGroups),
+		)
+	}
+
+	// The user's peers are the changed entity in every scenario the update can
+	// produce — group membership, IPv6 assignment, SSH mappings — so they refresh
+	// together with every peer they can connect to, like on a regular peer update.
+	// An update that changes neither the auto-groups nor the active-user set has no
+	// peer-visible effect and refreshes nobody.
+	if len(change.UserGroupIDs) > 0 || change.AllowedUsersChanged {
+		for _, peer := range userPeers {
+			change.ChangedPeerIDs = append(change.ChangedPeerIDs, peer.ID)
+		}
+	}
+
 	var removedGroups, addedGroups []string
 	if update.AutoGroups != nil && settings.GroupsPropagationEnabled {
 		removedGroups = util.Difference(oldUser.AutoGroups, update.AutoGroups)
@@ -814,26 +848,38 @@ func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transact
 		for _, peer := range userPeers {
 			for _, groupID := range removedGroups {
 				if err := transaction.RemovePeerFromGroup(ctx, peer.ID, groupID); err != nil {
-					return false, nil, nil, nil, fmt.Errorf("failed to remove peer %s from group %s: %w", peer.ID, groupID, err)
+					return change, nil, nil, nil, fmt.Errorf("failed to remove peer %s from group %s: %w", peer.ID, groupID, err)
 				}
 			}
 			for _, groupID := range addedGroups {
 				if err := transaction.AddPeerToGroup(ctx, accountID, peer.ID, groupID); err != nil {
-					return false, nil, nil, nil, fmt.Errorf("failed to add peer %s to group %s: %w", peer.ID, groupID, err)
+					return change, nil, nil, nil, fmt.Errorf("failed to add peer %s to group %s: %w", peer.ID, groupID, err)
 				}
 			}
 		}
 
 		allGroupChanges := slices.Concat(removedGroups, addedGroups)
+		change.LinkGroups = allGroupChanges
+
 		if err := am.reconcileIPv6ForGroupChanges(ctx, transaction, accountID, allGroupChanges); err != nil {
-			return false, nil, nil, nil, fmt.Errorf("reconcile IPv6 for group changes: %w", err)
+			return change, nil, nil, nil, fmt.Errorf("reconcile IPv6 for group changes: %w", err)
 		}
 	}
 
-	updateAccountPeers := len(userPeers) > 0
 	userEventsToAdd := am.prepareUserUpdateEvents(ctx, updatedUser.AccountID, initiatorUserId, oldUser, updatedUser, transferredOwnerRole, isNewUser, removedGroups, addedGroups, transaction)
 
-	return updateAccountPeers, updatedUser, peersToExpire, userEventsToAdd, nil
+	return change, updatedUser, peersToExpire, userEventsToAdd, nil
+}
+
+// allGroupIDs returns the ID of the account's All group, which every active user maps
+// into, as a slice so callers can concatenate it.
+func allGroupIDs(groupsMap map[string]*types.Group) []string {
+	for _, group := range groupsMap {
+		if group.IsGroupAll() {
+			return []string{group.ID}
+		}
+	}
+	return nil
 }
 
 // getUserOrCreateIfNotExists retrieves the existing user or creates a new one if it doesn't exist.