diff --git a/CHANGELOG.md b/CHANGELOG.md index e2acd4eb2..92dc5189a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Restoring a data archive into PostgreSQL from a gateway that does not run in UTC no longer shifts every timestamp by the host offset, and no longer shifts it again on each further restore ([#1624](https://github.com/rmyndharis/OpenWA/issues/1624)). - Retention sweeps on PostgreSQL delete the rows their window names instead of taking up to the host's UTC offset of younger rows with them, and the `today` message counts cover the host's local day. - Session leases on PostgreSQL compare as instants across nodes in different time zones and across a daylight-saving change. +- Live WebSocket sockets are re-validated against the API-key table once a minute, so a key deleted, revoked, expired or narrowed on another node or by a direct database write drops its sockets there too, and a socket that connected while its key was being revoked no longer keeps that authorization for the life of the connection ([#1625](https://github.com/rmyndharis/OpenWA/issues/1625)). +- A WebSocket subscribe whose socket is evicted while it is in flight no longer registers its rooms after the disconnect. ### Upgrade notes (behavior changes) @@ -25,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A table that has had an archive from a **SQLite** gateway restored into it holds those rows in correct UTC while the app wrote its own in local time. The two are indistinguishable within the column, so a blanket `UPDATE` would move the rows that are already right; correct such a column row by row against a known archive, or leave it as it is. - An archive a **PostgreSQL** gateway outside UTC exported before this release is shifted the other way, and restoring it moved the whole table, `DEFAULT now()` columns included: the export read every value back through the local-time parser and the import bound the resulting ISO text into a column that drops the zone, so each restore took the table one offset backward. No row in such a table is correct, and the conversion above would move those rows a further offset the wrong way. Where the table holds nothing but restored rows, apply the inverse once per restore taken: `UPDATE sessions SET "createdAt" = ("createdAt" AT TIME ZONE 'UTC') AT TIME ZONE 'Asia/Jakarta';`. Where it also holds rows written after the restore, no blanket conversion is safe. - Re-export after upgrading for an archive whose stamps are the instants they claim; restoring a pre-release archive carries its shift in as it is. +- A WebSocket client can now be disconnected with an `UNAUTHORIZED` frame up to a minute after its key changed, where before only the node that processed the change disconnected it; reconnect and resubscribe on that frame. A rename, and the key's usage counters, evict nobody. ## [0.23.5] - 2026-09-15 diff --git a/docs/06-api-specification.md b/docs/06-api-specification.md index 042e57eca..bf1355d98 100644 --- a/docs/06-api-specification.md +++ b/docs/06-api-specification.md @@ -6785,7 +6785,8 @@ A subscribe request whose `events` array contains no recognized name (after filt - **`sessionId: "*"`** subscribes to every session; **`events: ["*"]`** subscribes to every subscribable event. They combine (e.g. `"*"` + `["*"]` = every event of every session). - The API key is **re-validated on every `subscribe`** (not just at connect), so a key revoked or expired mid-connection is caught — the server replies `UNAUTHORIZED` and disconnects. - **Per-key session scope is enforced** against the fresh key: a key restricted via `allowedSessions` may NOT subscribe to `"*"` and may NOT subscribe to a session outside its allowlist — either is rejected with `FORBIDDEN_SESSION`. An unrestricted key (no `allowedSessions`) may subscribe to anything, including `"*"`. -- **`session.qr` requires the OPERATOR role**, matching `GET /api/sessions/{sessionId}/qr`. A VIEWER key may still subscribe to it, by name or through a wildcard, but the QR is never delivered to its sockets; every other event is. The role is read from the key re-validated on each `subscribe`, so a key narrowed to VIEWER stops receiving the QR from its next `subscribe` (a role change also disconnects the key's sockets). +- **Live sockets are re-validated against the database once a minute**, with no client activity required. A socket carries the key as it stood when it connected, and rooms joined earlier are never revisited, so that snapshot is what the sweep compares against the current row, along with any later `subscribe` whose key no longer matched it (what that `subscribe` granted outlives the change, so putting the row back does not spare the socket). It closes the key's sockets with an `UNAUTHORIZED` frame naming the cause: `API key has been deleted`, `API key has been revoked`, `API key has expired`, or `API key authorization changed; please reconnect` when `role`, `allowedIps`, `allowedSessions` or `expiresAt` moved. A change made through this API still evicts synchronously in the same request; the sweep is what catches a change made on another node, written straight to the database, or committed in the instant a socket was connecting. A rename, and the usage counters the gateway itself writes, evict nobody. Treat these frames as "reconnect and resubscribe", not as fatal. +- **`session.qr` requires the OPERATOR role**, matching `GET /api/sessions/{sessionId}/qr`. A VIEWER key may still subscribe to it, by name or through a wildcard, but the QR is never delivered to its sockets; every other event is. The role is read from the key re-validated on each `subscribe`, so a key narrowed to VIEWER stops receiving the QR from its next `subscribe`. Between subscribes the QR gate rests on the same snapshot as every other event: a key demoted right after a `subscribe` keeps receiving the QR until its sockets are evicted, which is immediate on the node processing the change and within the sweep's minute anywhere else. ### Example (socket.io-client) diff --git a/docs/10-devops-infrastructure.md b/docs/10-devops-infrastructure.md index 7890468b9..97bfaf166 100644 --- a/docs/10-devops-infrastructure.md +++ b/docs/10-devops-infrastructure.md @@ -270,9 +270,9 @@ volumes: > **Keep `replicas: 1`.** Session ownership gained claim/lease fencing (`nodeId` owner + > `leaseExpiresAt`), which bounds any two-engine overlap on one session to roughly one heartbeat > interval instead of eliminating it — and docs/13 still says DO NOT run its multi-replica examples -> yet: process-local key eviction, WebSocket rate-limit buckets, the unfenced liveness watchdog, -> bulk-batch state and MCP locality all remain per-process. Follow -> [13 - Horizontal Scaling Guide](./13-horizontal-scaling.md) for the full list and the design +> yet: WebSocket key eviction on a peer node lags by up to a minute, and WebSocket rate-limit +> buckets, the unfenced liveness watchdog, bulk-batch state and MCP locality all remain per-process. +> Follow [13 - Horizontal Scaling Guide](./13-horizontal-scaling.md) for the full list and the design > sketch. What multi-node eventually buys is engine capacity, not shared engine state: live engine > handles live in exactly one process's `EngineRegistry` (`src/engine/engine-registry.service.ts`), > and the hard requirements include a stable `NODE_ID` across restarts, NTP-synced clocks (lease skew diff --git a/docs/13-horizontal-scaling.md b/docs/13-horizontal-scaling.md index 50bafe9b7..55c597f94 100644 --- a/docs/13-horizontal-scaling.md +++ b/docs/13-horizontal-scaling.md @@ -99,14 +99,25 @@ > the same flag the throttler and cache already use). The gateway broadcasts to rooms; a Redis > pub/sub adapter attached to Socket.IO relays those broadcasts to every replica, so a client > connected to node A receives an event raised on node B. Scope honestly: this distributes event -> **fan-out only**. Mid-connection key eviction (`socketsByKeyId`) is still process-local — a key -> revoked on node A tears down only A's sockets — as are the per-key WS rate-limit buckets (counted -> per replica) and the engine registry. Without `REDIS_ENABLED` the adapter is inert and delivery -> is single-node, exactly as before. +> **fan-out only**. The per-key WS rate-limit buckets (counted per replica) and the engine registry +> are still process-local. Without `REDIS_ENABLED` the adapter is inert and delivery is single-node, +> exactly as before. > -> **What does not exist yet, and is why one replica is still the answer.** The cross-replica gaps -> just named (key eviction, WS rate-limit state) remain process-local. Not every lifecycle path is -> fenced: the liveness watchdog and reconnect timers still act on whatever is in the local +> **Mid-connection key eviction converges on a timer, not a broadcast.** The node that processes a +> revoke, delete, or narrowing tears down that key's sockets synchronously, in the same request. +> Nothing is published to peers; instead every node re-validates the keys behind its own live +> sockets against the database once a minute (`EventsGateway.sweepApiKeyAuthorization`, one batched +> read of the key ids currently holding sockets) and evicts on a row that is gone, inactive, expired, +> or whose role, `allowedIps`, `allowedSessions` or expiry no longer matches the snapshot the socket +> authenticated with. So a peer node's sockets close within a minute of the change. Before, a revoke, +> delete or expiry there waited for the client's next subscribe, and a narrowing was never caught at +> all: it leaves the key valid, so only the new subscribe is rejected while every room joined earlier +> stays joined. That minute is the current worst case for a socket streaming events its key has just +> lost; a key's REST calls are rejected immediately everywhere, since REST reads the row per request. +> +> **What does not exist yet, and is why one replica is still the answer.** The cross-replica gap just +> named (WS rate-limit state) remains process-local. Not every lifecycle path is fenced: the +> liveness watchdog and reconnect timers still act on whatever is in the local > registry. `BulkMessageService` keeps its live batch state in process, so a takeover cannot resume > a batch — only fail it. MCP/agent tool invocations execute on the node that received them rather > than being forwarded. diff --git a/src/modules/auth/api-key-authorization.ts b/src/modules/auth/api-key-authorization.ts new file mode 100644 index 000000000..bee3dfffe --- /dev/null +++ b/src/modules/auth/api-key-authorization.ts @@ -0,0 +1,48 @@ +import type { ApiKey } from './entities/api-key.entity'; + +/** + * Collapse an `allowedSessions` list to the two shapes the enforcement sites actually distinguish. + * + * The column is `simple-array`: TypeORM joins on write and splits on read, so `['']` is stored as + * `''` and read back as `[]`. Every site treats a zero-length list as "every session", so a write + * that looked like a scoping landed as a widening. The DTO validator now refuses such an entry at + * the boundary, and this is the second half: whatever reaches storage is either a non-empty list of + * real ids, or NULL. + * + * NULL rather than `[]` on purpose. Both already exist in the table for the same intent, and the + * published contract says an unscoped key omits the field, which only NULL produces. + */ +export function normalizeScopeList(list: string[] | null | undefined): string[] | null { + if (list == null) return null; + const cleaned = list.map(entry => entry.trim()).filter(entry => entry.length > 0); + return cleaned.length > 0 ? cleaned : null; +} + +/** An `expiresAt` in milliseconds, or null when it is unset or unparseable (a Date from the driver, a string from a snapshot). */ +export function apiKeyExpiryTime(value: Date | string | null | undefined): number | null { + if (!value) return null; + const time = value instanceof Date ? value.getTime() : new Date(value).getTime(); + return Number.isFinite(time) ? time : null; +} + +/** The columns that decide what a key may do. Everything else on the row is descriptive or advisory. */ +export type ApiKeyAuthorization = Pick; + +/** + * A stable string for what a key is AUTHORIZED to do: role, IP allowlist, session scope, expiry. + * Two rows sharing a fingerprint authorize identically, so any other column moving must leave it + * untouched, in particular `lastUsedAt`/`usageCount`/`updatedAt`, which the usage tracker rewrites + * for every key in active use. A fingerprint that moved with them would disconnect every live + * WebSocket client on the next windowed statistics write. + * + * Membership, not order: both allowlists are enforced with `.includes()`, so a reorder authorizes + * exactly the same and is sorted away here. `''`, `[]` and NULL all mean "unscoped" at every + * enforcement site and normalize to the same value, so a legacy row does not read as a change. + */ +export function apiKeyAuthorizationFingerprint(key: ApiKeyAuthorization): string { + const scope = (list: string[] | null | undefined): string[] | null => { + const normalized = normalizeScopeList(list); + return normalized ? [...normalized].sort() : null; + }; + return JSON.stringify([key.role, scope(key.allowedIps), scope(key.allowedSessions), apiKeyExpiryTime(key.expiresAt)]); +} diff --git a/src/modules/auth/auth.service.ts b/src/modules/auth/auth.service.ts index 66e3ce058..afa1bf55a 100644 --- a/src/modules/auth/auth.service.ts +++ b/src/modules/auth/auth.service.ts @@ -8,7 +8,7 @@ import { } from '@nestjs/common'; import { ModuleRef } from '@nestjs/core'; import { InjectRepository } from '@nestjs/typeorm'; -import { Repository, UpdateQueryBuilder, DeleteQueryBuilder, type QueryDeepPartialEntity } from 'typeorm'; +import { In, Repository, UpdateQueryBuilder, DeleteQueryBuilder, type QueryDeepPartialEntity } from 'typeorm'; import { randomBytes } from 'crypto'; import { ipMatches } from '../../common/utils/ip'; import { hashApiKey } from './api-key-hash'; @@ -17,6 +17,7 @@ import { CreateApiKeyDto, UpdateApiKeyDto } from './dto'; import { createLogger } from '../../common/services/logger.service'; import { readBootstrapKey, removeBootstrapKey, writeBootstrapKey } from './bootstrap-key-file'; import { ApiKeyUsageTracker } from './api-key-usage-tracker.service'; +import { apiKeyAuthorizationFingerprint, normalizeScopeList } from './api-key-authorization'; import { EventsGateway, type ApiKeyEvictionReason } from '../events/events.gateway'; /** @@ -49,24 +50,6 @@ export function bannerKeyLine(displayKey: string, isNewKey: boolean): string { return `${displayKey.slice(0, 8)}… (full key in data/.api-key or the dashboard)`; } -/** - * Collapse an `allowedSessions` list to the two shapes the enforcement sites actually distinguish. - * - * The column is `simple-array`: TypeORM joins on write and splits on read, so `['']` is stored as - * `''` and read back as `[]`. Every site treats a zero-length list as "every session", so a write - * that looked like a scoping landed as a widening. The DTO validator now refuses such an entry at - * the boundary, and this is the second half: whatever reaches storage is either a non-empty list of - * real ids, or NULL. - * - * NULL rather than `[]` on purpose. Both already exist in the table for the same intent, and the - * published contract says an unscoped key omits the field, which only NULL produces. - */ -function normalizeScopeList(list: string[] | null | undefined): string[] | null { - if (list == null) return null; - const cleaned = list.map(entry => entry.trim()).filter(entry => entry.length > 0); - return cleaned.length > 0 ? cleaned : null; -} - @Injectable() export class AuthService implements OnModuleInit, OnModuleDestroy { private readonly logger = createLogger('AuthService'); @@ -276,21 +259,10 @@ export class AuthService implements OnModuleInit, OnModuleDestroy { saved = await this.findOne(id); } - // Compare membership, not order: a pure reorder of allowedIps/allowedSessions is a no-op for the - // .includes()-based enforcement, so sort before stringify to avoid a spurious eviction on a reorder. - // Normalize before comparing: a legacy row stored as '' reads back as [], which means the same - // as NULL at every enforcement site, so treating them as different would evict live sockets for a - // write that changed nothing. - const ordered = (v: string[] | null) => { - const normalized = normalizeScopeList(v); - return normalized ? [...normalized].sort() : null; - }; - const authzChanged = - saved.role !== before.role || - saved.expiresAt?.getTime() !== before.expiresAt?.getTime() || - JSON.stringify(ordered(saved.allowedIps)) !== JSON.stringify(ordered(before.allowedIps)) || - JSON.stringify(ordered(saved.allowedSessions)) !== JSON.stringify(ordered(before.allowedSessions)); - if (authzChanged) { + // One fingerprint definition, two callers: this immediate eviction and the gateway's periodic + // re-validation sweep. Sharing it keeps the two from disagreeing about what an authorization + // change is (membership over order, '' and NULL alike, usage statistics ignored). + if (apiKeyAuthorizationFingerprint(saved) !== apiKeyAuthorizationFingerprint(before)) { this.evictActiveSockets(id, 'authorization_changed'); } return saved; @@ -447,6 +419,17 @@ export class AuthService implements OnModuleInit, OnModuleDestroy { } } + /** + * The current rows for a set of key ids, in one statement. Feeds the WebSocket gateway's periodic + * re-validation of the keys behind its live sockets: an id whose row is gone simply comes back + * absent, which the gateway reads as deleted. Usage statistics are deliberately not recorded here, + * so a passive socket does not look like traffic. + */ + async findAuthorizationStates(ids: string[]): Promise { + if (ids.length === 0) return []; + return this.apiKeyRepository.findBy({ id: In(ids) }); + } + async validateApiKey(rawKey: string, clientIp?: string, sessionId?: string): Promise { // Trim before hashing so every surface agrees on what the credential is. HTTP already strips // surrounding whitespace from header values, so a pasted key with a stray space/newline diff --git a/src/modules/events/events.gateway.authz-sweep.spec.ts b/src/modules/events/events.gateway.authz-sweep.spec.ts new file mode 100644 index 000000000..8201a8f7d --- /dev/null +++ b/src/modules/events/events.gateway.authz-sweep.spec.ts @@ -0,0 +1,299 @@ +import 'reflect-metadata'; +import { DataSource, Repository } from 'typeorm'; +import type { ModuleRef } from '@nestjs/core'; +import type { ConfigService } from '@nestjs/config'; +import type { Socket } from 'socket.io'; +import { ApiKey, ApiKeyRole } from '../auth/entities/api-key.entity'; +import { AuthService } from '../auth/auth.service'; +import { ApiKeyUsageTracker } from '../auth/api-key-usage-tracker.service'; +import { AuditService } from '../audit/audit.service'; +import { EventsGateway } from './events.gateway'; +import type { WSErrorResponse, WSSubscribedResponse } from './dto/ws-messages.dto'; + +/** + * The gateway's periodic re-validation of the keys behind its live sockets, against a REAL + * better-sqlite3 api_keys table and the real AuthService/ApiKeyUsageTracker. + * + * A stub cannot carry this: the sweep's whole risk is a fingerprint that moves for a column the + * authentication hot path rewrites on its own (lastUsedAt, usageCount, updatedAt), which would + * disconnect every live client once a minute. Only a real table, written by the real tracker, + * proves it does not. The changes that never reach this process, a key deleted, revoked, expired or + * narrowed by another node or a direct write, are expressed the same way: straight to the table. + */ +describe('EventsGateway API-key authorization sweep', () => { + const CLIENT_IP = '203.0.113.5'; + + let ds: DataSource; + let repo: Repository; + let service: AuthService; + let gateway: EventsGateway; + + interface MockSocket { + id: string; + handshake: { + headers: Record; + query: Record; + auth: { apiKey?: string }; + address: string; + }; + data: Record; + emit: jest.Mock; + disconnect: jest.Mock; + join: jest.Mock; + leave: jest.Mock; + rooms: Set; + disconnected: boolean; + } + + const makeSocket = (apiKey: string, id = 'sock-1'): MockSocket => { + const sock: MockSocket = { + id, + handshake: { headers: {}, query: {}, auth: { apiKey }, address: CLIENT_IP }, + data: {}, + emit: jest.fn(), + // Socket.IO flips `disconnected` synchronously on disconnect(); the gateway reads it. + disconnect: jest.fn(() => { + sock.disconnected = true; + }), + join: jest.fn(), + leave: jest.fn(), + rooms: new Set(), + disconnected: false, + }; + return sock; + }; + + const asSocket = (s: MockSocket): Socket => s as unknown as Socket; + + const connect = async (rawKey: string, id = 'sock-1'): Promise => { + const sock = makeSocket(rawKey, id); + await gateway.handleConnection(asSocket(sock)); + expect(sock.disconnect).not.toHaveBeenCalled(); + return sock; + }; + + const subscribe = async (sock: MockSocket, sessionId: string, events: string[], requestId = 'r1') => + (await gateway.handleMessage(asSocket(sock), { + type: 'subscribe', + sessionId, + events, + requestId, + })) as WSSubscribedResponse | WSErrorResponse; + + const sweep = (now?: number): Promise => + (gateway as unknown as { sweepApiKeyAuthorization: (now?: number) => Promise }).sweepApiKeyAuthorization(now); + + const evictionMessage = (sock: MockSocket): string | undefined => + sock.emit.mock.calls.map(([, frame]) => frame as WSErrorResponse).find(frame => frame?.code === 'UNAUTHORIZED') + ?.message; + + beforeAll(async () => { + ds = new DataSource({ type: 'better-sqlite3', database: ':memory:', entities: [ApiKey], synchronize: true }); + await ds.initialize(); + repo = ds.getRepository(ApiKey); + }); + + afterAll(async () => { + await ds.destroy(); + }); + + beforeEach(async () => { + await repo.clear(); + const moduleRef = { get: () => gateway } as unknown as ModuleRef; + service = new AuthService(repo, new ApiKeyUsageTracker(repo), moduleRef); + gateway = new EventsGateway( + service, + { logWarn: jest.fn().mockResolvedValue(null) } as unknown as AuditService, + { get: (_key: string, fallback?: unknown) => fallback } as unknown as ConfigService, + ); + }); + + it('evicts nobody when only the usage statistics moved', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'usage probe', role: ApiKeyRole.VIEWER }); + const sock = await connect(rawKey); + + // What the authentication hot path writes on its own, for every key in use. + await repo.update({ id: apiKey.id }, { lastUsedAt: new Date(Date.now() + 5_000), usageCount: 99 }); + await service.validateApiKey(rawKey, CLIENT_IP); + + await sweep(); + + expect(sock.disconnect).not.toHaveBeenCalled(); + }); + + it('evicts nobody for a rename or a reordered allowlist', async () => { + const { apiKey, rawKey } = await service.createApiKey({ + name: 'scoped key', + allowedSessions: ['sess-b', 'sess-a'], + allowedIps: [CLIENT_IP, '198.51.100.9'], + }); + const sock = await connect(rawKey); + + await service.update(apiKey.id, { + name: 'renamed key', + allowedSessions: ['sess-a', 'sess-b'], + allowedIps: ['198.51.100.9', CLIENT_IP], + }); + + expect(sock.disconnect).not.toHaveBeenCalled(); + await sweep(); + expect(sock.disconnect).not.toHaveBeenCalled(); + }); + + it('evicts with the deleted reason when the row is gone', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'doomed key' }); + const sock = await connect(rawKey); + + await repo.delete({ id: apiKey.id }); + await sweep(); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key has been deleted'); + }); + + it('evicts with the revoked reason when the row went inactive elsewhere', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'revoked key' }); + const sock = await connect(rawKey); + + await repo.update({ id: apiKey.id }, { isActive: false }); + await sweep(); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key has been revoked'); + }); + + it('evicts with the expired reason once the stored expiry has passed', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'expiring key' }); + const sock = await connect(rawKey); + + // An expiry change also moves the authorization fingerprint; the client must still be told the + // reason that actually applies. + await repo.update({ id: apiKey.id }, { expiresAt: new Date(Date.now() - 1_000) }); + await sweep(); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key has expired'); + }); + + it('keeps a socket whose key has no expiry or a future one', async () => { + const open = await connect((await service.createApiKey({ name: 'no expiry' })).rawKey, 'sock-1'); + const future = await service.createApiKey({ + name: 'future expiry', + expiresAt: new Date(Date.now() + 3_600_000).toISOString(), + }); + const later = await connect(future.rawKey, 'sock-2'); + + await sweep(); + + expect(open.disconnect).not.toHaveBeenCalled(); + expect(later.disconnect).not.toHaveBeenCalled(); + }); + + it('evicts on the snapshot expiry when the api_keys table cannot be read', async () => { + const { rawKey } = await service.createApiKey({ + name: 'expiring key', + expiresAt: new Date(Date.now() + 3_600_000).toISOString(), + }); + const sock = await connect(rawKey); + jest.spyOn(service, 'findAuthorizationStates').mockRejectedValue(new Error('database is locked')); + + await sweep(Date.now() + 7_200_000); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key has expired'); + }); + + it('evicts a socket that subscribed under a widening the row no longer carries', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'scoped key', allowedSessions: ['sess-a'] }); + const sock = await connect(rawKey); + + // Unscoped by a write this process never saw, subscribed to every session under it, then put + // back. The snapshot matches the row again, but the wildcard rooms the widening granted are + // never revisited, so the socket keeps receiving every session's events unless it is evicted. + await repo.update({ id: apiKey.id }, { allowedSessions: null }); + expect((await subscribe(sock, '*', ['*'])).type).toBe('subscribed'); + await repo.update({ id: apiKey.id }, { allowedSessions: ['sess-a'] }); + + await sweep(); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key authorization changed; please reconnect'); + }); + + it('evicts when the key was narrowed by a write this process never saw', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'narrowed key' }); + const sock = await connect(rawKey); + + await repo.update({ id: apiKey.id }, { allowedSessions: ['sess-1'] }); + await sweep(); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key authorization changed; please reconnect'); + }); + + it('evicts a socket that connected while its key was being revoked', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'racing key' }); + // The revoke commits (and evicts, finding nothing) after this socket validated and before it is + // registered: without the sweep its snapshot stays authoritative for the life of the connection. + const validate = jest.spyOn(service, 'validateApiKey').mockImplementationOnce(async (raw, ip) => { + const key = await AuthService.prototype.validateApiKey.call(service, raw, ip); + await repo.update({ id: apiKey.id }, { isActive: false }); + gateway.evictApiKey(apiKey.id, 'revoked'); + return key; + }); + const sock = await connect(rawKey); + validate.mockRestore(); + + await sweep(); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key has been revoked'); + }); + + it('does not refresh the connect-time snapshot on subscribe, so a narrowed key is still evicted', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'wildcard key' }); + const sock = await connect(rawKey); + await subscribe(sock, '*', ['*']); + + // Narrowed elsewhere. The wildcard rooms joined above are never revisited, so the socket keeps + // them until it is evicted; re-subscribing within the new scope must not launder its snapshot. + await repo.update({ id: apiKey.id }, { allowedSessions: ['sess-1'] }); + const res = (await subscribe(sock, 'sess-1', ['message.received'], 'r2')) as WSSubscribedResponse; + expect(res.type).toBe('subscribed'); + + await sweep(); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + }); + + it('joins no room when the socket is evicted while its subscribe is in flight', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'in flight key' }); + const sock = await connect(rawKey); + jest.spyOn(service, 'validateApiKey').mockImplementationOnce(async (raw, ip) => { + const key = await AuthService.prototype.validateApiKey.call(service, raw, ip); + gateway.evictApiKey(apiKey.id, 'revoked'); // an operator revoke landing on the same tick + return key; + }); + + const res = (await subscribe(sock, 'sess-1', ['message.received'])) as WSErrorResponse; + + expect(res.code).toBe('UNAUTHORIZED'); + expect(sock.join).not.toHaveBeenCalled(); + }); + + it('still evicts synchronously on an operator-driven change, before any sweep', async () => { + const { apiKey, rawKey } = await service.createApiKey({ name: 'demoted key' }); + const sock = await connect(rawKey); + + await service.update(apiKey.id, { role: ApiKeyRole.VIEWER }); + + expect(sock.disconnect).toHaveBeenCalledWith(true); + expect(evictionMessage(sock)).toBe('API key authorization changed; please reconnect'); + }); + + it('reads nothing when no socket is connected', async () => { + const read = jest.spyOn(service, 'findAuthorizationStates'); + await sweep(); + expect(read).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/events/events.gateway.spec.ts b/src/modules/events/events.gateway.spec.ts index 72247018d..38bb452db 100644 --- a/src/modules/events/events.gateway.spec.ts +++ b/src/modules/events/events.gateway.spec.ts @@ -332,48 +332,8 @@ describe('EventsGateway connection auth + subscribe re-validation', () => { expect(() => gateway.evictApiKey('k1')).not.toThrow(); }); - it('evicts a passive socket once its cached API key expires', async () => { - authService.validateApiKey.mockResolvedValue({ - id: 'k1', - name: 'k', - allowedSessions: null, - expiresAt: new Date('2026-01-01T00:00:00Z'), - }); - const sock = makeSocket({ apiKey: 'good' }); - await gateway.handleConnection(asSocket(sock)); - - (gateway as unknown as { sweepExpiredApiKeys: (now: number) => void }).sweepExpiredApiKeys( - Date.parse('2026-01-01T00:00:01Z'), - ); - - expect(sock.disconnect).toHaveBeenCalledWith(true); - expect(sock.emit).toHaveBeenCalledWith( - 'message', - expect.objectContaining({ code: 'UNAUTHORIZED', message: 'API key has expired' }), - ); - }); - - it('keeps sockets with no expiry or a future expiry', async () => { - const noExpiry = makeSocket({ apiKey: 'a' }); - const future = { ...makeSocket({ apiKey: 'b' }), id: 'sock-2' }; - authService.validateApiKey - .mockResolvedValueOnce({ id: 'k1', name: 'a', allowedSessions: null, expiresAt: null }) - .mockResolvedValueOnce({ - id: 'k2', - name: 'b', - allowedSessions: null, - expiresAt: new Date('2026-01-02T00:00:00Z'), - }); - await gateway.handleConnection(asSocket(noExpiry)); - await gateway.handleConnection(asSocket(future)); - - (gateway as unknown as { sweepExpiredApiKeys: (now: number) => void }).sweepExpiredApiKeys( - Date.parse('2026-01-01T00:00:00Z'), - ); - - expect(noExpiry.disconnect).not.toHaveBeenCalled(); - expect(future.disconnect).not.toHaveBeenCalled(); - }); + // The periodic sweep (expiry, revocation, deletion, narrowing) reads the api_keys table, so it is + // exercised against a real one in events.gateway.authz-sweep.spec.ts rather than a stub here. }); }); diff --git a/src/modules/events/events.gateway.ts b/src/modules/events/events.gateway.ts index 2cf7667a4..885713a34 100644 --- a/src/modules/events/events.gateway.ts +++ b/src/modules/events/events.gateway.ts @@ -18,6 +18,7 @@ import { resolveCorsPolicy } from '../../config/bootstrap-security'; import { resolveClientIp as resolveRequestClientIp, type RequestLike } from '../../common/utils/ip'; import { DEFAULT_WEBHOOK_MEDIA_INLINE_MAX_BYTES, shedInlineMedia } from '../../common/utils/inline-media'; import { ApiKeyRole, type ApiKey } from '../auth/entities/api-key.entity'; +import { apiKeyAuthorizationFingerprint, apiKeyExpiryTime } from '../auth/api-key-authorization'; import { readWsRateLimitConfig, TokenBucketLimiter, @@ -115,7 +116,7 @@ export class EventsGateway implements OnGatewayInit, OnGatewayConnection, OnGate * an already-subscribed socket keeps receiving events until it happens to disconnect). */ private readonly socketsByKeyId = new Map>(); - private expirySweepTimer?: ReturnType; + private authzSweepTimer?: ReturnType; /** * Rate limiting for the WS surface (see ws-rate-limit.ts). Frames never pass through the @@ -152,33 +153,101 @@ export class EventsGateway implements OnGatewayInit, OnGatewayConnection, OnGate afterInit() { this.logger.log('WebSocket Gateway initialized'); - this.expirySweepTimer = setInterval(() => { - try { - this.sweepExpiredApiKeys(); - } catch (error) { - this.logger.error('Failed to sweep expired WebSocket API keys', error instanceof Error ? error.stack : error); - } + this.authzSweepTimer = setInterval(() => { + void this.sweepApiKeyAuthorization().catch(error => + this.logger.error( + 'Failed to sweep WebSocket API key authorization', + error instanceof Error ? error.stack : error, + ), + ); }, 60_000); - this.expirySweepTimer.unref?.(); + this.authzSweepTimer.unref?.(); } onModuleDestroy(): void { - if (this.expirySweepTimer) clearInterval(this.expirySweepTimer); - this.expirySweepTimer = undefined; + if (this.authzSweepTimer) clearInterval(this.authzSweepTimer); + this.authzSweepTimer = undefined; } - private sweepExpiredApiKeys(now = Date.now()): void { + /** + * Re-validate the keys behind the live sockets against the database, once per tick. + * + * A socket carries the key snapshot taken at connect and never refreshes it, so every later change + * to the row is invisible to it: a key deleted, revoked, expired or narrowed by another node, by a + * direct database write, or by an operator change that committed in the window between this + * socket's validation and its registration here. The operator path still evicts synchronously in + * the same request (see AuthService.update/revoke/delete); this is the backstop for the changes + * that never reached this process. + * + * One batched read over the distinct key ids currently holding sockets, then eviction per key with + * the reason that actually applies. Only the authorization columns are compared (see + * apiKeyAuthorizationFingerprint), so the usage tracker's windowed lastUsedAt/usageCount write, + * which touches every key in use, evicts nobody. + */ + private async sweepApiKeyAuthorization(now = Date.now()): Promise { + // The expiry a socket already carries is decided first, and without the database: it needs no row + // to be read, and a table that is unreachable or locked must not keep an expired key streaming + // events until the first tick whose read succeeds. for (const [keyId, sockets] of Array.from(this.socketsByKeyId.entries())) { - const expired = Array.from(sockets).some(client => { - const expiresAt = (client.data as { apiKey?: Pick } | undefined)?.apiKey?.expiresAt; - if (!expiresAt) return false; - const expiry = expiresAt instanceof Date ? expiresAt.getTime() : new Date(expiresAt).getTime(); - return Number.isFinite(expiry) && expiry <= now; - }); - if (expired) this.evictApiKey(keyId, 'expired'); + if (Array.from(sockets).some(client => this.isSnapshotExpired(client, now))) { + this.evictApiKey(keyId, 'expired'); + } + } + const keyIds = Array.from(this.socketsByKeyId.keys()); + if (keyIds.length === 0) return; + const current = await this.authService.findAuthorizationStates(keyIds); + const byId = new Map(current.map(key => [key.id, key])); + for (const keyId of keyIds) { + const reason = this.evictionReason(byId.get(keyId), this.socketsByKeyId.get(keyId), now); + if (reason) this.evictApiKey(keyId, reason); } } + /** + * Why a key's sockets must go, or null to keep them. `current` is the row as it stands now, absent + * when the key was deleted. Order matters: the reason a client is told should be the strongest one + * that applies, not merely the first field that differs from its snapshot. + */ + private evictionReason( + current: ApiKey | undefined, + sockets: Set | undefined, + now: number, + ): ApiKeyEvictionReason | null { + if (!sockets || sockets.size === 0) return null; + if (!current) return 'deleted'; + if (!current.isActive) return 'revoked'; + const expiry = apiKeyExpiryTime(current.expiresAt); + if (expiry !== null && expiry <= now) return 'expired'; + const authorization = apiKeyAuthorizationFingerprint(current); + // Per socket, not per key: sockets under one key connected at different moments, so one can hold + // a stale snapshot while another already carries the new authorization. A socket that subscribed + // under something other than its snapshot goes too, even when the row matches that snapshot + // again: the rooms that subscribe granted are never revisited, so a widening reverted before this + // tick would otherwise leave them joined for the life of the connection. + const stale = Array.from(sockets).some( + client => this.snapshotFingerprint(client) !== authorization || this.hasDivergentGrant(client), + ); + return stale ? 'authorization_changed' : null; + } + + /** The authorization fingerprint of the key snapshot a socket has been carrying since connect. */ + private snapshotFingerprint(client: Socket): string { + const snapshot = (client.data as { apiKey?: ApiKey } | undefined)?.apiKey; + return snapshot ? apiKeyAuthorizationFingerprint(snapshot) : ''; + } + + /** Whether the key snapshot a socket carries has expired, decided from the socket alone. */ + private isSnapshotExpired(client: Socket, now: number): boolean { + const snapshot = (client.data as { apiKey?: Pick } | undefined)?.apiKey; + const expiry = apiKeyExpiryTime(snapshot?.expiresAt); + return expiry !== null && expiry <= now; + } + + /** Whether a subscribe ever granted this socket something under a key other than its snapshot. */ + private hasDivergentGrant(client: Socket): boolean { + return (client.data as { authorizationDiverged?: boolean } | undefined)?.authorizationDiverged === true; + } + /** * Resolve the trusted-proxy-aware client IP for a socket, reusing the same shared * `resolveClientIp` helper as the REST guard and MCP mount. X-Forwarded-For is only @@ -217,10 +286,11 @@ export class EventsGateway implements OnGatewayInit, OnGatewayConnection, OnGate /** * Tear down every active socket authenticated with `keyId`. Called by AuthService when a key is - * revoked, deleted, or has its authorization (role/allowedSessions/allowedIps/expiry) narrowed, so - * the key's already-subscribed sockets stop receiving events immediately instead of lingering until - * they disconnect on their own. Each socket gets a clean close (an `UNAUTHORIZED` reason) reflecting - * the actual trigger, rather than a silent drop. + * revoked, deleted, or has its authorization (role/allowedSessions/allowedIps/expiry) narrowed, and + * by sweepApiKeyAuthorization for the same changes when they only reach this process through the + * database, so the key's already-subscribed sockets stop receiving events immediately instead of + * lingering until they disconnect on their own. Each socket gets a clean close (an `UNAUTHORIZED` + * reason) reflecting the actual trigger, rather than a silent drop. */ evictApiKey(keyId: string, reason: ApiKeyEvictionReason = 'revoked'): void { const sockets = this.socketsByKeyId.get(keyId); @@ -379,7 +449,7 @@ export class EventsGateway implements OnGatewayInit, OnGatewayConnection, OnGate // here too, not just at connect. const rawApiKey = (client.data as { rawApiKey?: string }).rawApiKey; const clientIp = this.resolveClientIp(client); - let subscriberKey: { allowedSessions?: string[] | null; role?: ApiKeyRole } | null; + let subscriberKey: ApiKey | null; try { subscriberKey = rawApiKey ? await this.authService.validateApiKey(rawApiKey, clientIp) : null; } catch { @@ -390,6 +460,22 @@ export class EventsGateway implements OnGatewayInit, OnGatewayConnection, OnGate client.disconnect(); return this.createError('UNAUTHORIZED', 'API key is no longer valid', requestId); } + + // The socket can have been evicted while this re-validation was in flight (a revoke landing on + // the same tick as a subscribe). Joining rooms now would register a disconnected socket in the + // adapter, where nothing prunes it again. + if (client.disconnected) { + return this.createError('UNAUTHORIZED', 'Connection is closed', requestId); + } + + // The fresh key decides THIS subscribe, and is deliberately not written back over the connect-time + // snapshot in client.data: rooms joined earlier are never revisited, so a socket that refreshed its + // snapshot here would look current to the sweep while still holding rooms its key has since lost. + // What it does record is that the two diverged, since everything granted below outlives the key + // state that granted it, and the row can be back to the snapshot by the time the sweep reads it. + if (apiKeyAuthorizationFingerprint(subscriberKey) !== this.snapshotFingerprint(client)) { + (client.data as { authorizationDiverged?: boolean }).authorizationDiverged = true; + } this.syncQrAccess(client, subscriberKey.role); // Enforce per-key session scope against the FRESH key: a key restricted to specific