diff --git a/server/channels/app/platform/cluster_handlers.go b/server/channels/app/platform/cluster_handlers.go index 2fa5e2b70c..75865d811f 100644 --- a/server/channels/app/platform/cluster_handlers.go +++ b/server/channels/app/platform/cluster_handlers.go @@ -66,11 +66,18 @@ func (ps *PlatformService) ClearSessionCacheForUserSkipClusterSend(userID string ps.invalidateWebConnSessionCacheForUserSkipClusterSend(userID) } -func (ps *PlatformService) ClearSessionCacheForAllUsersSkipClusterSend() { +// ClearSessionCacheForAllUsersSkipClusterSend purges the in-memory +// session cache and invalidates every WebConn on this node. The hub +// fan-out runs even if the cache purge fails; the purge error is +// returned so wrappers can propagate it. +func (ps *PlatformService) ClearSessionCacheForAllUsersSkipClusterSend() error { ps.logger.Info("Purging sessions cache") - if err := ps.ClearAllUsersSessionCacheLocal(); err != nil { + err := ps.ClearAllUsersSessionCacheLocal() + if err != nil { ps.logger.Error("Failed to purge session cache", mlog.Err(err)) } + ps.invalidateWebConnSessionCacheForAllUsersSkipClusterSend() + return err } func (ps *PlatformService) clusterClearSessionCacheForUserHandler(msg *model.ClusterMessage) { @@ -78,7 +85,9 @@ func (ps *PlatformService) clusterClearSessionCacheForUserHandler(msg *model.Clu } func (ps *PlatformService) clusterClearSessionCacheForAllUsersHandler(msg *model.ClusterMessage) { - ps.ClearSessionCacheForAllUsersSkipClusterSend() + if err := ps.ClearSessionCacheForAllUsersSkipClusterSend(); err != nil { + ps.logger.Error("Failed to clear session cache for all users from cluster handler", mlog.Err(err)) + } } func (ps *PlatformService) clusterBusyStateChgHandler(msg *model.ClusterMessage) { @@ -102,6 +111,17 @@ func (ps *PlatformService) invalidateWebConnSessionCacheForUserSkipClusterSend(u } } +// invalidateWebConnSessionCacheForAllUsersSkipClusterSend signals +// every hub on this node to invalidate the cached session state of +// all of its WebConns. Companion to ClearAllUsersSessionCacheLocal. +func (ps *PlatformService) invalidateWebConnSessionCacheForAllUsersSkipClusterSend() { + for _, hub := range ps.hubs { + if hub != nil { + hub.InvalidateAll() + } + } +} + func (ps *PlatformService) InvalidateAllCachesSkipSend() *model.AppError { ps.logger.Info("Purging all caches") if err := ps.ClearAllUsersSessionCacheLocal(); err != nil { diff --git a/server/channels/app/platform/cluster_handlers_revoke_test.go b/server/channels/app/platform/cluster_handlers_revoke_test.go new file mode 100644 index 0000000000..e4ef9c22bd --- /dev/null +++ b/server/channels/app/platform/cluster_handlers_revoke_test.go @@ -0,0 +1,97 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package platform + +import ( + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/mattermost/mattermost/server/public/model" +) + +// TestRevokeSessionsFromAllUsersInvalidatesWebConnSession asserts that +// after the public RevokeSessionsFromAllUsers entry point returns, +// every live WebConn on this node — across multiple users (hashed to +// different hubs) and multiple connections per user — has its cached +// session reset to the authenticated-as-no-one state. +func TestRevokeSessionsFromAllUsersInvalidatesWebConnSession(t *testing.T) { + mainHelper.Parallel(t) + th := Setup(t).InitBasic() + defer th.TearDown() + + s := httptest.NewServer(dummyWebsocketHandler(t)) + defer s.Close() + + // Spread connections across multiple hubs via GetHubForUserId's + // hash(userID) mod len(hubs) sharding, and use multiple conns per + // user to cover the multi-device case. + type userConns struct { + userID string + wcs []*WebConn + } + users := []*userConns{ + {userID: th.BasicUser.Id}, + {userID: th.BasicUser2.Id}, + } + for range 4 { + users = append(users, &userConns{userID: model.NewId()}) + } + + for _, u := range users { + preWarmStatusOnline(th, u.userID) + } + + const connsPerUser = 2 + for _, u := range users { + for range connsPerUser { + session, err := th.Service.CreateSession(th.Context, &model.Session{ + UserId: u.userID, + }) + require.NoError(t, err) + + session.ExpiresAt = model.GetMillis() + time.Hour.Milliseconds() + + wc := registerDummyWebConn(t, th, s.Listener.Addr(), session) + t.Cleanup(func() { wc.Close() }) + u.wcs = append(u.wcs, wc) + + waitForWebConnRegistered(t, th, session) + } + } + + for _, u := range users { + for _, wc := range u.wcs { + require.NotNil(t, wc.GetSession(), + "precondition: webconn for user %q must have a cached session before revoke", u.userID) + require.Greater(t, wc.GetSessionExpiresAt(), model.GetMillis(), + "precondition: cached expiry for user %q must be in the future before revoke", u.userID) + require.NotEmpty(t, wc.GetSessionToken(), + "precondition: webconn for user %q must have a cached session token before revoke", u.userID) + } + } + + require.NoError(t, th.Service.RevokeSessionsFromAllUsers(), + "RevokeSessionsFromAllUsers should not error") + + require.Eventually(t, func() bool { + for _, u := range users { + for _, wc := range u.wcs { + if wc.GetSession() != nil { + return false + } + if wc.GetSessionExpiresAt() != 0 { + return false + } + if wc.GetSessionToken() != "" { + return false + } + } + } + return true + }, 5*time.Second, 25*time.Millisecond, + "RevokeSessionsFromAllUsers did not invalidate every live WebConn across all hubs") +} diff --git a/server/channels/app/platform/cluster_handlers_test.go b/server/channels/app/platform/cluster_handlers_test.go new file mode 100644 index 0000000000..18b6b3dd9a --- /dev/null +++ b/server/channels/app/platform/cluster_handlers_test.go @@ -0,0 +1,112 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package platform + +import ( + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/mattermost/mattermost/server/public/model" +) + +// waitForWebConnRegistered blocks until the hub has processed the +// WebConn registration. Without it, a test can race past the async +// register channel and signal invalidation against an empty connIndex. +func waitForWebConnRegistered(t *testing.T, th *TestHelper, session *model.Session) { + t.Helper() + require.Eventually(t, func() bool { + return th.Service.SessionIsRegistered(*session) + }, 2*time.Second, 10*time.Millisecond, + "WebConn for session %q (user %q) was not registered with the hub in time", + session.Id, session.UserId) +} + +// preWarmStatusOnline marks the user online before any WebConn is +// created so the async SetStatusOnline goroutine in NewWebConn skips +// the broadcast path. Otherwise the broadcast can race with the +// invalidation and re-populate the WebConn's cached session via +// IsBasicAuthenticated, flipping the post-invalidate assertions. +func preWarmStatusOnline(th *TestHelper, userID string) { + th.Service.AddStatusCacheSkipClusterSend(&model.Status{ + UserId: userID, + Status: model.StatusOnline, + LastActivityAt: model.GetMillis(), + }) +} + +// TestClearSessionCacheInvalidatesWebConnSession asserts that after either +// the per-user or the global session-cache clear runs, every matching +// active WebSocket connection has its cached session reset to the +// authenticated-as-no-one state (GetSession() == nil and +// GetSessionExpiresAt() == 0). +func TestClearSessionCacheInvalidatesWebConnSession(t *testing.T) { + mainHelper.Parallel(t) + + tests := []struct { + name string + revoke func(ps *PlatformService, userID string) + }{ + { + name: "PerUserRevokeInvalidatesWebConnSession", + revoke: func(ps *PlatformService, userID string) { + ps.ClearSessionCacheForUserSkipClusterSend(userID) + }, + }, + { + name: "GlobalRevokeInvalidatesWebConnSession", + revoke: func(ps *PlatformService, _ string) { + _ = ps.ClearSessionCacheForAllUsersSkipClusterSend() + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + s := httptest.NewServer(dummyWebsocketHandler(t)) + defer s.Close() + + session, err := th.Service.CreateSession(th.Context, &model.Session{ + UserId: th.BasicUser.Id, + }) + require.NoError(t, err) + + // Pin a future expiry so IsBasicAuthenticated trusts the + // cached session and doesn't re-validate against the store. + session.ExpiresAt = model.GetMillis() + time.Hour.Milliseconds() + + preWarmStatusOnline(th, th.BasicUser.Id) + + wc := registerDummyWebConn(t, th, s.Listener.Addr(), session) + defer wc.Close() + + waitForWebConnRegistered(t, th, session) + + require.NotNil(t, wc.GetSession(), + "precondition: webconn must have a cached session before revoke") + require.Greater(t, wc.GetSessionExpiresAt(), model.GetMillis(), + "precondition: webconn cached session expiry must be in the future before revoke") + + tt.revoke(th.Service, th.BasicUser.Id) + + // Hub invalidation is async, so poll for the end state. + require.Eventually(t, func() bool { + return wc.GetSession() == nil && wc.GetSessionExpiresAt() == 0 + }, 2*time.Second, 25*time.Millisecond, + "webconn cached session was not invalidated after %s; "+ + "expected GetSession()==nil and GetSessionExpiresAt()==0, "+ + "but got GetSession()!=nil=%t, GetSessionExpiresAt()=%d, GetSessionToken()=%q", + tt.name, + wc.GetSession() != nil, + wc.GetSessionExpiresAt(), + wc.GetSessionToken(), + ) + }) + } +} diff --git a/server/channels/app/platform/session.go b/server/channels/app/platform/session.go index bfb440083f..f4d9a2eb0a 100644 --- a/server/channels/app/platform/session.go +++ b/server/channels/app/platform/session.go @@ -110,9 +110,11 @@ func (ps *PlatformService) ClearUserSessionCache(userID string) { } func (ps *PlatformService) ClearAllUsersSessionCache() error { - if err := ps.ClearAllUsersSessionCacheLocal(); err != nil { - return err - } + // Mirrors the per-user shape: the SkipClusterSend helper handles the + // local cache purge and the WebConn hub fan-out, then we broadcast to + // peer nodes. The broadcast still runs on local-purge failure so peers + // can act independently. + err := ps.ClearSessionCacheForAllUsersSkipClusterSend() if ps.clusterIFace != nil { msg := &model.ClusterMessage{ @@ -121,7 +123,7 @@ func (ps *PlatformService) ClearAllUsersSessionCache() error { } ps.clusterIFace.SendClusterMessage(msg) } - return nil + return err } func (ps *PlatformService) GetSession(c request.CTX, token string) (*model.Session, error) { diff --git a/server/channels/app/platform/web_hub.go b/server/channels/app/platform/web_hub.go index b38924742b..95f90c6b02 100644 --- a/server/channels/app/platform/web_hub.go +++ b/server/channels/app/platform/web_hub.go @@ -84,6 +84,7 @@ type Hub struct { stop chan struct{} didStop chan struct{} invalidateUser chan string + invalidateAll chan struct{} activity chan *webConnActivityMessage directMsg chan *webConnDirectMessage explicitStop bool @@ -106,6 +107,7 @@ func newWebHub(ps *PlatformService) *Hub { stop: make(chan struct{}), didStop: make(chan struct{}), invalidateUser: make(chan string), + invalidateAll: make(chan struct{}), activity: make(chan *webConnActivityMessage), directMsg: make(chan *webConnDirectMessage), checkRegistered: make(chan *webConnSessionMessage), @@ -453,6 +455,15 @@ func (h *Hub) InvalidateUser(userID string) { } } +// InvalidateAll invalidates the cached session state of every WebConn +// registered with this hub. Global counterpart of InvalidateUser. +func (h *Hub) InvalidateAll() { + select { + case h.invalidateAll <- struct{}{}: + case <-h.stop: + } +} + // UpdateActivity sets the LastUserActivityAt field for the connection // of the user. func (h *Hub) UpdateActivity(userID, sessionToken string, activityAt int64) { @@ -654,6 +665,18 @@ func (h *Hub) Start() { closeAndRemoveConn(connIndex, webConn) } } + case <-h.invalidateAll: + // Mirrors the invalidateUser arm across every conn, + // also clearing the session token so the next + // IsBasicAuthenticated check short-circuits instead + // of re-fetching from the cache. + for webConn := range connIndex.All() { + webConn.InvalidateCache() + webConn.SetSessionToken("") + } + if *h.platform.Config().ServiceSettings.EnableWebHubChannelIteration { + connIndex.clearChannels() + } case activity := <-h.activity: for webConn := range connIndex.ForUser(activity.userID) { if !webConn.Active.Load() { @@ -947,6 +970,16 @@ func (i *hubConnectionIndex) ForChannel(channelID string) iter.Seq[*WebConn] { return maps.Keys(i.byChannelID[channelID]) } +// clearChannels empties the channel-routing index in one shot. Intended +// for paths that have already invalidated every conn registered with +// the hub: any broadcast addressed to a channel will be filtered out +// upstream by ShouldSendEvent, so the routing entries are dead weight +// until conns either re-handshake or fully reconnect (both of which +// repopulate the index via Add). +func (i *hubConnectionIndex) clearChannels() { + clear(i.byChannelID) +} + // ForUserActiveCount returns the number of active connections for a userID func (i *hubConnectionIndex) ForUserActiveCount(id string) int { cnt := 0 diff --git a/server/channels/app/session.go b/server/channels/app/session.go index 112e0117d6..0d21276f17 100644 --- a/server/channels/app/session.go +++ b/server/channels/app/session.go @@ -241,7 +241,9 @@ func (a *App) ClearSessionCacheForUserSkipClusterSend(userID string) { } func (a *App) ClearSessionCacheForAllUsersSkipClusterSend() { - a.Srv().Platform().ClearSessionCacheForAllUsersSkipClusterSend() + if err := a.Srv().Platform().ClearSessionCacheForAllUsersSkipClusterSend(); err != nil { + a.Srv().Platform().Log().Error("Failed to clear session cache for all users", mlog.Err(err)) + } } func (a *App) RevokeSessionsForDeviceId(c request.CTX, userID string, deviceID string, currentSessionId string) *model.AppError {