Files
481a240a1e MM-68543 Invalidate active WebConn session cache on global session revocation (#36332)
* MM-68543 Invalidate active WebConn session cache on global session revocation

Mirrors the per-user revocation pattern (ClearUserSessionCache ->
ClearSessionCacheForUserSkipClusterSend -> hub fan-out) for the global
revocation path so that ClearAllUsersSessionCache invokes the same
local-side primitive on the originating node as the cluster handler
runs on remote nodes. Also covers single-node deployments where the
cluster broadcast was previously the only trigger of the WebConn
invalidation.

Adds a Hub.InvalidateAll fan-out primitive on the websocket hub and
two contract tests covering the SkipClusterSend variant and the
production RevokeSessionsFromAllUsers entry point.

Made-with: Cursor

* MM-68543 Restore error propagation on ClearAllUsersSessionCache

The previous commit moved the local-side work into
ClearSessionCacheForAllUsersSkipClusterSend, which returned no error,
so ClearAllUsersSessionCache started always returning nil even when the
underlying session-cache purge failed.

Make the helper return the cache-purge error and propagate it back
through ClearAllUsersSessionCache, restoring the historical error
contract for callers (RevokeSessionsFromAllUsers,
App.ClearSessionCacheForAllUsers, TestCache). The hub fan-out and the
cluster broadcast still run unconditionally so security invalidation
happens even on local-purge failure.

Made-with: Cursor

* MM-68543 Trim comments and fix unchecked errcheck on App wrapper

Address review feedback: trim verbose comments across the touched
files and check the error returned by ClearSessionCacheForAllUsersSkipClusterSend
in the App-level wrapper to fix the golangci-lint errcheck failure
introduced when the helper started returning an error.

Made-with: Cursor

* MM-68543 Wipe channel routing index instead of rebuilding it on global revoke

Rebuilding byChannelID per user via InvalidateCMCacheForUser issues a
GetAllChannelMembersForUser DB query for every user with a live conn
on the hub, which is wasted work when those conns have just been
invalidated. Replace it with a single clear() of byChannelID, hidden
behind a small clearChannels() helper. Index entries repopulate
naturally as conns re-handshake or fully reconnect.

---------

Co-authored-by: Hanzo Dev <[email protected]>
2026-05-07 09:01:36 -04:00

303 lines
9.1 KiB
Go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package platform
import (
"fmt"
"strconv"
"time"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/public/shared/mlog"
"github.com/mattermost/mattermost/server/public/shared/request"
"github.com/mattermost/mattermost/server/v8/platform/services/cache"
)
func (ps *PlatformService) CreateSession(rctx request.CTX, session *model.Session) (*model.Session, error) {
session.Token = ""
session, err := ps.Store.Session().Save(rctx, session)
if err != nil {
return nil, err
}
if err := ps.AddSessionToCache(session); err != nil {
ps.Logger().Error("Failed to add session to cache", mlog.Err(err))
}
return session, nil
}
func (ps *PlatformService) GetSessionContext(rctx request.CTX, token string) (*model.Session, error) {
return ps.Store.Session().Get(rctx, token)
}
func (ps *PlatformService) GetSessions(rctx request.CTX, userID string) ([]*model.Session, error) {
return ps.Store.Session().GetSessions(rctx, userID)
}
func (ps *PlatformService) GetLRUSessions(rctx request.CTX, userID string, limit uint64, offset uint64) ([]*model.Session, error) {
return ps.Store.Session().GetLRUSessions(rctx, userID, limit, offset)
}
func (ps *PlatformService) AddSessionToCache(session *model.Session) error {
return ps.sessionCache.SetWithExpiry(session.Token, session, time.Duration(int64(*ps.Config().ServiceSettings.SessionCacheInMinutes))*time.Minute)
}
func (ps *PlatformService) ClearUserSessionCacheLocal(userID string) {
var toDelete []string
// First, we iterate over the entire session cache.
err := ps.sessionCache.Scan(func(keys []string) error {
if len(keys) == 0 {
return nil
}
// This always needs to be model.Session, not *model.Session.
// Otherwise the msp unmarshaler will fail to work.
toPass := allocateCacheTargets[model.Session](len(keys))
errs := ps.sessionCache.GetMulti(keys, toPass)
for i, err := range errs {
if err != nil {
if err != cache.ErrKeyNotFound {
return err
}
continue
}
gotSession := toPass[i].(*model.Session)
if gotSession == nil {
ps.logger.Warn("Found nil session in ClearUserSessionCacheLocal. This is not expected")
continue
}
// If we find the userID matches the passed userID,
// we mark it up for deletion.
if gotSession.UserId == userID {
toDelete = append(toDelete, keys[i])
if m := ps.metricsIFace; m != nil {
m.IncrementMemCacheInvalidationCounterSession()
}
}
}
return nil
})
if err != nil {
ps.logger.Warn("Error while scanning in ClearUserSessionCacheLocal", mlog.Err(err))
return
}
// Now, we delete everything.
err = ps.sessionCache.RemoveMulti(toDelete)
if err != nil {
ps.logger.Warn("Error while removing keys in ClearUserSessionCacheLocal", mlog.Err(err))
return
}
}
func (ps *PlatformService) ClearAllUsersSessionCacheLocal() error {
return ps.sessionCache.Purge()
}
func (ps *PlatformService) ClearUserSessionCache(userID string) {
ps.ClearSessionCacheForUserSkipClusterSend(userID)
if ps.clusterIFace != nil {
msg := &model.ClusterMessage{
Event: model.ClusterEventClearSessionCacheForUser,
SendType: model.ClusterSendReliable,
Data: []byte(userID),
}
ps.clusterIFace.SendClusterMessage(msg)
}
}
func (ps *PlatformService) ClearAllUsersSessionCache() error {
// 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{
Event: model.ClusterEventClearSessionCacheForAllUsers,
SendType: model.ClusterSendReliable,
}
ps.clusterIFace.SendClusterMessage(msg)
}
return err
}
func (ps *PlatformService) GetSession(rctx request.CTX, token string) (*model.Session, error) {
var session model.Session
if err := ps.sessionCache.Get(token, &session); err == nil {
if m := ps.metricsIFace; m != nil {
m.IncrementMemCacheHitCounterSession()
}
} else {
if m := ps.metricsIFace; m != nil {
m.IncrementMemCacheMissCounterSession()
}
}
if session.Id != "" {
return &session, nil
}
return ps.GetSessionContext(rctx, token)
}
func (ps *PlatformService) GetSessionByID(rctx request.CTX, sessionID string) (*model.Session, error) {
return ps.Store.Session().Get(rctx, sessionID)
}
func (ps *PlatformService) RevokeSessionsFromAllUsers() error {
// revoke tokens before sessions so they can't be used to relogin
nErr := ps.Store.OAuth().RemoveAllAccessData()
if nErr != nil {
return fmt.Errorf("%s: %w", nErr.Error(), DeleteAllAccessDataError)
}
err := ps.Store.Session().RemoveAllSessions()
if err != nil {
return err
}
if err := ps.ClearAllUsersSessionCache(); err != nil {
ps.logger.Error("Failed to clear session cache", mlog.Err(err))
}
return nil
}
func (ps *PlatformService) RevokeSessionsForDeviceId(rctx request.CTX, userID string, deviceID string, currentSessionId string) error {
sessions, err := ps.Store.Session().GetSessions(rctx, userID)
if err != nil {
return err
}
for _, session := range sessions {
if session.DeviceId == deviceID && session.Id != currentSessionId {
rctx.Logger().Debug("Revoking sessionId for userId. Re-login with the same device Id", mlog.String("session_id", session.Id), mlog.String("user_id", userID))
if err := ps.RevokeSession(rctx, session); err != nil {
rctx.Logger().Warn("Could not revoke session for device", mlog.String("device_id", deviceID), mlog.Err(err))
}
}
}
return nil
}
func (ps *PlatformService) RevokeSession(rctx request.CTX, session *model.Session) error {
if session.IsOAuth {
if err := ps.RevokeAccessToken(rctx, session.Token); err != nil {
return err
}
} else {
if err := ps.Store.Session().Remove(session.Id); err != nil {
return fmt.Errorf("%s: %w", err.Error(), DeleteSessionError)
}
}
ps.ClearUserSessionCache(session.UserId)
return nil
}
func (ps *PlatformService) RevokeAccessToken(rctx request.CTX, token string) error {
session, _ := ps.GetSession(rctx, token)
schan := make(chan error, 1)
go func() {
schan <- ps.Store.Session().Remove(token)
close(schan)
}()
if _, err := ps.Store.OAuth().GetAccessData(token); err != nil {
return fmt.Errorf("%s: %w", err.Error(), GetTokenError)
}
if err := ps.Store.OAuth().RemoveAccessData(token); err != nil {
return fmt.Errorf("%s: %w", err.Error(), DeleteTokenError)
}
if err := <-schan; err != nil {
return fmt.Errorf("%s: %w", err.Error(), DeleteSessionError)
}
if session != nil {
ps.ClearUserSessionCache(session.UserId)
}
return nil
}
// SetSessionExpireInHours sets the session's expiry the specified number of hours
// relative to either the session creation date or the current time, depending
// on the `ExtendSessionOnActivity` config setting.
func (ps *PlatformService) SetSessionExpireInHours(session *model.Session, hours int) {
if session.CreateAt == 0 || *ps.Config().ServiceSettings.ExtendSessionLengthWithActivity {
session.ExpiresAt = model.GetMillis() + (1000 * 60 * 60 * int64(hours))
} else {
session.ExpiresAt = session.CreateAt + (1000 * 60 * 60 * int64(hours))
}
}
func (ps *PlatformService) ExtendSessionExpiry(session *model.Session, newExpiry int64) error {
if err := ps.Store.Session().UpdateExpiresAt(session.Id, newExpiry); err != nil {
return err
}
// Update local cache. No need to invalidate cache for cluster as the session cache timeout
// ensures each node will get an extended expiry within the next 10 minutes.
// Worst case is another node may generate a redundant expiry update.
session.ExpiresAt = newExpiry
if err := ps.AddSessionToCache(session); err != nil {
ps.Logger().Error("Failed to update session cache", mlog.Err(err))
}
return nil
}
func (ps *PlatformService) UpdateSessionsIsGuest(rctx request.CTX, user *model.User, isGuest bool) error {
sessions, err := ps.GetSessions(rctx, user.Id)
if err != nil {
return err
}
_, err = ps.Store.Session().UpdateRoles(user.Id, user.GetRawRoles())
if err != nil {
return err
}
for _, session := range sessions {
session.AddProp(model.SessionPropIsGuest, strconv.FormatBool(isGuest))
err := ps.Store.Session().UpdateProps(session)
if err != nil {
rctx.Logger().Warn("Unable to update isGuest session", mlog.Err(err))
continue
}
if err := ps.AddSessionToCache(session); err != nil {
ps.Logger().Error("Failed to update session cache", mlog.Err(err))
}
}
return nil
}
func (ps *PlatformService) RevokeAllSessions(rctx request.CTX, userID string) error {
sessions, err := ps.Store.Session().GetSessions(rctx, userID)
if err != nil {
return fmt.Errorf("%s: %w", err.Error(), GetSessionError)
}
for _, session := range sessions {
if session.IsOAuth {
if err := ps.RevokeAccessToken(rctx, session.Token); err != nil {
return err
}
} else {
if err := ps.Store.Session().Remove(session.Id); err != nil {
return fmt.Errorf("%s: %w", err.Error(), DeleteSessionError)
}
}
}
ps.ClearUserSessionCache(userID)
return nil
}