From 33f99b97e5adfa18c07dec4fa8d3d718308e9a25 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sun, 4 Oct 2026 02:54:02 -0700 Subject: [PATCH] feat: end every session on sign-out-everywhere through RevokeOtherSessions with cache eviction after commit, refuse refresh on a revoked session, close a user's realtime sockets on any session revocation via a SESSIONS_REVOKED event, and verify tokens without a purpose claim for no flow --- AGENTS.md | 4 +- cmd/backend/main.go | 4 ++ .../docs/api/reference/account-org.mdx | 4 +- internal/app/token/gen.go | 11 +-- internal/app/token/logout.go | 52 +++----------- internal/app/token/purpose_test.go | 21 +++--- internal/app/token/refresh.go | 1 + internal/app/token/service.go | 11 +++ internal/app/token/sessions.go | 17 ++++- internal/app/token/verify.go | 14 +--- internal/infrastructure/pubsub/events.go | 23 ++++++ internal/repository/pg_token.go | 71 +------------------ realtime/lib/realtime/event_broadcaster.ex | 10 +++ .../test/realtime/event_broadcaster_test.exs | 10 +++ 14 files changed, 107 insertions(+), 146 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 74cff93c6..034853719 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -219,10 +219,10 @@ Every instance that has not updated yet runs the code an attacker can read here. ### Sessions and tokens -- **every token carries a purpose** and is verified against the one purpose its consumer accepts (`internal/app/token/config.go`: `access`, `refresh`, `ws`, `login`, `registration`, `reset`, `2fa`). A token minted for one flow must never verify in another. A new token type gets a new purpose constant, not a reused one +- **every token carries a purpose** and is verified against the one purpose its consumer accepts (`internal/app/token/config.go`: `access`, `refresh`, `ws`, `login`, `registration`, `reset`, `2fa`, `sso_link`). A token minted for one flow must never verify in another, and one with no purpose verifies for none. A new token type gets a new purpose constant, not a reused one - `VerifyToken` pins the algorithm to HS256 and requires an expiry. Do not relax either, and do not add a verification path that skips `token.VerifyToken` - `AUTH_SECRET` has a hard floor of `config.MinAuthSecretLength` (32 bytes) and the backend refuses to boot below it. The realtime service applies the same floor to `JWT_SECRET`, which is the same value. Neither check may become a warning -- banning a user, changing a password and revoking a session all terminate the sessions they invalidate. A new "lock this account" path must revoke too, or it locks nothing +- banning a user, changing a password and revoking a session all terminate the sessions they invalidate, and through `RevokeOtherSessions` close the user's realtime sockets. A new "lock this account" path must revoke too, or it locks nothing ### Access control: the rule that is easiest to get wrong diff --git a/cmd/backend/main.go b/cmd/backend/main.go index d92405525..887fc922b 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -868,6 +868,10 @@ func main() { } tokenService = token.NewService(primaryDB, tokenRepostory, cache, geoloc, authCfg.AuthSecret) + // A revoked session takes the user's open sockets with it. + if streamingPublisher != nil { + tokenService.WireRevocationPublisher(streamingPublisher) + } userService = user.NewService(userRepostory, cache) // A removed member's sessions and app authorizations end with the membership, on every removal path. diff --git a/docs/content/docs/api/reference/account-org.mdx b/docs/content/docs/api/reference/account-org.mdx index 29f82798c..535f98c2c 100644 --- a/docs/content/docs/api/reference/account-org.mdx +++ b/docs/content/docs/api/reference/account-org.mdx @@ -309,7 +309,7 @@ Self-service session management for the signed-in user. Revoking by id can only `POST /auth/logout` -Revokes the current session. Returns `204 No Content`. +Revokes the current session and closes the user's open realtime connections. Returns `204 No Content`. Auth: Session only (not available to API keys). @@ -317,7 +317,7 @@ Auth: Session only (not available to API keys). `POST /auth/logout-all` -Revokes every session for the user. Returns `204 No Content`. +Revokes every session for the user, this one included, and closes their open realtime connections. A revoked session can no longer be refreshed. Returns `204 No Content`. Auth: Session only (not available to API keys). diff --git a/internal/app/token/gen.go b/internal/app/token/gen.go index 510fa9c72..c214da5d7 100644 --- a/internal/app/token/gen.go +++ b/internal/app/token/gen.go @@ -101,11 +101,14 @@ func (s *tokenService) ReissueSession(ctx context.Context, userID uuid.UUID, cur mfaVerified = current.MFAVerified } - if err := s.RevokeOtherSessions(ctx, userID, uuid.Nil); err != nil { - return nil, err + err := s.revokeOthers(ctx, userID, uuid.Nil) + var tok *models.Token + if err == nil { + tok, err = s.generateSession(ctx, userID, "", ipaddr, userAgent, provider, orgID, mfaVerified) } - - return s.generateSession(ctx, userID, "", ipaddr, userAgent, provider, orgID, mfaVerified) + // After the new session exists, so the caller's socket reconnects with it. + s.notifyRevoked(ctx, userID) + return tok, err } func (s *tokenService) generateSession(ctx context.Context, userID uuid.UUID, email, ipaddr, userAgent, authProvider string, orgID *uuid.UUID, mfaVerified bool) (*models.Token, *errx.Error) { diff --git a/internal/app/token/logout.go b/internal/app/token/logout.go index d198becc2..13ebdd663 100644 --- a/internal/app/token/logout.go +++ b/internal/app/token/logout.go @@ -4,6 +4,7 @@ import ( "context" "time" + "github.com/google/uuid" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/infrastructure/db" ) @@ -18,7 +19,7 @@ func (s *tokenService) RevokeSession(ctx context.Context, accessToken string) *e tx, xerr := s.db.Begin(ctx) if xerr != nil { - db.CaptureError(err, "", nil, "begin") + db.CaptureError(xerr, "", nil, "begin") return errx.InternalError() } defer tx.Rollback(ctx) @@ -27,58 +28,25 @@ func (s *tokenService) RevokeSession(ctx context.Context, accessToken string) *e return err } - if err := s.deleteSession(ctx, sess.ID); err != nil { - return err - } - if err := tx.Commit(ctx); err != nil { db.CaptureError(err, "", nil, "commit") return errx.InternalError() } + // Evicted after the commit, so a concurrent read cannot re-cache the live row. + if err := s.evictRevoked(ctx, sess.ID, sess.UserID, now); err != nil { + return err + } + s.notifyRevoked(ctx, sess.UserID) + return nil } +// RevokeAllSession ends every session the caller's user holds, this one included. func (s *tokenService) RevokeAllSession(ctx context.Context, accessToken string) *errx.Error { sess, err := s.ValidateAccessToken(ctx, accessToken) if err != nil { return err } - - now := time.Now() - - tx, xerr := s.db.Begin(ctx) - if xerr != nil { - db.CaptureError(err, "", nil, "begin") - return errx.InternalError() - } - defer tx.Rollback(ctx) - - if err := s.tokenRepository.RevokeSession(ctx, tx, sess.ID, now); err != nil { - return err - } - - sess.RevokedAt = &now - - sessions, err := s.tokenRepository.FindExpiredSessions(ctx, sess.UserID, now.Truncate(SessionTTL)) - if err != nil { - return err - } - - for _, sess := range sessions { - if err := s.deleteSession(ctx, sess); err != nil { - return err - } - } - - if err := s.tokenRepository.RevokeSessions(ctx, sess.UserID); err != nil { - return err - } - - if err := tx.Commit(ctx); err != nil { - db.CaptureError(err, "", nil, "commit") - return errx.InternalError() - } - - return nil + return s.RevokeOtherSessions(ctx, sess.UserID, uuid.Nil) } diff --git a/internal/app/token/purpose_test.go b/internal/app/token/purpose_test.go index 69329cba2..f69f64a20 100644 --- a/internal/app/token/purpose_test.go +++ b/internal/app/token/purpose_test.go @@ -25,6 +25,7 @@ func TestTokenPurposeIsEnforced(t *testing.T) { PurposeRegistration, PurposePasswordReset, PurposeTwoFAPending, + PurposeSSOLink, } for _, minted := range purposes { @@ -48,26 +49,20 @@ func TestTokenPurposeIsEnforced(t *testing.T) { } } -// Tokens issued before the purpose claim existed are still inside their window -// during a deploy, so they read as access tokens rather than signing everyone -// out. They must not satisfy any other purpose. -func TestLegacyTokenIsAccessOnly(t *testing.T) { +// A token with no purpose claim verifies for no flow at all. +func TestTokenWithoutPurposeIsRefused(t *testing.T) { s := &tokenService{AuthSecret: "test-secret-at-least-32-characters-long"} now := time.Now() - legacy, err := s.GenerateTokenFor("", uuid.New(), uuid.New(), "", "nonce", now, now.Add(time.Minute)) + bare, err := s.GenerateTokenFor("", uuid.New(), uuid.New(), "", "nonce", now, now.Add(time.Minute)) if err != nil { t.Fatalf("mint: %v", err) } - if _, xerr := s.VerifyTokenFor(PurposeAccess, legacy); xerr != nil { - t.Errorf("a token with no purpose should still work as an access token: %v", xerr) - } - if _, xerr := s.VerifyTokenFor(PurposeWebSocket, legacy); xerr == nil { - t.Error("a token with no purpose must not open a websocket") - } - if _, xerr := s.VerifyTokenFor(PurposePasswordReset, legacy); xerr == nil { - t.Error("a token with no purpose must not pass as a password reset") + for _, p := range []string{PurposeAccess, PurposeRefresh, PurposeWebSocket, PurposePasswordReset, PurposeSSOLink} { + if _, xerr := s.VerifyTokenFor(p, bare); xerr == nil { + t.Errorf("a token with no purpose must not pass as %s", p) + } } } diff --git a/internal/app/token/refresh.go b/internal/app/token/refresh.go index 661022db2..89cfc2d0a 100644 --- a/internal/app/token/refresh.go +++ b/internal/app/token/refresh.go @@ -119,6 +119,7 @@ func (s *tokenService) refusedRefresh(ctx context.Context, sessionID, userID uui if xerr := s.evictRevoked(ctx, sessionID, userID, time.Now()); xerr != nil { errs.CaptureException(xerr) } + s.notifyRevoked(ctx, userID) } return errx.ErrToken } diff --git a/internal/app/token/service.go b/internal/app/token/service.go index ad1c2975d..133df6168 100644 --- a/internal/app/token/service.go +++ b/internal/app/token/service.go @@ -30,6 +30,7 @@ type TokenService interface { // require it, which is what the admin panel does. GenerateMFASession(ctx context.Context, userID uuid.UUID, email, ipaddr, userAgent, authProvider string) (*models.Token, *errx.Error) WireSignInAlerter(a SignInAlerter) + WireRevocationPublisher(p RevocationPublisher) GetSession(ctx context.Context, sessionID uuid.UUID) (*models.Session, *errx.Error) ValidateAccessToken(ctx context.Context, accessToken string) (*models.Session, *errx.Error) RefreshToken(ctx context.Context, refreshToken string) (*models.Token, *errx.Error) @@ -65,6 +66,7 @@ type tokenService struct { geo *geo.Client cache *cache.Cache signInAlert SignInAlerter + revocations RevocationPublisher AuthSecret string } @@ -79,6 +81,15 @@ type SignInAlerter interface { // WireSignInAlerter attaches the new-device alerter after construction. func (s *tokenService) WireSignInAlerter(a SignInAlerter) { s.signInAlert = a } +// RevocationPublisher announces that a user's sessions were revoked, so the +// realtime service drops their open sockets. Satisfied by the streaming publisher. +type RevocationPublisher interface { + PublishSessionsRevoked(ctx context.Context, userID uuid.UUID) +} + +// WireRevocationPublisher attaches the revocation announcer (nil = off). +func (s *tokenService) WireRevocationPublisher(p RevocationPublisher) { s.revocations = p } + func NewService(db *db.DB, tokenRepository repository.TokenRepository, cache *cache.Cache, geo *geo.Client, authSecret string) TokenService { return &tokenService{ db: db, diff --git a/internal/app/token/sessions.go b/internal/app/token/sessions.go index 6b53ef9a4..a8eb8119a 100644 --- a/internal/app/token/sessions.go +++ b/internal/app/token/sessions.go @@ -89,13 +89,28 @@ func (s *tokenService) RevokeSessionByID(ctx context.Context, userID, sessionID, if err := s.deleteSession(ctx, sessionID); err != nil { return err } + s.notifyRevoked(ctx, userID) return nil } // RevokeOtherSessions ends every active session except the caller's current -// one ("sign out everywhere else"). +// one ("sign out everywhere else"). uuid.Nil ends all of them. func (s *tokenService) RevokeOtherSessions(ctx context.Context, userID, currentSessionID uuid.UUID) *errx.Error { + err := s.revokeOthers(ctx, userID, currentSessionID) + s.notifyRevoked(ctx, userID) + return err +} + +// notifyRevoked closes the user's open websockets; each reconnects only with a +// ticket minted from a session that is still live. +func (s *tokenService) notifyRevoked(ctx context.Context, userID uuid.UUID) { + if s.revocations != nil { + s.revocations.PublishSessionsRevoked(ctx, userID) + } +} + +func (s *tokenService) revokeOthers(ctx context.Context, userID, currentSessionID uuid.UUID) *errx.Error { ids, err := s.tokenRepository.ListOtherActiveSessionIDs(ctx, userID, currentSessionID) if err != nil { return err diff --git a/internal/app/token/verify.go b/internal/app/token/verify.go index 84f28a720..8c8438c90 100644 --- a/internal/app/token/verify.go +++ b/internal/app/token/verify.go @@ -68,23 +68,13 @@ func (s *tokenService) ValidateAccessToken(ctx context.Context, accessToken stri } // VerifyTokenFor is VerifyToken plus the check that the token was minted for -// this flow. -// -// A token with no purpose is treated as an access token: tokens issued before -// the claim existed are still inside their 12-hour window during a deploy, and -// refusing them would sign everybody out. Every other flow demands its purpose -// explicitly, so that leniency cannot be used to spend a reset or challenge -// token as a session. +// this flow. A token with no purpose verifies for none. func (s *tokenService) VerifyTokenFor(purpose, tokenStr string) (*TokenClaims, *errx.Error) { claims, err := s.VerifyToken(tokenStr) if err != nil { return nil, err } - got := claims.Purpose - if got == "" { - got = PurposeAccess - } - if got != purpose { + if claims.Purpose == "" || claims.Purpose != purpose { return nil, errx.ErrToken } return claims, nil diff --git a/internal/infrastructure/pubsub/events.go b/internal/infrastructure/pubsub/events.go index d21f961c6..be57d5463 100644 --- a/internal/infrastructure/pubsub/events.go +++ b/internal/infrastructure/pubsub/events.go @@ -136,6 +136,10 @@ const ( // this internally (re-track / untrack / strip activity) to apply the new // policy live; it is not forwarded to web clients. EventPresencePolicyUpdated EventType = "PRESENCE_POLICY_UPDATED" + + // A user's sessions were revoked (sign out, password change, ban). The + // realtime service closes every socket the user holds; not forwarded. + EventSessionsRevoked EventType = "SESSIONS_REVOKED" ) // BaseEvent contains common fields for all events @@ -1003,6 +1007,25 @@ func (p *StreamingPublisher) PublishPresencePolicy(ctx context.Context, orgID uu } } +// PublishSessionsRevoked tells the realtime service to drop the user's sockets. +func (p *StreamingPublisher) PublishSessionsRevoked(ctx context.Context, userID uuid.UUID) { + if p == nil || p.client == nil { + return + } + event := &BaseEvent{ + EventType: EventSessionsRevoked, + UserID: userID.String(), + Timestamp: time.Now(), + } + attrs := map[string]string{ + "user_id": userID.String(), + "event_type": string(EventSessionsRevoked), + } + if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil { + // Best-effort: the sessions are already revoked, and every reconnect needs a live one. + } +} + // PublishToUser publishes a generic event to a user func (p *StreamingPublisher) PublishToUser(ctx context.Context, userID string, event interface{}) { if p.client == nil { diff --git a/internal/repository/pg_token.go b/internal/repository/pg_token.go index c935c5cec..8d02fefee 100644 --- a/internal/repository/pg_token.go +++ b/internal/repository/pg_token.go @@ -28,9 +28,6 @@ type TokenRepository interface { ListOtherActiveSessionIDs(ctx context.Context, userID, exceptID uuid.UUID) ([]uuid.UUID, *errx.Error) RevokeOtherSessions(ctx context.Context, userID, exceptID uuid.UUID) *errx.Error - FindExpiredSessions(ctx context.Context, userID uuid.UUID, cutoff time.Time) ([]uuid.UUID, *errx.Error) - RevokeSessions(ctx context.Context, userID uuid.UUID) *errx.Error - // Organization switching UpdateCurrentOrganization(ctx context.Context, sessionID uuid.UUID, orgID *uuid.UUID) *errx.Error DefaultOrganization(ctx context.Context, userID uuid.UUID) (*uuid.UUID, *errx.Error) @@ -266,7 +263,7 @@ func (r *tokenRepository) RefreshToken(ctx context.Context, sessionID uuid.UUID, UPDATE sessions SET last_refreshed_at = $5, access_nonce = $1, refresh_nonce = $2, previous_refresh_nonce = $3 - WHERE refresh_nonce = $3 AND id = $4 + WHERE refresh_nonce = $3 AND id = $4 AND revoked_at IS NULL ` params := []any{ @@ -336,72 +333,6 @@ func (r *tokenRepository) RevokeSession(ctx context.Context, tx pgx.Tx, sessionI return nil } -func (r *tokenRepository) FindExpiredSessions(ctx context.Context, userID uuid.UUID, cutoff time.Time) ([]uuid.UUID, *errx.Error) { - query := ` - SELECT id - FROM sessions - WHERE revoked_at IS NULL - AND last_refreshed_at < $1 - ` - - params := []any{ - cutoff, - } - - rows, err := r.DB.Query( - ctx, - query, - cutoff, - ) - if err != nil { - db.CaptureError(err, query, params, "query") - return nil, errx.InternalError() - } - defer rows.Close() - - var sessions []uuid.UUID - for rows.Next() { - var s uuid.UUID - if err := rows.Scan(&s); err != nil { - db.CaptureError(err, "", nil, "scan") - return nil, errx.InternalError() - } - sessions = append(sessions, s) - } - - if err := rows.Err(); err != nil { - db.CaptureError(err, "", nil, "rows_err") - return nil, errx.InternalError() - } - - return sessions, nil -} - -func (r *tokenRepository) RevokeSessions(ctx context.Context, userID uuid.UUID) *errx.Error { - query := ` - UPDATE sessions - SET revoked_at = now() - WHERE revoked_at IS NULL - AND user_id = $1 - ` - - params := []any{ - userID, - } - - _, err := r.DB.Exec( - ctx, - query, - params..., - ) - if err != nil { - db.CaptureError(err, query, params, "exec") - return errx.InternalError() - } - - return nil -} - func (r *tokenRepository) UpdateCurrentOrganization(ctx context.Context, sessionID uuid.UUID, orgID *uuid.UUID) *errx.Error { query := ` UPDATE sessions diff --git a/realtime/lib/realtime/event_broadcaster.ex b/realtime/lib/realtime/event_broadcaster.ex index 18d6bb0e4..46daf7e03 100644 --- a/realtime/lib/realtime/event_broadcaster.ex +++ b/realtime/lib/realtime/event_broadcaster.ex @@ -11,6 +11,16 @@ defmodule Realtime.EventBroadcaster do @doc """ Broadcast a decoded event map. Unknown shapes are ignored. """ + # A revoked session closes every socket the user holds; each reconnect needs + # a credential that is still live. A control signal, so nothing is forwarded. + def broadcast(%{"event_type" => "SESSIONS_REVOKED", "user_id" => user_id}) + when is_binary(user_id) and user_id != "" do + RealtimeWeb.Endpoint.broadcast("user_socket:#{user_id}", "disconnect", %{}) + :ok + end + + def broadcast(%{"event_type" => "SESSIONS_REVOKED"}), do: :ok + def broadcast(event) when is_map(event) do user_id = event["user_id"] event_type = event["event_type"] diff --git a/realtime/test/realtime/event_broadcaster_test.exs b/realtime/test/realtime/event_broadcaster_test.exs index 062c43abb..9e95abe8f 100644 --- a/realtime/test/realtime/event_broadcaster_test.exs +++ b/realtime/test/realtime/event_broadcaster_test.exs @@ -25,6 +25,16 @@ defmodule Realtime.EventBroadcasterTest do assert EventBroadcaster.removed_member(%{"event_type" => "EMAIL_SENT"}) == nil end + test "a session revocation disconnects the user's sockets" do + RealtimeWeb.Endpoint.subscribe("user_socket:#{@removed}") + EventBroadcaster.broadcast(%{"event_type" => "SESSIONS_REVOKED", "user_id" => @removed}) + + assert_receive %Phoenix.Socket.Broadcast{ + event: "disconnect", + topic: "user_socket:" <> @removed + } + end + test "a removal disconnects the removed user's sockets" do RealtimeWeb.Endpoint.subscribe("user_socket:#{@removed}") EventBroadcaster.broadcast(removal())