diff --git a/package-lock.json b/package-lock.json index 836dc3a30..0c01a2da6 100644 --- a/package-lock.json +++ b/package-lock.json @@ -5400,6 +5400,25 @@ "url": "https://opencollective.com/express" } }, + "node_modules/@nestjs/platform-ws": { + "version": "11.1.29", + "resolved": "https://registry.npmjs.org/@nestjs/platform-ws/-/platform-ws-11.1.29.tgz", + "integrity": "sha512-AYs3VZ5PxcDLiyGfvxsAS8DP9d7q+jxCKoWfMdl9RQLz69WAzSX2x+FgWNupTHt19zf8rmFXt2hvGd2tBUROAQ==", + "license": "MIT", + "dependencies": { + "tslib": "2.8.1", + "ws": "8.21.1" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/nest" + }, + "peerDependencies": { + "@nestjs/common": "^11.0.0", + "@nestjs/websockets": "^11.0.0", + "rxjs": "^7.1.0" + } + }, "node_modules/@nestjs/schedule": { "version": "6.1.3", "resolved": "https://registry.npmjs.org/@nestjs/schedule/-/schedule-6.1.3.tgz", @@ -5474,6 +5493,29 @@ } } }, + "node_modules/@nestjs/websockets": { + "version": "11.1.29", + "resolved": "https://registry.npmjs.org/@nestjs/websockets/-/websockets-11.1.29.tgz", + "integrity": "sha512-Mg0sNK2k+PpamcJpDy9ZP24ZvM08xB6AZs061Kb6dhTesGxr6zrxlbZ7+40r5xwuxD/PN8LM2tSRKf5ojbFNDQ==", + "license": "MIT", + "dependencies": { + "iterare": "1.2.1", + "object-hash": "3.0.0", + "tslib": "2.8.1" + }, + "peerDependencies": { + "@nestjs/common": "^11.0.0", + "@nestjs/core": "^11.0.0", + "@nestjs/platform-socket.io": "^11.0.0", + "reflect-metadata": "^0.1.12 || ^0.2.0", + "rxjs": "^7.1.0" + }, + "peerDependenciesMeta": { + "@nestjs/platform-socket.io": { + "optional": true + } + } + }, "node_modules/@noble/hashes": { "version": "1.8.0", "resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-1.8.0.tgz", @@ -15362,7 +15404,6 @@ "version": "3.0.0", "resolved": "https://registry.npmjs.org/object-hash/-/object-hash-3.0.0.tgz", "integrity": "sha512-RSn9F68PjH9HqtltsSnqYC1XXoWe9Bju5+213R98cNGttag9q9yAOTzdbsqvIa7aNm5WffBZFpWYr2aWrklWAw==", - "dev": true, "license": "MIT", "engines": { "node": ">= 6" @@ -20934,9 +20975,9 @@ "license": "ISC" }, "node_modules/ws": { - "version": "8.21.0", - "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz", - "integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==", + "version": "8.21.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz", + "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==", "license": "MIT", "engines": { "node": ">=10.0.0" @@ -21151,8 +21192,10 @@ "@nestjs/config": "^4.0.4", "@nestjs/core": "^11.1.24", "@nestjs/platform-express": "^11.1.24", + "@nestjs/platform-ws": "^11.1.29", "@nestjs/schedule": "^6.1.3", "@nestjs/swagger": "^11.4.5", + "@nestjs/websockets": "^11.1.29", "@simplewebauthn/server": "^13.1.2", "@trek/nest-mcp": "*", "@trek/shared": "*", diff --git a/server/package.json b/server/package.json index b1e3a6ff4..10a5f58d1 100644 --- a/server/package.json +++ b/server/package.json @@ -30,8 +30,10 @@ "@nestjs/config": "^4.0.4", "@nestjs/core": "^11.1.24", "@nestjs/platform-express": "^11.1.24", + "@nestjs/platform-ws": "^11.1.29", "@nestjs/schedule": "^6.1.3", "@nestjs/swagger": "^11.4.5", + "@nestjs/websockets": "^11.1.29", "@simplewebauthn/server": "^13.1.2", "@trek/nest-mcp": "*", "@trek/shared": "*", diff --git a/server/src/bootstrap.ts b/server/src/bootstrap.ts index ebe7dd49e..b0c8269c0 100644 --- a/server/src/bootstrap.ts +++ b/server/src/bootstrap.ts @@ -1,3 +1,4 @@ +import nodeHttp from 'node:http'; import { NestFactory } from '@nestjs/core'; import { ExpressAdapter, type NestExpressApplication } from '@nestjs/platform-express'; import type { INestApplication } from '@nestjs/common'; @@ -12,6 +13,7 @@ import { McpRegistryService } from '@trek/nest-mcp'; import { setMcpRegistry } from './mcp/registry-handoff'; import { validateBodyContracts } from './nest/common/validate-body-contracts'; import { validateRouteGuards } from './nest/common/validate-route-guards'; +import { TrekWsAdapter } from './nest/realtime/trek-ws.adapter'; /** * Builds the unified TREK NestJS application that serves the ENTIRE surface — the @@ -43,12 +45,35 @@ import { validateRouteGuards } from './nest/common/validate-route-guards'; * SpaFallbackFilter (APP_FILTER in AppModule); the global error envelope is the * TrekExceptionFilter (also APP_FILTER). */ +let boundHttpServer: nodeHttp.Server | null = null; + +/** + * The http server buildApp created and bound the ws gateway to. + * + * index.ts listens on it; the websocket suites connect to it. Anyone creating a + * second server around the same express instance would get a working REST API + * and a /ws that no client can reach. + */ +export function getHttpServer(): nodeHttp.Server { + if (!boundHttpServer) throw new Error('buildApp() has not run yet'); + return boundHttpServer; +} + export async function buildApp(): Promise { // rawBody keeps the unparsed request bytes on req.rawBody so a plugin webhook // route can verify a provider's HMAC signature over the exact payload (the // parsed JSON alone can't be re-serialised byte-for-byte). const app = await NestFactory.create(AppModule, new ExpressAdapter(), { rawBody: true }); const instance = app.getHttpAdapter().getInstance(); + // The http server is created HERE, not by the caller after buildApp returns, + // and that ordering is the whole point: Nest binds gateways during app.init(), + // so the ws adapter has to exist before it and has to already know which + // server to attach to. Handing it the app instead would bind the ws server to + // Nest's own internal http server, which this process never listens on: the + // boot succeeds, the gateway logs as registered, every test passes, and no + // browser can connect. Callers take the server from getHttpServer() below. + boundHttpServer = nodeHttp.createServer(instance); + app.useWebSocketAdapter(new TrekWsAdapter(boundHttpServer)); // ConfigModule.forRoot's load factories already ran inside NestFactory.create, // so the boot-stable snapshot is resolvable here, BEFORE app.init() — this is // the one bridge that lets the pre-init Express layer consume the validated diff --git a/server/src/index.ts b/server/src/index.ts index 1f665a0bf..9504ed2c0 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -8,7 +8,7 @@ import path from 'node:path'; import fs from 'node:fs'; import http from 'node:http'; import type { INestApplication } from '@nestjs/common'; -import { buildApp } from './bootstrap'; +import { buildApp, getHttpServer } from './bootstrap'; // Create upload and data directories on startup. // Every uploads subdir the app writes to must be listed here (#1762): a dir @@ -86,9 +86,8 @@ const onListen = () => { if (env.demo.enabled && env.app.isProduction) { sLogWarn('SECURITY WARNING: DEMO_MODE is enabled in production!'); } - import('./websocket').then(({ setupWebSocket }) => { - setupWebSocket(server); - }); + // Nothing else to boot by hand: the crons are domain providers on the + // scheduling registrar, and /ws is the Nest gateway buildApp already bound. }; let server: http.Server; @@ -101,7 +100,10 @@ async function bootstrap(): Promise { // (/mcp, /.well-known, OAuth SDK, SPA catch-all). buildApp() owns the composition // order; it is shared with the integration-test harness so they can't drift. nestApp = await buildApp(); - server = http.createServer(nestApp.getHttpAdapter().getInstance()); + // The server buildApp created and bound /ws to. Creating a second one here + // would serve the REST API fine and leave the gateway attached to a socket + // nobody listens on. + server = getHttpServer(); if (HOST) server.listen(PORT, HOST, onListen); else server.listen(PORT, onListen); } diff --git a/server/src/nest/README.md b/server/src/nest/README.md index bc2a3464f..941603036 100644 --- a/server/src/nest/README.md +++ b/server/src/nest/README.md @@ -146,6 +146,44 @@ tools, which never pass through an HTTP guard. In the five domains piloted for files of import churn, and `scripts/coverage-thresholds.mjs` buckets on one path segment, so merging `photos/` into it would collapse two ratchet entries into one and drop the stricter of the two. +- `realtime/` — the /ws transport, as a Nest gateway. + - `realtime.gateway.ts` owns the connection lifecycle and the handshake, with + `DatabaseService` and `EphemeralTokenService` injected instead of imported. + It replaced `setupWebSocket(server)`, which index.ts kicked off with a + dynamic import after listen(). + - `ws-state.ts` holds the socket registry, and that is **module state on + purpose**. Seventeen `*.bridge.ts` files and `notifications.instance.ts` + build `new RealtimeService()` outside the container; rooms on a provider + instance would leave their broadcasts going to an empty map, with no error + and no log. Same reasoning as the geo throttle cursor. + - `trek-ws.adapter.ts` exists because the stock `WsAdapter` dispatches on + `{ event, data }` and every deployed client sends `{ type, tripId }`. It + also builds the ws server itself rather than delegating: the base's port-0 + branch uses `noServer: true` and performs the upgrade from its own registry, + which never consults `verifyClient`, so delegating would silently drop the + origin check. The per-socket `error` listener is attached FIRST, because an + unhandled one kills the process (#1576), and the flood guard counts every + inbound frame before parsing, because malformed frames are the ones worth + limiting. + - The socket id stays a **monotonic integer**. The client echoes it as + `X-Socket-Id` and `broadcast` excludes the originator with + `Number(excludeSid)`; a uuid is `NaN`, `NaN === NaN` is false, and every + client would receive its own writes back. It shows up as drag-and-drop that + jumps, not as an error. + - `RealtimeGatewayModule` is imported by AppModule alone. RealtimeModule is + @Global, so a gateway there follows into every hand-assembled TestingModule, + and Nest's SocketModule then loads an adapter none of them set: it reaches + for `@nestjs/platform-socket.io` and calls `process.exit(1)` when it is + absent. + - `buildApp()` creates the http server and binds the adapter to it BEFORE + `app.init()`, and callers take it from `getHttpServer()`. Nest binds gateways + during init, so an adapter registered later binds nothing, and one given the + app instead of the server binds to Nest's internal server, which this process + never listens on. Either way the boot is green and no browser can connect. + - `src/websocket.ts` survives as a re-export. 115 test files `vi.mock` that + path to assert on broadcasts; repointing them all would be a large diff that + proves nothing. + - `geo/` — the ONE Nominatim client (#576). There used to be two: atlas throttled to the published 1.1s and cached its answers, maps did neither, so five interactive routes fired straight at the service while the atlas loops diff --git a/server/src/nest/app.module.ts b/server/src/nest/app.module.ts index d27347edd..733965d4d 100644 --- a/server/src/nest/app.module.ts +++ b/server/src/nest/app.module.ts @@ -61,6 +61,7 @@ import { TrekExceptionFilter } from './common/trek-exception.filter'; import { SpaFallbackFilter } from './platform/spa-fallback.filter'; import { IdempotencyInterceptor } from './common/idempotency.interceptor'; import { IdempotencyCleanupJob } from './common/idempotency-cleanup.job'; +import { RealtimeGatewayModule } from './realtime/realtime-gateway.module'; /** * Root NestJS module for the incremental migration. Domain modules @@ -68,7 +69,7 @@ import { IdempotencyCleanupJob } from './common/idempotency-cleanup.job'; * migrated. */ @Module({ - imports: [AppConfigModule, DatabaseModule, RealtimeModule, SchedulingModule, McpModule.forRoot({ accessPolicy: trekMcpAccessPolicy, validateAccess: trekMcpValidateAccess }), HealthModule, WeatherModule, HelpModule, AirportsModule, ConfigModule, SystemNoticesModule, GeoModule, MapsModule, PlaceEnrichmentModule, CategoriesModule, TagsModule, NotificationsModule, AtlasModule, VacayModule, PackingModule, TodoModule, BudgetModule, ReservationsModule, DaysModule, DayNotesModule, AccommodationsModule, AssignmentsModule, PlacesModule, TripsModule, CollabModule, FilesModule, PhotosModule, MemoriesModule, AirtrailModule, JourneyModule, CollectionsModule, ShareModule, TripInviteModule, TransitModule, FeedsModule, SettingsModule, BackupModule, AuthModule, OidcModule, OauthModule, AdminModule, AddonsModule, AuditModule, PermissionsModule, PluginsModule, BookingImportModule, ReservationImportModule, LlmParseModule], + imports: [AppConfigModule, DatabaseModule, RealtimeModule, RealtimeGatewayModule, SchedulingModule, McpModule.forRoot({ accessPolicy: trekMcpAccessPolicy, validateAccess: trekMcpValidateAccess }), HealthModule, WeatherModule, HelpModule, AirportsModule, ConfigModule, SystemNoticesModule, GeoModule, MapsModule, PlaceEnrichmentModule, CategoriesModule, TagsModule, NotificationsModule, AtlasModule, VacayModule, PackingModule, TodoModule, BudgetModule, ReservationsModule, DaysModule, DayNotesModule, AccommodationsModule, AssignmentsModule, PlacesModule, TripsModule, CollabModule, FilesModule, PhotosModule, MemoriesModule, AirtrailModule, JourneyModule, CollectionsModule, ShareModule, TripInviteModule, TransitModule, FeedsModule, SettingsModule, BackupModule, AuthModule, OidcModule, OauthModule, AdminModule, AddonsModule, AuditModule, PermissionsModule, PluginsModule, BookingImportModule, ReservationImportModule, LlmParseModule], providers: [ // Default-deny: a route is authenticated unless it carries @Public() or // @OptionalAuth(), or declares its own @UseGuards chain. Protection used to diff --git a/server/src/nest/realtime/realtime-gateway.module.ts b/server/src/nest/realtime/realtime-gateway.module.ts new file mode 100644 index 000000000..570b5ac1b --- /dev/null +++ b/server/src/nest/realtime/realtime-gateway.module.ts @@ -0,0 +1,23 @@ +import { Module } from '@nestjs/common'; +import { RealtimeGateway } from './realtime.gateway'; +import { EphemeralTokenModule } from '../auth/ephemeral-token.module'; + +/** + * The transport, kept out of RealtimeModule on purpose. + * + * RealtimeModule is @Global and every domain pulls it in for the broadcast + * facade. A gateway there would follow into every e2e TestingModule, and Nest's + * SocketModule loads a websocket adapter for any app that has one: with none + * set, it reaches for @nestjs/platform-socket.io and calls process.exit(1) when + * it is absent. Eighteen harnesses died that way. + * + * So this module is imported by AppModule alone. Test apps that assemble a few + * domain modules by hand get the facade without the transport, which is what + * they had before; the ones that boot the real buildApp() get both, with + * TrekWsAdapter already registered. + */ +@Module({ + imports: [EphemeralTokenModule], + providers: [RealtimeGateway], +}) +export class RealtimeGatewayModule {} diff --git a/server/src/nest/realtime/realtime.gateway.ts b/server/src/nest/realtime/realtime.gateway.ts new file mode 100644 index 000000000..7cdcba016 --- /dev/null +++ b/server/src/nest/realtime/realtime.gateway.ts @@ -0,0 +1,161 @@ +import { Injectable, OnModuleDestroy } from '@nestjs/common'; +import { + ConnectedSocket, + MessageBody, + OnGatewayConnection, + OnGatewayDisconnect, + OnGatewayInit, + SubscribeMessage, + WebSocketGateway, +} from '@nestjs/websockets'; +import type { IncomingMessage } from 'node:http'; +import type { WebSocketServer } from 'ws'; +import { DatabaseService } from '../database/database.service'; +import { EphemeralTokenService } from '../auth/ephemeral-token.service'; +import { User } from '../../types'; +import { + joinRoom, + leaveAllRooms, + leaveRoom, + registerSocket, + userOf, + type TrekWebSocket, +} from './ws-state'; + +const HEARTBEAT_INTERVAL = 30_000; + +/** + * The /ws transport, as a Nest gateway. + * + * It replaces the hand-rolled `setupWebSocket(server)` that index.ts kicked off + * with a dynamic import after listen(). What moved is ownership: the handshake + * reads its collaborators from the container instead of importing the db + * singleton and the token store, and the heartbeat is a lifecycle hook rather + * than an interval nobody stops. + * + * What did NOT move is the socket registry. It stays module-scoped in + * ws-state.ts because seventeen bridges construct `new RealtimeService()` + * outside the container and must see the same rooms; see the note there. + * + * The wire protocol is unchanged, down to the frame names. TrekWsAdapter maps + * `{ type }` onto @SubscribeMessage, because the stock adapter dispatches on + * `{ event, data }` and every deployed client speaks the former. + */ +@Injectable() +@WebSocketGateway({ path: '/ws' }) +export class RealtimeGateway + implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect, OnModuleDestroy +{ + private heartbeat: ReturnType | null = null; + + constructor( + private readonly db: DatabaseService, + private readonly tokens: EphemeralTokenService, + ) {} + + afterInit(server: WebSocketServer): void { + this.heartbeat = setInterval(() => { + server.clients.forEach((ws) => { + const tws = ws as TrekWebSocket; + if (tws.isAlive === false) return tws.terminate(); + tws.isAlive = false; + tws.ping(); + }); + }, HEARTBEAT_INTERVAL); + this.heartbeat.unref?.(); + } + + onModuleDestroy(): void { + if (this.heartbeat) clearInterval(this.heartbeat); + this.heartbeat = null; + } + + /** + * Authenticate the upgrade, then admit the socket. + * + * Every rejection closes with the code the client already handles: 4001 for + * anything about identity, 4403 for the MFA policy. The order matters and is + * unchanged — a missing token is refused before the store is touched, and the + * password-version gate runs before the MFA one. + */ + handleConnection(socket: TrekWebSocket, request: IncomingMessage): void { + const url = new URL(request.url ?? '/', 'http://localhost'); + const token = url.searchParams.get('token'); + if (!token) { + socket.close(4001, 'Authentication required'); + return; + } + + const consumed = this.tokens.consumeWithMeta(token, 'ws'); + if (!consumed) { + socket.close(4001, 'Invalid or expired token'); + return; + } + + const row = this.db.get( + 'SELECT id, username, email, role, mfa_enabled, password_version FROM users WHERE id = ?', + consumed.userId, + ); + if (!row) { + socket.close(4001, 'User not found'); + return; + } + + // Session gate (defence-in-depth): reject a ws-token minted before a + // password change. Tokens carry the pv they were issued with; tokens minted + // without a pv (legacy) are treated as version 0, matching the JWT `pv` + // claim semantics in verifyJwtAndLoadUser. + const tokenPv = typeof consumed.pv === 'number' ? consumed.pv : 0; + const currentPv = typeof row.password_version === 'number' ? row.password_version : 0; + if (tokenPv !== currentPv) { + socket.close(4001, 'Invalid or expired token'); + return; + } + + // Don't leak password_version beyond the handshake. + const { password_version: _pv, ...user } = row; + const requireMfa = + this.db.get<{ value: string }>("SELECT value FROM app_settings WHERE key = 'require_mfa'")?.value === 'true'; + const mfaOk = user.mfa_enabled === 1 || user.mfa_enabled === true; + if (requireMfa && !mfaOk) { + socket.close(4403, 'MFA required'); + return; + } + + socket.isAlive = true; + const sid = registerSocket(socket, user as User); + socket.send(JSON.stringify({ type: 'welcome', socketId: sid })); + socket.on('pong', () => { socket.isAlive = true; }); + } + + handleDisconnect(socket: TrekWebSocket): void { + leaveAllRooms(socket); + } + + @SubscribeMessage('join') + handleJoin( + @MessageBody() message: { tripId?: number | string }, + @ConnectedSocket() socket: TrekWebSocket, + ): { type: string; tripId?: number; message?: string } | undefined { + const user = userOf(socket); + if (!user || !message?.tripId) return undefined; + + const tripId = Number(message.tripId); + if (!this.db.canAccessTrip(tripId, user.id)) { + return { type: 'error', message: 'Access denied' }; + } + joinRoom(socket, tripId); + return { type: 'joined', tripId }; + } + + @SubscribeMessage('leave') + handleLeave( + @MessageBody() message: { tripId?: number | string }, + @ConnectedSocket() socket: TrekWebSocket, + ): { type: string; tripId?: number; message?: string } | undefined { + if (!message?.tripId) return undefined; + const tripId = Number(message.tripId); + leaveRoom(socket, tripId); + return { type: 'left', tripId }; + } +} diff --git a/server/src/nest/realtime/realtime.module.ts b/server/src/nest/realtime/realtime.module.ts index a6eee3472..fd46a7eb2 100644 --- a/server/src/nest/realtime/realtime.module.ts +++ b/server/src/nest/realtime/realtime.module.ts @@ -3,10 +3,14 @@ import { RealtimeService } from './realtime.service'; /** * Global so every migrated module can inject RealtimeService without - * re-importing (same rationale as DatabaseModule). The service is - * stateless and dependency-free — the ws server itself is still wired - * in index.ts via setupWebSocket(); this module only provides the - * injectable broadcast seam. + * re-importing (same rationale as DatabaseModule). The service stays + * stateless and dependency-free so the out-of-container bridges can build it + * with a bare `new`. + * + * The transport itself lives in RealtimeGatewayModule, which AppModule imports + * directly. It cannot live here: this module is @Global, so a gateway would + * follow into every hand-assembled TestingModule and make Nest look for a + * websocket adapter none of them set. */ @Global() @Module({ diff --git a/server/src/nest/realtime/trek-ws.adapter.ts b/server/src/nest/realtime/trek-ws.adapter.ts new file mode 100644 index 000000000..973212a3e --- /dev/null +++ b/server/src/nest/realtime/trek-ws.adapter.ts @@ -0,0 +1,161 @@ +import { WsAdapter } from '@nestjs/platform-ws'; +import type { INestApplicationContext } from '@nestjs/common'; +import type { Server as HttpServer } from 'node:http'; +import type { MessageMappingProperties } from '@nestjs/websockets'; +import { WebSocketServer } from 'ws'; +import type { Observable } from 'rxjs'; +import { readEnv } from '../../app-config'; +import { setServer, type TrekWebSocket } from './ws-state'; + +// Per-connection message rate limiting. It lives in the adapter, not in the +// gateway's handlers, because the original counted EVERY inbound frame before +// parsing it: malformed JSON and unknown types count too. Counting only the +// frames that reach a @SubscribeMessage handler would leave a client free to +// flood the process with garbage. +const WS_MSG_LIMIT = 30; // max messages +const WS_MSG_WINDOW = 10_000; // per 10 seconds +const rates = new WeakMap(); + +function withinRate(socket: TrekWebSocket): boolean { + let rate = rates.get(socket); + if (!rate) { + rate = { count: 0, windowStart: Date.now() }; + rates.set(socket, rate); + } + const now = Date.now(); + if (now - rate.windowStart > WS_MSG_WINDOW) { + rate.count = 1; + rate.windowStart = now; + return true; + } + rate.count++; + return rate.count <= WS_MSG_LIMIT; +} + +/** + * TREK's wire protocol, on Nest's gateway plumbing. + * + * The stock WsAdapter dispatches on `{ event, data }`. TREK's client has always + * sent `{ type: 'join', tripId }` and expects `{ type: 'joined' }` back, and + * changing that would break every deployed client and every offline queue that + * has a frame parked in it. So the adapter translates: `type` names the handler, + * and the whole message is the payload. + * + * It also owns two things the stock adapter has no concept of, both of which + * predate this migration and neither of which may be dropped: + * + * - The origin check. `verifyClient` rejects a cross-origin upgrade with 403 + * before a socket exists, which is cheaper and stricter than checking after. + * - The per-socket `error` listener, attached FIRST. `ws` surfaces protocol + * violations (malformed frames, reserved close codes such as 1006) as an + * `error` event on the socket; unhandled, Node rethrows it and the process + * dies. That is #1576, and it only reproduces against a hostile client, so + * nothing in CI would catch its removal. + */ +export class TrekWsAdapter extends WsAdapter { + private readonly httpServerRef: HttpServer; + + /** + * Takes the http.Server directly, not the app. + * + * The base adapter accepts either, but only the server form guarantees the ws + * server lands on the socket this process actually listens on. Given the app, + * it would reach for Nest's own internal server, which buildApp never uses. + */ + constructor(httpServer: HttpServer) { + super(httpServer as unknown as INestApplicationContext); + this.httpServerRef = httpServer; + } + + /** + * Builds the ws server directly rather than delegating to the base. + * + * The base's port-0 branch creates it with `noServer: true` and performs the + * upgrade itself from its own registry. That path never consults + * `verifyClient`, because ws only calls it when ws owns the upgrade — so + * delegating would silently drop the origin check and answer 101 to any + * origin. Attaching to the http server keeps ws in charge of the handshake, + * which is where the 403 has always come from. + */ + create(_port: number, options?: Record): unknown { + const allowedOrigins = readEnv().http.wsOrigins; + const server = new WebSocketServer({ + server: this.httpServerRef, + path: (options?.path as string) ?? '/ws', + maxPayload: 64 * 1024, // 64 KB max message size + ...(allowedOrigins + ? { + verifyClient: ( + { origin }: { origin: string }, + cb: (ok: boolean, code?: number, msg?: string) => void, + ) => { + if (!origin || allowedOrigins.includes(origin)) cb(true); + else cb(false, 403, 'Origin not allowed'); + }, + } + : {}), + }); + + // A server-level error must not reach the process. The base binds one on the + // instances it creates; this one is ours, so it needs its own. + server.on('error', () => {}); + + // The fan-out primitives reach the live server through here, and so do the + // out-of-container bridges. + setServer(server); + return server; + } + + bindClientConnect(server: WebSocketServer, callback: (...args: unknown[]) => void): void { + server.on('connection', (socket: TrekWebSocket, request: unknown) => { + // Must stay above anything that can close early: a socket that never gets + // this listener can still crash the process while it finishes closing. + socket.on('error', () => socket.terminate()); + // The request rides along: the handshake reads the ws token off its query + // string, so dropping it here would leave handleConnection with nothing + // to authenticate. + callback(socket, request); + }); + } + + bindMessageHandlers( + socket: TrekWebSocket, + handlers: MessageMappingProperties[], + transform: (data: unknown) => Observable, + ): void { + socket.on('message', (buffer: Buffer) => { + if (!withinRate(socket)) { + if (socket.readyState === 1) { + socket.send(JSON.stringify({ type: 'error', message: 'Rate limit exceeded' })); + } + return; + } + + let message: { type?: unknown }; + try { + message = JSON.parse(buffer.toString()); + } catch { + return; // Malformed JSON, ignored exactly as before. + } + if (!message || typeof message !== 'object' || typeof message.type !== 'string') return; + + const handler = handlers.find((h) => h.message === message.type); + if (!handler) return; // An unknown type is ignored, not an error frame. + + // The whole message is the payload: TREK's frames are flat, so `tripId` + // sits beside `type` rather than under a `data` key. + transform(handler.callback(message, socket)).subscribe({ + next: (response) => { + if (response !== undefined && socket.readyState === 1) { + socket.send(JSON.stringify(response)); + } + }, + }); + }); + } + + async close(server: WebSocketServer): Promise { + setServer(null); + await super.close(server); + } +} diff --git a/server/src/nest/realtime/ws-state.ts b/server/src/nest/realtime/ws-state.ts new file mode 100644 index 000000000..488e7d403 --- /dev/null +++ b/server/src/nest/realtime/ws-state.ts @@ -0,0 +1,143 @@ +import { WebSocketServer, WebSocket } from 'ws'; +import { emitPluginEvent, pluginEventMeta } from '../../plugin-event-sink'; +import { User } from '../../types'; + +/** + * The socket registry: rooms, per-socket identity, and the three fan-out + * primitives. + * + * MODULE state, not provider state, and that is load-bearing. Seventeen + * `*.bridge.ts` files and notifications.instance.ts build `new + * RealtimeService()` by hand, outside the container, because they run outside + * it. If the rooms lived on a provider instance, those broadcasts would go to + * an empty map: no error, no log, just a client that stops updating. Same + * reasoning as the geo throttle cursor and the permissions cache. + * + * The gateway next door owns the connection lifecycle and the handshake and + * writes into here. This file deliberately knows nothing about auth. + */ + +export interface TrekWebSocket extends WebSocket { + isAlive: boolean; +} + +const rooms = new Map>(); +const socketRooms = new WeakMap>(); +const socketUser = new WeakMap(); +const socketId = new WeakMap(); + +/** + * A monotonic integer, NOT a uuid. + * + * The client echoes it back as the X-Socket-Id header and `broadcast` excludes + * the originator with `Number(excludeSid)`. `Number()` is NaN, and + * `NaN === NaN` is false, so every client would receive its own writes back. + * Nothing throws; it shows up as drag-and-drop that jumps back under your + * cursor. + */ +let nextSocketId = 1; + +let wss: WebSocketServer | null = null; + +export function setServer(server: WebSocketServer | null): void { + wss = server; +} + +export function getServer(): WebSocketServer | null { + return wss; +} + +export function registerSocket(ws: TrekWebSocket, user: User): number { + const sid = nextSocketId++; + socketId.set(ws, sid); + socketUser.set(ws, user); + socketRooms.set(ws, new Set()); + return sid; +} + +export function userOf(ws: TrekWebSocket): User | undefined { + return socketUser.get(ws); +} + +export function joinRoom(ws: TrekWebSocket, tripId: number): void { + if (!rooms.has(tripId)) rooms.set(tripId, new Set()); + rooms.get(tripId)!.add(ws); + socketRooms.get(ws)?.add(tripId); +} + +export function leaveRoom(ws: TrekWebSocket, tripId: number): void { + const room = rooms.get(tripId); + if (room) { + room.delete(ws); + if (room.size === 0) rooms.delete(tripId); + } + socketRooms.get(ws)?.delete(tripId); +} + +export function leaveAllRooms(ws: TrekWebSocket): void { + const mine = socketRooms.get(ws); + if (!mine) return; + for (const tripId of mine) leaveRoom(ws, tripId); +} + +/** + * Broadcast an event to all sockets in a trip room, optionally excluding a + * socket. + * + * When `onlyUserId` is given the event is delivered only to that user's sockets + * in the room — used to keep private packing items (#858) off other members' + * screens while still syncing the owner's own tabs. + */ +export function broadcast( + tripId: number | string, + eventType: string, + payload: Record, + excludeSid?: number | string, + onlyUserId?: number, +): void { + tripId = Number(tripId); + // Announce every CORE trip event (name only, never the payload) to subscribed + // plugins — before the room check so it fires even with no connected viewers + // and with no ws server at all, and skipping plugin:* re-broadcasts so a + // plugin's own events can't loop back. + if (!eventType.startsWith('plugin:')) emitPluginEvent(tripId, eventType, pluginEventMeta(eventType, payload)); + const room = rooms.get(tripId); + if (!room || room.size === 0) return; + + const excludeNum = excludeSid ? Number(excludeSid) : null; + + for (const ws of room) { + if (ws.readyState !== 1) continue; // WebSocket.OPEN === 1 + if (excludeNum && socketId.get(ws) === excludeNum) continue; + if (onlyUserId != null && socketUser.get(ws)?.id !== onlyUserId) continue; + ws.send(JSON.stringify({ type: eventType, tripId, ...payload })); + } +} + +/** Send a message to all sockets belonging to a specific user (e.g. trip invitations). */ +export function broadcastToUser( + userId: number, + payload: Record, + excludeSid?: number | string, +): void { + if (!wss) return; + const excludeNum = excludeSid ? Number(excludeSid) : null; + for (const ws of wss.clients) { + const tws = ws as TrekWebSocket; + if (tws.readyState !== 1) continue; + if (excludeNum && socketId.get(tws) === excludeNum) continue; + if (socketUser.get(tws)?.id === userId) tws.send(JSON.stringify(payload)); + } +} + +export function getOnlineUserIds(): Set { + const ids = new Set(); + if (!wss) return ids; + for (const ws of wss.clients) { + const tws = ws as TrekWebSocket; + if (tws.readyState !== 1) continue; + const user = socketUser.get(tws); + if (user) ids.add(user.id); + } + return ids; +} diff --git a/server/src/websocket.ts b/server/src/websocket.ts index 392942270..f273122ea 100644 --- a/server/src/websocket.ts +++ b/server/src/websocket.ts @@ -1,247 +1,14 @@ -import { WebSocketServer, WebSocket } from 'ws'; -import { readEnv } from './app-config'; -import { db, canAccessTrip } from './db/database'; -import { consumeEphemeralTokenWithMeta } from './nest/auth/ephemeral-tokens'; -import { emitPluginEvent, pluginEventMeta } from './plugin-event-sink'; -import { User } from './types'; -import http from 'node:http'; - -interface NomadWebSocket extends WebSocket { - isAlive: boolean; -} - -// Room management: tripId -> Set -const rooms = new Map>(); - -// Track which rooms each socket is in -const socketRooms = new WeakMap>(); - -// Track user info per socket -const socketUser = new WeakMap(); - -// Track unique socket ID -const socketId = new WeakMap(); -let nextSocketId = 1; - -let wss: WebSocketServer | null = null; - -// Per-connection message rate limiting -const WS_MSG_LIMIT = 30; // max messages -const WS_MSG_WINDOW = 10_000; // per 10 seconds -const socketMsgCounts = new WeakMap(); - -/** Attaches a WebSocket server with JWT auth, room-based trip channels, and heartbeat keep-alive. */ -function setupWebSocket(server: http.Server): void { - const allowedOrigins = readEnv().http.wsOrigins; - - wss = new WebSocketServer({ - server, - path: '/ws', - maxPayload: 64 * 1024, // 64 KB max message size - verifyClient: allowedOrigins - ? ({ origin }, cb) => { - if (!origin || allowedOrigins.includes(origin)) cb(true); - else cb(false, 403, 'Origin not allowed'); - } - : undefined, - }); - - const HEARTBEAT_INTERVAL = 30000; // 30 seconds - const heartbeat = setInterval(() => { - wss.clients.forEach((ws) => { - const nws = ws as NomadWebSocket; - if (nws.isAlive === false) return nws.terminate(); - nws.isAlive = false; - nws.ping(); - }); - }, HEARTBEAT_INTERVAL); - - wss.on('close', () => clearInterval(heartbeat)); - - wss.on('connection', (ws: WebSocket, req: http.IncomingMessage) => { - const nws = ws as NomadWebSocket; - - // ws emits protocol violations (malformed frames, reserved close codes such as 1006) - // as an 'error' event on the socket; unhandled, Node rethrows it and the process dies. - // Must stay above the auth guards below: they close and return early, and a socket that - // never gets this listener can still crash the server before it finishes closing. - nws.on('error', () => nws.terminate()); - - // Extract token from query param - const url = new URL(req.url, 'http://localhost'); - const token = url.searchParams.get('token'); - - if (!token) { - nws.close(4001, 'Authentication required'); - return; - } - - const consumed = consumeEphemeralTokenWithMeta(token, 'ws'); - if (!consumed) { - nws.close(4001, 'Invalid or expired token'); - return; - } - const { userId } = consumed; - - let row: (User & { password_version?: number }) | undefined; - row = db.prepare( - 'SELECT id, username, email, role, mfa_enabled, password_version FROM users WHERE id = ?' - ).get(userId) as (User & { password_version?: number }) | undefined; - if (!row) { - nws.close(4001, 'User not found'); - return; - } - // Session gate (defence-in-depth): reject a ws-token minted before a - // password change. Tokens carry the pv they were issued with; tokens - // minted without a pv (legacy) are treated as version 0, matching the - // JWT `pv` claim semantics in verifyJwtAndLoadUser. - const tokenPv = typeof consumed.pv === 'number' ? consumed.pv : 0; - const currentPv = typeof row.password_version === 'number' ? row.password_version : 0; - if (tokenPv !== currentPv) { - nws.close(4001, 'Invalid or expired token'); - return; - } - // Don't leak password_version beyond the handshake. - const { password_version: _pv, ...user } = row; - const requireMfa = (db.prepare("SELECT value FROM app_settings WHERE key = 'require_mfa'").get() as { value: string } | undefined)?.value === 'true'; - const mfaOk = user.mfa_enabled === 1 || user.mfa_enabled === true; - if (requireMfa && !mfaOk) { - nws.close(4403, 'MFA required'); - return; - } - - nws.isAlive = true; - const sid = nextSocketId++; - socketId.set(nws, sid); - socketUser.set(nws, user); - socketRooms.set(nws, new Set()); - nws.send(JSON.stringify({ type: 'welcome', socketId: sid })); - - nws.on('pong', () => { nws.isAlive = true; }); - - socketMsgCounts.set(nws, { count: 0, windowStart: Date.now() }); - - nws.on('message', (data) => { - // Rate limiting - const rate = socketMsgCounts.get(nws); - const now = Date.now(); - if (now - rate.windowStart > WS_MSG_WINDOW) { - rate.count = 1; - rate.windowStart = now; - } else { - rate.count++; - if (rate.count > WS_MSG_LIMIT) { - nws.send(JSON.stringify({ type: 'error', message: 'Rate limit exceeded' })); - return; - } - } - - let msg: { type: string; tripId?: number | string }; - try { - msg = JSON.parse(data.toString()); - } catch { - return; // Malformed JSON, ignore - } - - // Basic validation - if (!msg || typeof msg !== 'object' || typeof msg.type !== 'string') return; - - if (msg.type === 'join' && msg.tripId) { - const tripId = Number(msg.tripId); - // Verify the user has access to this trip - if (!canAccessTrip(tripId, user.id)) { - nws.send(JSON.stringify({ type: 'error', message: 'Access denied' })); - return; - } - // Add to room - if (!rooms.has(tripId)) rooms.set(tripId, new Set()); - rooms.get(tripId).add(nws); - socketRooms.get(nws).add(tripId); - nws.send(JSON.stringify({ type: 'joined', tripId })); - } - - if (msg.type === 'leave' && msg.tripId) { - const tripId = Number(msg.tripId); - leaveRoom(nws, tripId); - nws.send(JSON.stringify({ type: 'left', tripId })); - } - }); - - nws.on('close', () => { - // Clean up all rooms this socket was in - const myRooms = socketRooms.get(nws); - if (myRooms) { - for (const tripId of myRooms) { - leaveRoom(nws, tripId); - } - } - }); - }); - - console.log('WebSocket server attached at /ws'); -} - -function leaveRoom(ws: NomadWebSocket, tripId: number): void { - const room = rooms.get(tripId); - if (room) { - room.delete(ws); - if (room.size === 0) rooms.delete(tripId); - } - const myRooms = socketRooms.get(ws); - if (myRooms) myRooms.delete(tripId); -} - /** - * Broadcast an event to all sockets in a trip room, optionally excluding a socket. - * When `onlyUserId` is given the event is delivered only to that user's sockets in - * the room — used to keep private packing items (#858) off other members' screens - * while still syncing the owner's own tabs. + * The transport's public surface, for code outside the Nest container. + * + * The implementation moved to nest/realtime: the connection lifecycle and the + * handshake are RealtimeGateway, the wire protocol is TrekWsAdapter, and the + * socket registry is ws-state.ts. This file stays because it is the import path + * the rest of the tree knows, including 115 test files that `vi.mock` it to + * assert on broadcasts. Repointing them all would be a large diff that proves + * nothing, and this re-export keeps every one of those mocks working. + * + * `setupWebSocket` is gone: the server is created by the adapter now, when Nest + * initialises the gateway, rather than by index.ts after listen(). */ -function broadcast(tripId: number | string, eventType: string, payload: Record, excludeSid?: number | string, onlyUserId?: number): void { - tripId = Number(tripId); - // Announce every CORE trip event (name only, never the payload) to subscribed - // plugins — before the room check so it fires even with no connected viewers, and - // skipping plugin:* re-broadcasts so a plugin's own events can't loop back. - if (!eventType.startsWith('plugin:')) emitPluginEvent(tripId, eventType, pluginEventMeta(eventType, payload)); - const room = rooms.get(tripId); - if (!room || room.size === 0) return; - - const excludeNum = excludeSid ? Number(excludeSid) : null; - - for (const ws of room) { - if (ws.readyState !== 1) continue; // WebSocket.OPEN === 1 - // Exclude the specific socket that triggered the change - if (excludeNum && socketId.get(ws) === excludeNum) continue; - if (onlyUserId != null && socketUser.get(ws)?.id !== onlyUserId) continue; - ws.send(JSON.stringify({ type: eventType, tripId, ...payload })); - } -} - -/** Send a message to all sockets belonging to a specific user (e.g., for trip invitations). */ -function broadcastToUser(userId: number, payload: Record, excludeSid?: number | string): void { - if (!wss) return; - const excludeNum = excludeSid ? Number(excludeSid) : null; - for (const ws of wss.clients) { - const nws = ws as NomadWebSocket; - if (nws.readyState !== 1) continue; - if (excludeNum && socketId.get(nws) === excludeNum) continue; - const user = socketUser.get(nws); - if (user?.id === userId) { - nws.send(JSON.stringify(payload)); - } - } -} - -function getOnlineUserIds(): Set { - const ids = new Set(); - if (!wss) return ids; - for (const ws of wss.clients) { - const nws = ws as NomadWebSocket; - if (nws.readyState !== 1) continue; - const user = socketUser.get(nws); - if (user) ids.add(user.id); - } - return ids; -} - -export { setupWebSocket, broadcast, broadcastToUser, getOnlineUserIds }; +export { broadcast, broadcastToUser, getOnlineUserIds } from './nest/realtime/ws-state'; diff --git a/server/tests/helpers/ws-client.ts b/server/tests/helpers/ws-client.ts index 27c248cfe..3667b050a 100644 --- a/server/tests/helpers/ws-client.ts +++ b/server/tests/helpers/ws-client.ts @@ -3,7 +3,7 @@ * * Usage: * import http from 'http'; - * import { setupWebSocket } from '../../src/websocket'; + * import { getHttpServer } from '../../src/bootstrap'; * import { WsTestClient, getWsToken } from '../helpers/ws-client'; * * let server: http.Server; @@ -12,7 +12,7 @@ * beforeAll(async () => { * const app = createApp(); * server = http.createServer(app); - * setupWebSocket(server); + * const server = getHttpServer(); * await new Promise(res => server.listen(0, res)); * }); * diff --git a/server/tests/integration/journey.test.ts b/server/tests/integration/journey.test.ts index 26128e9be..7f912d71f 100644 --- a/server/tests/integration/journey.test.ts +++ b/server/tests/integration/journey.test.ts @@ -52,7 +52,6 @@ vi.mock('../../src/config', () => ({ vi.mock('../../src/websocket', () => ({ broadcast: vi.fn(), broadcastToUser: vi.fn(), - setupWebSocket: vi.fn(), getOnlineUserIds: vi.fn(() => []), })); vi.mock('../../src/services/memories/immichService', () => ({ diff --git a/server/tests/unit/nest/realtime.gateway.test.ts b/server/tests/unit/nest/realtime.gateway.test.ts new file mode 100644 index 000000000..1c0af527d --- /dev/null +++ b/server/tests/unit/nest/realtime.gateway.test.ts @@ -0,0 +1,288 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +/** + * The gateway's handshake, and the socket registry it writes into. + * + * The transport suites in tests/websocket drive real sockets end to end; these + * cases go at the pieces directly, so the ways this can fail SILENTLY are + * pinned by name rather than by whether a browser happened to work. + */ +vi.mock('../../../src/plugin-event-sink', () => ({ + emitPluginEvent: vi.fn(), + pluginEventMeta: vi.fn(() => ({})), +})); + +import { RealtimeGateway } from '../../../src/nest/realtime/realtime.gateway'; +import { + broadcast, + broadcastToUser, + getOnlineUserIds, + joinRoom, + registerSocket, + setServer, + type TrekWebSocket, +} from '../../../src/nest/realtime/ws-state'; +import { emitPluginEvent } from '../../../src/plugin-event-sink'; +import type { DatabaseService } from '../../../src/nest/database/database.service'; +import type { EphemeralTokenService } from '../../../src/nest/auth/ephemeral-token.service'; +import type { User } from '../../../src/types'; + +interface FakeSocket extends TrekWebSocket { + sent: string[]; + closedWith: [number, string] | null; +} + +function socket(): FakeSocket { + const sent: string[] = []; + const s = { + readyState: 1, + isAlive: false, + sent, + closedWith: null as [number, string] | null, + send: (raw: string) => { sent.push(raw); }, + close: (code: number, reason: string) => { s.closedWith = [code, reason]; }, + on: vi.fn(), + terminate: vi.fn(), + ping: vi.fn(), + }; + return s as unknown as FakeSocket; +} + +const rows = new Map(); +const db = { + get: (sql: string) => (sql.includes('app_settings') ? rows.get('mfa') : rows.get('user')), + canAccessTrip: (tripId: number) => tripId === 7, +} as unknown as DatabaseService; + +const consumeWithMeta = vi.fn(); +const tokens = { consumeWithMeta } as unknown as EphemeralTokenService; + +function connect(url: string) { + const gw = new RealtimeGateway(db, tokens); + const ws = socket(); + gw.handleConnection(ws, { url } as never); + return { gw, ws }; +} + +beforeEach(() => { + vi.clearAllMocks(); + rows.clear(); + rows.set('user', { id: 3, username: 'm', email: 'm@x.test', role: 'user', mfa_enabled: 0, password_version: 2 }); + consumeWithMeta.mockReturnValue({ userId: 3, pv: 2 }); +}); + +describe('RealtimeGateway handshake', () => { + it('WSGW-001: refuses a connect with no token before touching the store', () => { + const { ws } = connect('/ws'); + expect(ws.closedWith).toEqual([4001, 'Authentication required']); + expect(consumeWithMeta).not.toHaveBeenCalled(); + }); + + it('WSGW-002: refuses an unknown or spent token', () => { + consumeWithMeta.mockReturnValue(null); + expect(connect('/ws?token=x').ws.closedWith).toEqual([4001, 'Invalid or expired token']); + }); + + it('WSGW-003: rejects a token minted before a password change', () => { + // The pv gate. Same close reason as an unknown token on purpose: a client + // must not be able to tell a stale token from a forged one. + consumeWithMeta.mockReturnValue({ userId: 3, pv: 1 }); + expect(connect('/ws?token=x').ws.closedWith).toEqual([4001, 'Invalid or expired token']); + }); + + it('WSGW-004: treats a token minted without a pv as version 0', () => { + consumeWithMeta.mockReturnValue({ userId: 3 }); + rows.set('user', { id: 3, email: 'm@x.test', role: 'user', mfa_enabled: 0, password_version: 0 }); + expect(connect('/ws?token=x').ws.closedWith).toBeNull(); + }); + + it('WSGW-005: enforces the MFA policy with its own close code', () => { + rows.set('mfa', { value: 'true' }); + expect(connect('/ws?token=x').ws.closedWith).toEqual([4403, 'MFA required']); + }); + + it('WSGW-006: admits an MFA-enabled user while the policy is on', () => { + rows.set('mfa', { value: 'true' }); + rows.set('user', { id: 3, email: 'm@x.test', role: 'user', mfa_enabled: 1, password_version: 2 }); + expect(connect('/ws?token=x').ws.closedWith).toBeNull(); + }); + + it('WSGW-007: the welcome frame carries a NUMERIC socket id', () => { + // Load-bearing: the client echoes this back as X-Socket-Id and broadcast + // excludes the originator with Number(excludeSid). A uuid would be NaN, + // NaN === NaN is false, and every client would receive its own writes back. + // Nothing throws; it shows up as drag-and-drop that jumps under the cursor. + const { ws } = connect('/ws?token=x'); + const welcome = JSON.parse(ws.sent[0]) as { type: string; socketId: unknown }; + expect(welcome.type).toBe('welcome'); + expect(Number.isInteger(welcome.socketId)).toBe(true); + }); + + it('WSGW-007b: survives an upgrade request with no url, rather than throwing at it', () => { + const gw = new RealtimeGateway(db, tokens); + const ws = socket(); + expect(() => gw.handleConnection(ws, {} as never)).not.toThrow(); + expect(ws.closedWith).toEqual([4001, 'Authentication required']); + }); + + it('WSGW-007c: a user row with no password_version reads as version 0', () => { + // Legacy rows predate the column. Treating a missing value as 0 is what + // makes a legacy token match a legacy row instead of being refused forever. + consumeWithMeta.mockReturnValue({ userId: 3, pv: 0 }); + rows.set('user', { id: 3, email: 'm@x.test', role: 'user', mfa_enabled: 0 }); + expect(connect('/ws?token=x').ws.closedWith).toBeNull(); + }); + + it('WSGW-008: never leaks password_version past the handshake', () => { + const { gw, ws } = connect('/ws?token=x'); + gw.handleJoin({ tripId: 7 }, ws); + broadcast(7, 'trip:updated' as never, {} as never); + expect(ws.sent.join('')).not.toContain('password_version'); + }); +}); + +describe('RealtimeGateway rooms', () => { + it('WSGW-010: join refuses a trip the user cannot reach', () => { + const { gw, ws } = connect('/ws?token=x'); + expect(gw.handleJoin({ tripId: 99 }, ws)).toEqual({ type: 'error', message: 'Access denied' }); + }); + + it('WSGW-011: join and leave answer flat, keyed by type', () => { + const { gw, ws } = connect('/ws?token=x'); + expect(gw.handleJoin({ tripId: 7 }, ws)).toEqual({ type: 'joined', tripId: 7 }); + expect(gw.handleLeave({ tripId: 7 }, ws)).toEqual({ type: 'left', tripId: 7 }); + }); + + it('WSGW-012: a frame with no tripId is ignored rather than answered', () => { + const { gw, ws } = connect('/ws?token=x'); + expect(gw.handleJoin({}, ws)).toBeUndefined(); + expect(gw.handleLeave({}, ws)).toBeUndefined(); + }); +}); + +describe('RealtimeGateway heartbeat', () => { + it('WSGW-020: pings a live socket and terminates one that missed the last ping', () => { + vi.useFakeTimers(); + try { + const alive = socket(); + alive.isAlive = true; + const stale = socket(); + stale.isAlive = false; + const gw = new RealtimeGateway(db, tokens); + gw.afterInit({ clients: new Set([alive, stale]) } as never); + + vi.advanceTimersByTime(30_000); + + // A socket that answered the last ping is asked again and marked pending. + expect(alive.ping).toHaveBeenCalledTimes(1); + expect(alive.isAlive).toBe(false); + // One that never answered is dropped rather than pinged forever. + expect(stale.terminate).toHaveBeenCalledTimes(1); + expect(stale.ping).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it('WSGW-021: the interval stops with the module, so a test process can exit', () => { + vi.useFakeTimers(); + try { + const ws = socket(); + ws.isAlive = true; + const gw = new RealtimeGateway(db, tokens); + gw.afterInit({ clients: new Set([ws]) } as never); + gw.onModuleDestroy(); + + vi.advanceTimersByTime(90_000); + expect(ws.ping).not.toHaveBeenCalled(); + // Idempotent: shutdown can run twice without clearing a stale handle. + expect(() => gw.onModuleDestroy()).not.toThrow(); + } finally { + vi.useRealTimers(); + } + }); + + it('WSGW-022: a pong marks the socket live again', () => { + const { ws } = connect('/ws?token=x'); + const pong = (ws.on as unknown as { mock: { calls: [string, () => void][] } }).mock.calls + .find(([event]) => event === 'pong'); + expect(pong).toBeDefined(); + ws.isAlive = false; + pong![1](); + expect(ws.isAlive).toBe(true); + }); +}); + +describe('ws-state fan-out', () => { + it('WSST-001: broadcast excludes the originating socket by its numeric id', () => { + const a = socket(); + const b = socket(); + const sidA = registerSocket(a, { id: 1 } as User); + registerSocket(b, { id: 2 } as User); + joinRoom(a, 42); + joinRoom(b, 42); + + broadcast(42, 'place:created' as never, { placeId: 9 } as never, sidA); + + expect(a.sent).toHaveLength(0); + expect(JSON.parse(b.sent[0])).toMatchObject({ type: 'place:created', tripId: 42, placeId: 9 }); + }); + + it('WSST-002: onlyUserId narrows delivery to one member', () => { + const mine = socket(); + const theirs = socket(); + registerSocket(mine, { id: 1 } as User); + registerSocket(theirs, { id: 2 } as User); + joinRoom(mine, 43); + joinRoom(theirs, 43); + + broadcast(43, 'packing:updated' as never, {} as never, undefined, 1); + + expect(mine.sent).toHaveLength(1); + expect(theirs.sent).toHaveLength(0); + }); + + it('WSST-003: announces core events to plugins BEFORE the room check', () => { + // No room, no viewers, and with a plugin subscribed the event still has to + // fire. Ordering it after the early return would silence plugin events on + // every trip nobody happens to be looking at. + broadcast(9999, 'day:updated' as never, {} as never); + expect(emitPluginEvent).toHaveBeenCalledWith(9999, 'day:updated', expect.anything()); + }); + + it('WSST-004: does not re-announce a plugin broadcast, which would loop', () => { + broadcast(9998, 'plugin:acme:ping' as never, {} as never); + expect(emitPluginEvent).not.toHaveBeenCalled(); + }); + + it('WSST-004b: a closing socket is skipped by both fan-outs', () => { + const open = socket(); + const closing = socket(); + (closing as { readyState: number }).readyState = 2; // CLOSING + registerSocket(open, { id: 1 } as User); + registerSocket(closing, { id: 2 } as User); + joinRoom(open, 44); + joinRoom(closing, 44); + setServer({ clients: new Set([open, closing]) } as never); + + broadcast(44, 'trip:updated' as never, {} as never); + broadcastToUser(2, { type: 'trip:invite' }); + + expect(open.sent).toHaveLength(1); + expect(closing.sent).toHaveLength(0); + // And it is not counted as online, which drives the admin presence dots. + expect(getOnlineUserIds()).toEqual(new Set([1])); + }); + + it('WSST-004c: a socket with no registered user is not reported online', () => { + const stray = socket(); + setServer({ clients: new Set([stray]) } as never); + expect(getOnlineUserIds()).toEqual(new Set()); + }); + + it('WSST-005: broadcastToUser and getOnlineUserIds answer empty with no server', () => { + setServer(null); + expect(() => broadcastToUser(1, { type: 'x' })).not.toThrow(); + expect(getOnlineUserIds()).toEqual(new Set()); + }); +}); diff --git a/server/tests/unit/nest/trek-ws-adapter.test.ts b/server/tests/unit/nest/trek-ws-adapter.test.ts new file mode 100644 index 000000000..bc5828c22 --- /dev/null +++ b/server/tests/unit/nest/trek-ws-adapter.test.ts @@ -0,0 +1,263 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +/** + * TrekWsAdapter: the wire protocol and the per-socket flood guard. + * + * Both are things the stock WsAdapter does differently, and both fail quietly + * if this class regresses. A protocol slip means `join` is never handled, the + * room stays empty, `broadcast` returns at its size check and every client + * simply stops updating. A missing flood guard is only visible under a hostile + * client. + */ +// The origin allowlist is env-driven and empty in the test environment, so the +// branch that builds verifyClient would never run. Driving readEnv lets both +// sides of it be asserted rather than assumed. +const { wsOrigins } = vi.hoisted(() => ({ wsOrigins: { value: null as string[] | null } })); +vi.mock('../../../src/app-config', async (importOriginal) => { + const actual = (await importOriginal()) as Record; + return { + ...actual, + readEnv: () => ({ + ...(actual.readEnv as () => { http: Record })(), + http: { + ...((actual.readEnv as () => { http: Record })().http), + wsOrigins: wsOrigins.value, + }, + }), + }; +}); + +import { TrekWsAdapter } from '../../../src/nest/realtime/trek-ws.adapter'; +import { getServer } from '../../../src/nest/realtime/ws-state'; +import type { Server as HttpServer } from 'node:http'; + +type MessageListener = (buffer: Buffer) => void; + +function fakeSocket() { + const sent: string[] = []; + const listeners: Record = {}; + return { + readyState: 1, + sent, + terminate: vi.fn(), + send: (raw: string) => { sent.push(raw); }, + on: (event: string, fn: MessageListener) => { + (listeners[event] ??= []).push(fn); + }, + emit: (event: string, payload: Buffer) => { + for (const fn of listeners[event] ?? []) fn(payload); + }, + listenerCount: (event: string) => (listeners[event] ?? []).length, + }; +} + +const adapter = new TrekWsAdapter({} as HttpServer); +const frame = (o: unknown) => Buffer.from(JSON.stringify(o)); +/** The adapter hands results to Nest's transform; here it is identity. */ +const transform = (v: unknown) => ({ subscribe: (o: { next: (x: unknown) => void }) => o.next(v) }) as never; + +let handled: unknown[]; +let handlers: { message: string; callback: (data: unknown, socket: unknown) => unknown }[]; + +beforeEach(() => { + handled = []; + handlers = [ + { message: 'join', callback: (data) => { handled.push(data); return { type: 'joined' }; } }, + { message: 'leave', callback: (data) => { handled.push(data); return undefined; } }, + ]; +}); + +describe('TrekWsAdapter wire protocol', () => { + it('WSAD-001: dispatches on `type`, and hands the WHOLE frame to the handler', () => { + // TREK's frames are flat: tripId sits beside type, not under a `data` key. + // The stock adapter reads `message.event` and passes `message.data`, which + // would leave join unhandled and the payload undefined. + const socket = fakeSocket(); + adapter.bindMessageHandlers(socket as never, handlers as never, transform); + socket.emit('message', frame({ type: 'join', tripId: 7 })); + expect(handled).toEqual([{ type: 'join', tripId: 7 }]); + expect(JSON.parse(socket.sent[0])).toEqual({ type: 'joined' }); + }); + + it('WSAD-001b: a socket that closed mid-dispatch is not written to', () => { + const socket = fakeSocket(); + adapter.bindMessageHandlers(socket as never, handlers as never, transform); + (socket as { readyState: number }).readyState = 3; // CLOSED + socket.emit('message', frame({ type: 'join', tripId: 7 })); + expect(handled).toHaveLength(1); // the handler still ran + expect(socket.sent).toHaveLength(0); // the answer went nowhere + }); + + it('WSAD-002: a handler that answers undefined sends nothing', () => { + const socket = fakeSocket(); + adapter.bindMessageHandlers(socket as never, handlers as never, transform); + socket.emit('message', frame({ type: 'leave', tripId: 7 })); + expect(socket.sent).toHaveLength(0); + }); + + it('WSAD-003: malformed JSON and unknown types are ignored, not answered', () => { + const socket = fakeSocket(); + adapter.bindMessageHandlers(socket as never, handlers as never, transform); + socket.emit('message', Buffer.from('{not json')); + socket.emit('message', frame({ type: 'nope' })); + socket.emit('message', frame({ tripId: 7 })); + socket.emit('message', frame('a string')); + expect(handled).toHaveLength(0); + expect(socket.sent).toHaveLength(0); + }); +}); + +describe('TrekWsAdapter flood guard', () => { + it('WSAD-010: counts EVERY frame, including the ones no handler wants', () => { + // The guard sits before parsing on purpose. Counting only the frames that + // reach a handler would let a client flood the process with garbage for + // free, which is what the limit exists to stop. + const socket = fakeSocket(); + adapter.bindMessageHandlers(socket as never, handlers as never, transform); + + for (let i = 0; i < 30; i++) socket.emit('message', Buffer.from('{not json')); + expect(socket.sent).toHaveLength(0); + + socket.emit('message', frame({ type: 'join', tripId: 7 })); + expect(JSON.parse(socket.sent[0])).toEqual({ type: 'error', message: 'Rate limit exceeded' }); + expect(handled).toHaveLength(0); + }); + + it('WSAD-010b: a socket that closed while flooding is not written to either', () => { + // The refusal is still a write. A client that hangs up mid-flood would + // otherwise take an ERR_HTTP_HEADERS-shaped error on the way out. + const socket = fakeSocket(); + adapter.bindMessageHandlers(socket as never, handlers as never, transform); + for (let i = 0; i < 30; i++) socket.emit('message', Buffer.from('{not json')); + (socket as { readyState: number }).readyState = 3; // CLOSED + socket.emit('message', frame({ type: 'join', tripId: 7 })); + expect(socket.sent).toHaveLength(0); + }); + + it('WSAD-011: the window reopens, so a well-behaved client recovers', () => { + vi.useFakeTimers(); + try { + const socket = fakeSocket(); + adapter.bindMessageHandlers(socket as never, handlers as never, transform); + for (let i = 0; i < 31; i++) socket.emit('message', frame({ type: 'join', tripId: 7 })); + expect(socket.sent.some((s) => s.includes('Rate limit'))).toBe(true); + + vi.advanceTimersByTime(10_001); + socket.sent.length = 0; + socket.emit('message', frame({ type: 'join', tripId: 7 })); + expect(JSON.parse(socket.sent[0])).toEqual({ type: 'joined' }); + } finally { + vi.useRealTimers(); + } + }); +}); + +describe('TrekWsAdapter connection binding', () => { + it('WSAD-020: attaches the error listener BEFORE handing the socket on', () => { + // A protocol violation (a malformed frame, a reserved close code such as + // 1006) arrives as an `error` event; unhandled, Node rethrows it and the + // process dies. That is #1576, and it only reproduces against a hostile + // client, so nothing in CI would notice this listener going missing. + const socket = fakeSocket(); + const server = { + on: (_event: string, cb: (s: unknown, r: unknown) => void) => cb(socket, { url: '/ws' }), + }; + let listenersWhenCalled = -1; + adapter.bindClientConnect(server as never, () => { + listenersWhenCalled = socket.listenerCount('error'); + }); + expect(listenersWhenCalled).toBe(1); + }); + + it('WSAD-021: forwards the upgrade request, which carries the ws token', () => { + const socket = fakeSocket(); + const request = { url: '/ws?token=abc' }; + const server = { + on: (_event: string, cb: (s: unknown, r: unknown) => void) => cb(socket, request), + }; + const seen: unknown[] = []; + adapter.bindClientConnect(server as never, (...args: unknown[]) => seen.push(...args)); + expect(seen[1]).toBe(request); + }); +}); + +describe('TrekWsAdapter server creation', () => { + /** A real http.Server is never listened on; ws only needs it to hang an upgrade handler. */ + const httpServer = { on: vi.fn(), once: vi.fn(), removeListener: vi.fn(), emit: vi.fn() } as unknown as HttpServer; + + it('WSAD-030: attaches to the http server it was given, on the gateway path', () => { + const ad = new TrekWsAdapter(httpServer); + const server = ad.create(0, { path: '/ws' }) as { options: Record }; + try { + expect(server.options.path).toBe('/ws'); + // The registry has to see the live server, or every bridge broadcast + // silently goes nowhere. + expect(getServer()).toBe(server); + } finally { + void ad.close(server as never); + } + }); + + it('WSAD-031: with no allowlist configured, ws gets no verifyClient at all', () => { + // Rather than one that always says yes: an always-true hook is a hook that + // can be broken into a false later without anyone noticing. + wsOrigins.value = null; + const ad = new TrekWsAdapter(httpServer); + const server = ad.create(0, { path: '/ws' }) as { options: { verifyClient?: unknown } }; + try { + expect(typeof server.options.verifyClient).not.toBe('function'); + } finally { + void ad.close(server as never); + } + }); + + it('WSAD-031b: with an allowlist, a foreign origin is refused 403 before a socket exists', () => { + wsOrigins.value = ['https://trip.example']; + const ad = new TrekWsAdapter(httpServer); + const server = ad.create(0, { path: '/ws' }) as { options: { verifyClient?: unknown } }; + try { + const verify = server.options.verifyClient as ( + info: { origin: string }, + cb: (ok: boolean, code?: number, msg?: string) => void, + ) => void; + const seen: unknown[][] = []; + verify({ origin: 'https://evil.example' }, (...args) => seen.push(args)); + verify({ origin: 'https://trip.example' }, (...args) => seen.push(args)); + // A same-origin upgrade sends no Origin header at all; refusing that would + // lock out every non-browser client. + verify({ origin: '' }, (...args) => seen.push(args)); + expect(seen).toEqual([[false, 403, 'Origin not allowed'], [true], [true]]); + } finally { + void ad.close(server as never); + } + }); + + it('WSAD-030b: falls back to /ws when the gateway declares no path', () => { + wsOrigins.value = null; + const ad = new TrekWsAdapter(httpServer); + const server = ad.create(0, {}) as { options: Record }; + try { + expect(server.options.path).toBe('/ws'); + } finally { + void ad.close(server as never); + } + }); + + it('WSAD-031c: a server-level error is absorbed, not rethrown at the process', () => { + wsOrigins.value = null; + const ad = new TrekWsAdapter(httpServer); + const server = ad.create(0, { path: '/ws' }) as { emit: (e: string, err: Error) => boolean }; + try { + expect(() => server.emit('error', new Error('upstream reset'))).not.toThrow(); + } finally { + void ad.close(server as never); + } + }); + + it('WSAD-032: closing clears the registry, so a stale server is never broadcast to', async () => { + const ad = new TrekWsAdapter(httpServer); + const server = ad.create(0, { path: '/ws' }); + await ad.close(server as never); + expect(getServer()).toBeNull(); + }); +}); diff --git a/server/tests/websocket/connection.test.ts b/server/tests/websocket/connection.test.ts index 2721ba084..f8aab7498 100644 --- a/server/tests/websocket/connection.test.ts +++ b/server/tests/websocket/connection.test.ts @@ -48,13 +48,12 @@ vi.mock('../../src/config', () => ({ })); import type { INestApplication } from '@nestjs/common'; -import { buildApp } from '../../src/bootstrap'; +import { buildApp, getHttpServer } from '../../src/bootstrap'; import { createTables } from '../../src/db/schema'; import { runMigrations } from '../../src/db/migrations'; import { resetTestDb, resetRateLimits } from '../helpers/test-db'; import { createUser, createTrip } from '../helpers/factories'; import { authCookie } from '../helpers/auth'; -import { setupWebSocket } from '../../src/websocket'; import { createEphemeralToken } from '../../src/nest/auth/ephemeral-tokens'; import { createWsToken } from '../../src/nest/auth/auth.bridge'; @@ -69,8 +68,9 @@ beforeAll(async () => { // Real WebSocket against the unified NestJS app (Express is gone). buildApp owns // the same composition production uses; we attach the real ws server to it. nestApp = await buildApp(); - server = http.createServer(nestApp.getHttpAdapter().getInstance()); - setupWebSocket(server); + // buildApp owns the server now, because the ws gateway is bound to it before + // app.init(). Creating a second one here would leave /ws unreachable. + server = getHttpServer(); await new Promise(resolve => server.listen(0, resolve)); const addr = server.address() as { port: number }; diff --git a/server/tests/websocket/wire-envelope.test.ts b/server/tests/websocket/wire-envelope.test.ts index c4c8fe5c9..3b129538a 100644 --- a/server/tests/websocket/wire-envelope.test.ts +++ b/server/tests/websocket/wire-envelope.test.ts @@ -50,12 +50,12 @@ vi.mock('../../src/config', () => ({ })); import type { INestApplication } from '@nestjs/common'; -import { buildApp } from '../../src/bootstrap'; +import { buildApp, getHttpServer } from '../../src/bootstrap'; import { createTables } from '../../src/db/schema'; import { runMigrations } from '../../src/db/migrations'; import { resetTestDb } from '../helpers/test-db'; import { createUser, createTrip } from '../helpers/factories'; -import { setupWebSocket, broadcast, broadcastToUser } from '../../src/websocket'; +import { broadcast, broadcastToUser } from '../../src/websocket'; import { createEphemeralToken } from '../../src/nest/auth/ephemeral-tokens'; let server: http.Server; @@ -66,8 +66,8 @@ beforeAll(async () => { createTables(testDb); runMigrations(testDb); nestApp = await buildApp(); - server = http.createServer(nestApp.getHttpAdapter().getInstance()); - setupWebSocket(server); + // buildApp binds /ws to the server it creates, before app.init(). + server = getHttpServer(); await new Promise(resolve => server.listen(0, resolve)); const addr = server.address() as { port: number }; wsUrl = `ws://127.0.0.1:${addr.port}/ws`;