diff --git a/management/server/account.go b/management/server/account.go index 619036d0c..700dfa04d 100644 --- a/management/server/account.go +++ b/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{ diff --git a/management/server/affected_peers_jwt_test.go b/management/server/affected_peers_jwt_test.go new file mode 100644 index 000000000..766745cd7 --- /dev/null +++ b/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") + }) +} diff --git a/management/server/affected_peers_user_test.go b/management/server/affected_peers_user_test.go new file mode 100644 index 000000000..c0dbbb84f --- /dev/null +++ b/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) + }) +} diff --git a/management/server/affectedpeers/resolver.go b/management/server/affectedpeers/resolver.go index 16a795539..cb2063ac9 100644 --- a/management/server/affectedpeers/resolver.go +++ b/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 diff --git a/management/server/affectedpeers/resolver_test.go b/management/server/affectedpeers/resolver_test.go index fe6ada347..b0c430c3c 100644 --- a/management/server/affectedpeers/resolver_test.go +++ b/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) { diff --git a/management/server/user.go b/management/server/user.go index fc8400e29..7c0a3088d 100644 --- a/management/server/user.go +++ b/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.