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

@@ -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
 

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.

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).
 

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) {

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)
 }

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)
+		}
 	}
 }
 

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
 }

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,

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

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

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 {

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

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"]

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())