From f3a46a9be48e131d337ceac39e5ccdafafff7ba6 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Fri, 28 Aug 2026 11:12:27 -0700 Subject: [PATCH] fix: stop sockets outliving the session that authenticated them (#3658) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A socket was checked once at handshake and never again. Nothing in the backend disconnected one, so logout-everywhere, password reset, session revoke and suspension all left every connection streaming legacy FS entries, upload paths and notification bodies — up to 400 per account — on a credential that had already been revoked. Three gaps, three fixes: - The handshake skipped the suspension and pending-verification checks every authenticated HTTP route gets. `decideSocketAuth` now applies both. - `revokeCascade` reports which rows it revoked, AuthService announces that as `auth.sessions.revoked`, and SocketService drops the account's room. The whole room goes, not just the revoked session: narrowing it would need `fetchSockets`, which the adapter builds on `serverCount()` — and that calls node-redis's `send_command`, which ioredis does not implement. A connection whose session survived reconnects on its own and re-authenticates. - A bulk suspension writes `user.suspended` without touching `sessions`, so no revoke fires. A five-minute sweep re-verifies each live socket's token and drops the ones that no longer authenticate. De-duplicated by token, since a browser's tabs share one. --- src/backend/clients/event/types.ts | 16 +- src/backend/services/auth/AuthService.ts | 39 ++++- .../services/socket/SocketService.test.ts | 123 ++++++++++++++++ src/backend/services/socket/SocketService.ts | 137 +++++++++++++++++- src/backend/stores/session/SessionStore.js | 14 +- .../stores/session/SessionStore.test.ts | 17 ++- 6 files changed, 324 insertions(+), 22 deletions(-) diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index 2fc2c2004..734b35962 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -494,6 +494,13 @@ export type EventMap = { 'web.socket.connected': { socket: unknown; user: unknown }; 'web.socket.user-connected': { socket: unknown; user: unknown }; + /** + * Sessions were revoked for a user. Anything holding one of them open — a + * live socket, say — should drop it: `revoked_at` is otherwise only read on + * the next handshake. + */ + 'auth.sessions.revoked': { user_id: number; session_uids: string[] }; + // ---- Extension hooks / misc ---- 'puter.gui.addons': { prependHeadContent?: string[]; @@ -634,10 +641,11 @@ export type EventKey = keyof EventMap & string; // Generates a wildcard for every non-final dot-separated prefix of K. export type WildcardPrefixes = K extends `${infer Head}.${infer Tail}` - ? | `${Head}.*` - | (Tail extends `${string}.${string}` - ? `${Head}.${WildcardPrefixes}` - : never) + ? + | `${Head}.*` + | (Tail extends `${string}.${string}` + ? `${Head}.${WildcardPrefixes}` + : never) : never; export type ListenKey = EventKey | WildcardPrefixes; diff --git a/src/backend/services/auth/AuthService.ts b/src/backend/services/auth/AuthService.ts index 19fb2d83a..dc66028ea 100644 --- a/src/backend/services/auth/AuthService.ts +++ b/src/backend/services/auth/AuthService.ts @@ -518,7 +518,9 @@ export class AuthService extends PuterService { (sessionPayload.session_uid as string | undefined) ?? sessionPayload.uuid; if (!sessionUuid) return; - await this.stores.session.revokeCascade(sessionUuid); + this.#announceRevocation( + await this.stores.session.revokeCascade(sessionUuid), + ); } /** @@ -621,7 +623,7 @@ export class AuthService extends PuterService { const tokenUids = (await this.stores.session.accessTokenUidsForCascade( uuid, )) as string[]; - await this.stores.session.revokeCascade(uuid); + this.#announceRevocation(await this.stores.session.revokeCascade(uuid)); for (const tokenUid of tokenUids) { await this.#dropAccessTokenGrants(tokenUid); } @@ -658,6 +660,23 @@ export class AuthService extends PuterService { } } + /** + * Tell the rest of the system which sessions just went away. + * `sessions.revoked_at` is otherwise only consulted on the next handshake, + * which leaves anything already holding a session open — a live socket — + * running on a credential that no longer exists. + */ + #announceRevocation( + revoked: { userId: number; uuids: string[] } | null | undefined, + ): void { + if (!revoked?.userId || revoked.uuids.length === 0) return; + this.clients.event?.emit( + 'auth.sessions.revoked', + { user_id: revoked.userId, session_uids: revoked.uuids }, + {}, + ); + } + /** * Admin-driven cascade: revoke EVERY session row for the given user (web, * app, access_token, asset, worker). No actor context — this is the @@ -675,7 +694,9 @@ export class AuthService extends PuterService { if (!userId) return; const rows = await this.stores.session.getByUserId(userId); for (const row of rows) { - await this.stores.session.revokeCascade(row.uuid as string); + this.#announceRevocation( + await this.stores.session.revokeCascade(row.uuid as string), + ); } } @@ -692,7 +713,9 @@ export class AuthService extends PuterService { const rows = await this.stores.session.getByUserId(userId); for (const row of rows) { if (row.kind === 'web' || row.kind === 'app') { - await this.stores.session.revokeCascade(row.uuid as string); + this.#announceRevocation( + await this.stores.session.revokeCascade(row.uuid as string), + ); } } } @@ -713,9 +736,13 @@ export class AuthService extends PuterService { for (const row of rows) { if (row.kind === 'web') { if (!opts.includeCurrent && row.uuid === currentUuid) continue; - await this.stores.session.revokeCascade(row.uuid as string); + this.#announceRevocation( + await this.stores.session.revokeCascade(row.uuid as string), + ); } else if (row.kind === 'app' && opts.includeApps) { - await this.stores.session.revokeCascade(row.uuid as string); + this.#announceRevocation( + await this.stores.session.revokeCascade(row.uuid as string), + ); } } } diff --git a/src/backend/services/socket/SocketService.test.ts b/src/backend/services/socket/SocketService.test.ts index b82d34991..5bdbc492c 100644 --- a/src/backend/services/socket/SocketService.test.ts +++ b/src/backend/services/socket/SocketService.test.ts @@ -114,6 +114,32 @@ describe('decideSocketAuth', () => { expect(decision.reject.message).toMatch(/only user tokens/); }); + it('rejects a suspended user — the HTTP gate the handshake never runs', () => { + const decision = decideSocketAuth({ + actor: { + user: { id: 1, uuid: 'u-1', username: 'u', suspended: 1 }, + }, + } as unknown as AuthResult); + if (!('reject' in decision)) throw new Error('expected reject'); + expect(decision.reject.message).toMatch(/suspended/i); + }); + + it.each([ + [ + 'email', + { requires_email_confirmation: 1, email_confirmed: 0 }, + /confirm your email/i, + ], + ['phone', { requires_phone_verification: 1 }, /verify your phone/i], + ['card', { requires_card_verification: 1 }, /verify your card/i], + ])('rejects an account pending %s verification', (_label, flags, match) => { + const decision = decideSocketAuth({ + actor: { user: { id: 1, uuid: 'u-1', username: 'u', ...flags } }, + } as unknown as AuthResult); + if (!('reject' in decision)) throw new Error('expected reject'); + expect(decision.reject.message).toMatch(match); + }); + it('rejects when AuthService returned no actor at all', () => { const decision = decideSocketAuth({ invalid: true } as AuthResult); if (!('reject' in decision)) throw new Error('expected reject'); @@ -410,6 +436,103 @@ describe('SocketService (live socket.io)', () => { }); }); + // -- Evicting a connection whose credential is gone --------------- + // + // The handshake is the only place the credential was checked, so nothing + // here is covered by the cases above: a revoke, a suspension and a new + // verification requirement all leave a live feed open otherwise. + + it('drops the connection when its session is revoked', async () => { + const evicted = await createTestUser(server, { + username: 'sock-evicted', + password: 'sock-evicted-password', + }); + const row = await server.stores.user.getByUsername(evicted.username); + const socket = await connect({ auth_token: `Bearer ${evicted.token}` }); + expect(socket.connected).toBe(true); + + const authService = server.services.auth as unknown as { + revokeAllSessionsForUserId: (id: number) => Promise; + }; + await authService.revokeAllSessionsForUserId(row!.id); + + await vi.waitFor(() => expect(socket.connected).toBe(false), { + timeout: 5_000, + }); + }); + + it('leaves other accounts alone when one is revoked', async () => { + const kept = await createTestUser(server, { + username: 'sock-kept', + password: 'sock-kept-password', + }); + const revoked = await createTestUser(server, { + username: 'sock-revoked', + password: 'sock-revoked-password', + }); + const revokedRow = await server.stores.user.getByUsername( + revoked.username, + ); + + const keptSocket = await connect({ + auth_token: `Bearer ${kept.token}`, + }); + const revokedSocket = await connect({ + auth_token: `Bearer ${revoked.token}`, + }); + + const authService = server.services.auth as unknown as { + revokeAllSessionsForUserId: (id: number) => Promise; + }; + await authService.revokeAllSessionsForUserId(revokedRow!.id); + + await vi.waitFor(() => expect(revokedSocket.connected).toBe(false), { + timeout: 5_000, + }); + expect(keptSocket.connected).toBe(true); + keptSocket.disconnect(); + }); + + it('drops a connection whose account was suspended without a revoke', async () => { + // A bulk suspension writes `user.suspended` and never touches + // `sessions`, so eviction-on-revoke never fires — the periodic + // re-check is what closes it. + const suspended = await createTestUser(server, { + username: 'sock-suspended', + password: 'sock-suspended-password', + }); + const row = await server.stores.user.getByUsername(suspended.username); + const socket = await connect({ + auth_token: `Bearer ${suspended.token}`, + }); + expect(socket.connected).toBe(true); + + await server.clients.db.write( + 'UPDATE user SET suspended = 1 WHERE id = ?', + [row!.id], + ); + await server.stores.user.invalidateById(row!.id); + + await socketService.reauthenticateSockets(); + await vi.waitFor(() => expect(socket.connected).toBe(false), { + timeout: 5_000, + }); + }); + + it('leaves a still-valid connection up across a re-check', async () => { + const healthy = await createTestUser(server, { + username: 'sock-healthy', + password: 'sock-healthy-password', + }); + const socket = await connect({ auth_token: `Bearer ${healthy.token}` }); + + await socketService.reauthenticateSockets(); + await new Promise((r) => setTimeout(r, 200)); + + expect(socket.connected).toBe(true); + socket.disconnect(); + }); + it('gives a slot back when the connection closes', async () => { await withLimits({ perOrigin: 1, perUser: 100 }, async () => { const first = await connectFrom('https://e.example'); diff --git a/src/backend/services/socket/SocketService.ts b/src/backend/services/socket/SocketService.ts index 5719272ff..a3918526b 100644 --- a/src/backend/services/socket/SocketService.ts +++ b/src/backend/services/socket/SocketService.ts @@ -22,6 +22,10 @@ import type { Server as HttpServer } from 'node:http'; import { Server as SocketIOServer, type Socket } from 'socket.io'; import type { Actor } from '../../core/actor.js'; import { isAccessTokenActor, isAppActor } from '../../core/actor.js'; +import { + assertNotSuspended, + assertVerifiedAccount, +} from '../../core/http/middleware/gates.js'; import { CONCURRENT_SLOT_TTL_MS, acquireConcurrent, @@ -65,7 +69,12 @@ export type SocketAuthDecision = { accept: Actor } | { reject: Error }; * 2. Missing actor → generic `socket auth failed`. * 3. App-under-user / access-token actor → rejected with a specific message; * sockets only accept plain user actors. - * 4. Otherwise → accept the actor. + * 4. Suspended, or pending a verification → rejected. A socket carries the same + * filesystem entries, upload paths and notification bodies as the HTTP + * routes, which get these two from `requireAuthGate` / + * `requireVerifiedAccount`; the handshake is not in that chain, so it has to + * apply them itself. + * 5. Otherwise → accept the actor. * * Pure / no side effects — the middleware logs the reauth event. */ @@ -80,6 +89,15 @@ export const decideSocketAuth = (result: AuthResult): SocketAuthDecision => { if (isAppActor(actor) || isAccessTokenActor(actor)) { return { reject: new Error('socket auth: only user tokens accepted') }; } + try { + assertNotSuspended(actor.user); + assertVerifiedAccount(actor.user); + } catch (err) { + return { + reject: + err instanceof Error ? err : new Error('socket auth failed'), + }; + } return { accept: actor }; }; @@ -131,6 +149,12 @@ interface UploadProgressPayload { */ interface AuthenticatedSocket extends Socket { actor?: Actor; + /** + * The handshake token, kept for the periodic re-check. On the socket + * instance rather than in `socket.data`, which the adapter serializes to + * other nodes — a session token has no business travelling over it. + */ + authToken?: string; } /** @@ -152,6 +176,7 @@ interface AuthenticatedSocket extends Socket { */ export class SocketService extends PuterService { #io: SocketIOServer | null = null; + #reauthTimer: ReturnType | null = null; // -- Lifecycle --------------------------------------------------- @@ -194,6 +219,7 @@ export class SocketService extends PuterService { this.#installAuthMiddleware(); this.#installConnectionHandler(); this.#subscribeEventBus(); + this.#installReauthLoop(); } /** @@ -209,6 +235,10 @@ export class SocketService extends PuterService { } override onServerPrepareShutdown(): Promise { + if (this.#reauthTimer) { + clearInterval(this.#reauthTimer); + this.#reauthTimer = null; + } // Close the io server so existing sockets disconnect cleanly // before http's close() starts waiting for connections. return new Promise((resolve) => { @@ -298,7 +328,8 @@ export class SocketService extends PuterService { // `{ auth: { ... } }`, not the query string. puter-js uses // `io(url, { auth: { auth_token } })`. const handshakeAuth = socket.handshake.auth as - Record | undefined; + | Record + | undefined; const tokenRaw = typeof handshakeAuth?.auth_token === 'string' ? handshakeAuth.auth_token @@ -345,6 +376,7 @@ export class SocketService extends PuterService { } socket.actor = decision.accept; + socket.authToken = token; // user.id is numeric in the DB; stringify for room name // so adapter lookups key on a stable type. socket.join(String(decision.accept.user!.id)); @@ -491,6 +523,90 @@ export class SocketService extends PuterService { if (socket.disconnected) finish(); } + // -- Keeping a live connection honest ---------------------------- + // + // The handshake is the only place a socket's credential was ever checked, + // and a connection outlives it indefinitely — a desktop tab stays up for + // days. Two mechanisms close that gap: an eviction on revoke for the paths + // that know a session ended, and a periodic re-check for everything that + // changes without touching `sessions` (a bulk suspension, a verification + // requirement added by the abuse harness). + + /** How often a live socket's credential is re-verified. */ + static REAUTH_INTERVAL_MS = 5 * 60_000; + + /** + * Drop every socket on this node whose token no longer authenticates to the + * same accepted actor. De-duplicated by token: one browser's tabs share a + * session, so a sweep costs one check per credential, not per connection. + * + * Driven by the interval below; public because that timer isn't drivable + * from a test. + */ + async reauthenticateSockets(): Promise { + const io = this.#io; + const authService = this.services.auth as AuthService | undefined; + if (!io || !authService) return; + + const decisions = new Map(); + for (const raw of io.sockets.sockets.values()) { + const socket = raw as AuthenticatedSocket; + const token = socket.authToken; + // Nothing to re-check against — it can't be shown to still be + // valid, so it goes. + if (!token) { + socket.disconnect(true); + continue; + } + let decision = decisions.get(token); + if (!decision) { + try { + decision = decideSocketAuth( + await authService.authenticate(token, {}), + ); + } catch { + decision = { reject: new Error('socket reauth failed') }; + } + decisions.set(token, decision); + } + if ('reject' in decision) { + socket.disconnect(true); + continue; + } + // Refresh the actor so anything reading it off the socket sees the + // current row rather than the one from connect time. + socket.actor = decision.accept; + } + } + + #installReauthLoop(): void { + const timer = setInterval(() => { + void this.reauthenticateSockets().catch((err: unknown) => { + console.error('[socket] reauth sweep failed', err); + }); + }, SocketService.REAUTH_INTERVAL_MS); + timer.unref?.(); + this.#reauthTimer = timer; + } + + /** + * Close a user's connections after any of their sessions was revoked. + * Cluster-wide: `disconnectSockets` publishes through the adapter, so a + * revoke handled on one node reaches sockets terminated on another. + * + * Every connection for the account goes, not just the revoked session's. + * Narrowing would mean matching each socket to its session via + * `fetchSockets`, which the adapter implements on top of `serverCount()` — + * and that path is unavailable with our Redis client. Dropping the room is + * the safe direction: a connection whose session survived reconnects on its + * own, and its handshake re-authenticates. + */ + async #evictUserSockets(userId: number): Promise { + const io = this.#io; + if (!io || !userId) return; + await io.in(String(userId)).disconnectSockets(true); + } + async #allowSocketEvent(userId: number, event: string): Promise { return checkRateLimit( `socket:${event}:${userId}`, @@ -583,6 +699,16 @@ export class SocketService extends PuterService { this.#handleUploadProgress(data as UploadProgressPayload); }, ); + + this.clients.event.on( + 'auth.sessions.revoked', + (_key: string, data: unknown) => { + const { user_id } = data as { user_id: number }; + this.#evictUserSockets(user_id).catch((err: unknown) => { + console.error('[socket] session eviction failed', err); + }); + }, + ); } async #handleOuterGui(key: string, data: OuterGuiPayload): Promise { @@ -621,7 +747,8 @@ export class SocketService extends PuterService { // their next poll of /cache/last-change-timestamp. const originalSocketId = ( data.response as - { original_client_socket_id?: string } | undefined + | { original_client_socket_id?: string } + | undefined )?.original_client_socket_id; await this.send({ room: userId }, 'cache.updated', { timestamp, @@ -635,7 +762,9 @@ export class SocketService extends PuterService { #handleUploadProgress(data: UploadProgressPayload): void { const meta = data.meta ?? {}; const userId = (meta.user_id ?? meta.userId) as - number | string | undefined; + | number + | string + | undefined; if (!userId) { console.warn('[socket] upload-progress missing user_id', { meta }); return; diff --git a/src/backend/stores/session/SessionStore.js b/src/backend/stores/session/SessionStore.js index 9f61d4f72..4f9da1ddd 100644 --- a/src/backend/stores/session/SessionStore.js +++ b/src/backend/stores/session/SessionStore.js @@ -346,9 +346,14 @@ export class SessionStore extends PuterStore { * Soft-revoke a root session and every derived session that points back to * it via `parent_session_id`. Broadcasts cache invalidation for each * affected row's uuid + composite keys. + * + * Returns which rows were revoked — `{ userId, uuids }`, or null when there + * was nothing active to revoke. Callers use it to tear down anything else + * holding the session open; the rows are read here anyway, so reporting + * them costs no extra query. */ async revokeCascade(rootUuid) { - if (!rootUuid) return; + if (!rootUuid) return null; // Read each affected row's identity columns up-front — every // composite cache mapping (app, legacy-token, legacy-web) must @@ -358,7 +363,7 @@ export class SessionStore extends PuterStore { 'SELECT `uuid`, `user_id`, `kind`, `app_uid`, `legacy_token_uid`, `meta`, `created_via`, `last_ip`, `last_user_agent` FROM `sessions` WHERE (`uuid` = ? OR `parent_session_id` = ?) AND `revoked_at` IS NULL', [rootUuid, rootUuid], ); - if (rows.length === 0) return; + if (rows.length === 0) return null; // Double-delete: see `removeByUuid` for rationale. const keys = []; @@ -372,6 +377,11 @@ export class SessionStore extends PuterStore { ); await this.publishCacheKeys({ keys, broadcast: true }); + + return { + userId: Number(rows[0].user_id), + uuids: rows.map((r) => String(r.uuid)), + }; } /** diff --git a/src/backend/stores/session/SessionStore.test.ts b/src/backend/stores/session/SessionStore.test.ts index f1563ceee..a7fd25849 100644 --- a/src/backend/stores/session/SessionStore.test.ts +++ b/src/backend/stores/session/SessionStore.test.ts @@ -233,7 +233,14 @@ describe('SessionStore', () => { parent_session_id: parent.uuid, }); - await target.revokeCascade(parent.uuid); + const revoked = await target.revokeCascade(parent.uuid); + + // Reported back so callers can tear down anything else holding + // one of these sessions open. + expect(revoked.userId).toBe(user.id); + expect(new Set(revoked.uuids)).toEqual( + new Set([parent.uuid, child1.uuid, child2.uuid]), + ); expect(await target.getByUuid(parent.uuid)).toBeNull(); expect(await target.getByUuid(child1.uuid)).toBeNull(); @@ -262,14 +269,12 @@ describe('SessionStore', () => { it('is a no-op when the root uuid does not exist', async () => { await expect( target.revokeCascade('nonexistent-uuid'), - ).resolves.toBeUndefined(); + ).resolves.toBeNull(); }); it('is a no-op when called with a falsy uuid', async () => { - await expect(target.revokeCascade('')).resolves.toBeUndefined(); - await expect( - target.revokeCascade(undefined), - ).resolves.toBeUndefined(); + await expect(target.revokeCascade('')).resolves.toBeNull(); + await expect(target.revokeCascade(undefined)).resolves.toBeNull(); }); });