mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-19 05:26:47 +00:00
Maurice
00cbc5aae3
tests/setup.ts mocks src/config once with tests/helpers/test-config.ts. A suite that needs other values calls overrideTestConfig(), and the setup file restores the defaults after every case. The 191 per-file factories are gone; config.test.ts takes the real module through vi.unmock, and the few suites that signed tokens with their own literal secret read it from TEST_CONFIG.
993 lines
37 KiB
TypeScript
993 lines
37 KiB
TypeScript
/**
|
|
* WebSocket connection tests.
|
|
* Covers WS-001 to WS-006, WS-008 to WS-017.
|
|
*
|
|
* Starts a real HTTP server on a random port and connects via the `ws` library.
|
|
*/
|
|
import { describe, it, expect, vi, beforeAll, afterAll, beforeEach } from 'vitest';
|
|
import http from 'http';
|
|
import net from 'node:net';
|
|
import crypto from 'node:crypto';
|
|
import request from 'supertest';
|
|
import WebSocket from 'ws';
|
|
import { broadcastToUser, getOnlineUserIds } from '../../src/nest/realtime/ws-state';
|
|
|
|
vi.mock('../../src/db/database', async () => {
|
|
const { createSnapshotTestDb, buildDbMock } = await import('../helpers/db-mock');
|
|
return buildDbMock(createSnapshotTestDb());
|
|
});
|
|
|
|
import type { INestApplication } from '@nestjs/common';
|
|
import { buildApp, getHttpServer } from '../../src/bootstrap';
|
|
import { db as testDb } from '../../src/db/database';
|
|
import { resetTestDb, resetRateLimits } from '../helpers/test-db';
|
|
import { createUser, createTrip } from '../helpers/factories';
|
|
import { authCookie } from '../helpers/auth';
|
|
import { createEphemeralToken } from '../../src/nest/auth-core/ephemeral-tokens';
|
|
import { TokenService } from '../../src/nest/tokens/token.service';
|
|
import { EphemeralTokenService } from '../../src/nest/auth-core/ephemeral-token.service';
|
|
import { createTestMcpTokensRepo, createTestUsersRepo } from '../helpers/test-uow';
|
|
import { MikroORM } from '@mikro-orm/core';
|
|
import { Users } from '../../src/db/entities/Users.entity';
|
|
import type { FactoryOrm } from '../helpers/factories/context';
|
|
import { deleteRows, updateRows } from '../helpers/factories/rows';
|
|
import { setAppSetting } from '../helpers/factories/settings';
|
|
import { readUser } from '../helpers/factories/users';
|
|
|
|
// The gateway consumes ws-tokens through its injected TokenService; the
|
|
// ephemeral store is module-scoped on purpose, so a directly-constructed
|
|
// instance mints tokens the app under test accepts (TokenService is a leaf —
|
|
// its own unit suite constructs it the same way). `createSnapshotTestDb`
|
|
// (behind the `../../src/db/database` mock above) already carries the full
|
|
// migrated schema, so the repositories can build immediately — no
|
|
// `createTables`/`runMigrations` step needed first, unlike suites that start
|
|
// from a bare `:memory:` handle.
|
|
let tokenService: TokenService;
|
|
const createWsToken = (...args: Parameters<TokenService['createWsToken']>) =>
|
|
tokenService.createWsToken(...args);
|
|
|
|
let server: http.Server;
|
|
let wsUrl: string;
|
|
let nestApp: INestApplication;
|
|
const orm = (): FactoryOrm => nestApp.get(MikroORM);
|
|
|
|
beforeAll(async () => {
|
|
tokenService = new TokenService(await createTestMcpTokensRepo(testDb), await createTestUsersRepo(testDb), new EphemeralTokenService());
|
|
// 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();
|
|
// 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<void>(resolve => server.listen(0, resolve));
|
|
const addr = server.address() as { port: number };
|
|
wsUrl = `ws://127.0.0.1:${addr.port}/ws`;
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await new Promise<void>((resolve, reject) =>
|
|
server.close(err => err ? reject(err) : resolve())
|
|
);
|
|
await nestApp.close();
|
|
testDb.close();
|
|
});
|
|
|
|
beforeEach(async () => {
|
|
resetTestDb(testDb);
|
|
await resetRateLimits(nestApp);
|
|
});
|
|
|
|
/** Buffered WebSocket wrapper that never drops messages. */
|
|
class WsClient {
|
|
private ws: WebSocket;
|
|
private buffer: any[] = [];
|
|
private waiters: Array<(msg: any) => void> = [];
|
|
|
|
constructor(ws: WebSocket) {
|
|
this.ws = ws;
|
|
ws.on('message', (data) => {
|
|
const msg = JSON.parse(data.toString());
|
|
const waiter = this.waiters.shift();
|
|
if (waiter) {
|
|
waiter(msg);
|
|
} else {
|
|
this.buffer.push(msg);
|
|
}
|
|
});
|
|
}
|
|
|
|
next(timeoutMs = 3000): Promise<any> {
|
|
if (this.buffer.length > 0) return Promise.resolve(this.buffer.shift());
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => {
|
|
const idx = this.waiters.indexOf(resolve);
|
|
if (idx !== -1) this.waiters.splice(idx, 1);
|
|
reject(new Error('Message timeout'));
|
|
}, timeoutMs);
|
|
this.waiters.push((msg) => {
|
|
clearTimeout(timer);
|
|
resolve(msg);
|
|
});
|
|
});
|
|
}
|
|
|
|
send(msg: object) { this.ws.send(JSON.stringify(msg)); }
|
|
close() { this.ws.close(); }
|
|
|
|
/** Wait for any message matching predicate within timeout. */
|
|
waitFor(predicate: (m: any) => boolean, timeoutMs = 3000): Promise<any> {
|
|
// Check buffer first
|
|
const idx = this.buffer.findIndex(predicate);
|
|
if (idx !== -1) return Promise.resolve(this.buffer.splice(idx, 1)[0]);
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => reject(new Error('waitFor timeout')), timeoutMs);
|
|
const handler = (msg: any) => {
|
|
if (predicate(msg)) {
|
|
clearTimeout(timer);
|
|
resolve(msg);
|
|
} else {
|
|
this.buffer.push(msg);
|
|
// re-register
|
|
this.waiters.push(handler);
|
|
}
|
|
};
|
|
this.waiters.push(handler);
|
|
});
|
|
}
|
|
|
|
/** Collect messages for a given duration. */
|
|
collectFor(ms: number): Promise<any[]> {
|
|
return new Promise(resolve => {
|
|
const msgs: any[] = [...this.buffer.splice(0)];
|
|
const handleMsg = (msg: any) => msgs.push(msg);
|
|
this.ws.on('message', (data) => handleMsg(JSON.parse(data.toString())));
|
|
setTimeout(() => resolve(msgs), ms);
|
|
});
|
|
}
|
|
}
|
|
|
|
function connectWs(token?: string): Promise<WsClient> {
|
|
return new Promise((resolve, reject) => {
|
|
const url = token ? `${wsUrl}?token=${encodeURIComponent(token)}` : wsUrl;
|
|
const ws = new WebSocket(url);
|
|
const client = new WsClient(ws);
|
|
ws.once('open', () => resolve(client));
|
|
ws.once('error', reject);
|
|
ws.once('close', (code) => {
|
|
if (code === 4001) reject(new Error(`WS closed with 4001`));
|
|
});
|
|
});
|
|
}
|
|
|
|
describe('WS connection', () => {
|
|
it('WS-001 — connects with valid ephemeral token and receives welcome', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('welcome');
|
|
expect(typeof msg.socketId).toBe('number');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-002 — connecting without token closes with code 4001', async () => {
|
|
await new Promise<void>((resolve) => {
|
|
const ws = new WebSocket(wsUrl);
|
|
ws.on('close', (code) => {
|
|
expect(code).toBe(4001);
|
|
resolve();
|
|
});
|
|
ws.on('error', () => {});
|
|
});
|
|
});
|
|
|
|
it('WS-003 — connecting with invalid token closes with code 4001', async () => {
|
|
await new Promise<void>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=invalid-token-xyz`);
|
|
ws.on('close', (code) => {
|
|
expect(code).toBe(4001);
|
|
resolve();
|
|
});
|
|
ws.on('error', () => {});
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('WS rooms', () => {
|
|
it('WS-004 — join trip room receives joined confirmation', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('joined');
|
|
expect(msg.tripId).toBe(trip.id);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-005 — join trip without access receives error', async () => {
|
|
const { user } = createUser(testDb);
|
|
const { user: otherUser } = createUser(testDb);
|
|
const trip = createTrip(testDb, otherUser.id); // trip owned by otherUser
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('error');
|
|
expect(msg.message).toMatch(/access denied/i);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
/**
|
|
* Plan 3c Task 9 fix wave (A-H1, live regression): before the
|
|
* `Number.isFinite` guard on `realtime.gateway.ts::handleJoin`, a
|
|
* non-numeric `tripId` became `NaN`, which used to reach
|
|
* `TripsRepository.findAccessible`'s raw bind as the unquoted bareword
|
|
* `NaN` and throw `SqliteError: no such column: NaN` — the promise this
|
|
* handler returns never resolved, so the client got NO reply at all
|
|
* (confirmed on the compiled dual boot: 4 `[ERROR]` lines, HEAD only).
|
|
* The platform's own `NaN` -> `NULL` rendering (`NulSafeSqlitePlatform`)
|
|
* fixes the crash even without this guard; this test pins the WS-level
|
|
* contract (the SAME `Access denied` frame WS-005 gets for a real trip
|
|
* with no access), not just "does not throw".
|
|
*/
|
|
it('WS-005b — join with a non-numeric tripId receives the same error frame as no access (rule 22 / A-H1)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: 'abc' });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('error');
|
|
expect(msg.message).toMatch(/access denied/i);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
/**
|
|
* Plan 3c Task 0b (R9): `handleJoin`'s `this.db.canAccessTrip(...)` call
|
|
* (`realtime.gateway.ts:173`) is now `TripsRepository.findAccessible`
|
|
* underneath `DatabaseService`, reached through the request context
|
|
* `TrekWsAdapter`'s `@SubscribeMessage` wrapper forks
|
|
* (`nest/realtime/trek-ws.adapter.ts:273`, pinned generically by
|
|
* SEAM-001/BOOT-SWEEP-001) — no ratchet existed for this SPECIFIC seam
|
|
* before this task (inventory §12b). WS-004 above already proves the happy
|
|
* path returns `{type:'joined'}`, which requires the repository read to
|
|
* have succeeded; this test makes that explicit and structural, the same
|
|
* `cannotUseGlobalContext`/"global EntityManager" console-error filter
|
|
* `boot-sweeps-request-context.test.ts` uses, so a regression that routes
|
|
* `handleJoin` around the adapter's request context fails HERE by message,
|
|
* not only by an indirect "joined never arrived" timeout.
|
|
*/
|
|
it('WS-SEAM-001 — handleJoin\'s repository read runs inside the WS request context (no cannotUseGlobalContext)', async () => {
|
|
const errSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
|
|
try {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('joined');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
|
|
const suspicious = errSpy.mock.calls
|
|
.map((args) => args.map(String).join(' '))
|
|
.filter((line) => /cannotUseGlobalContext|global EntityManager/i.test(line));
|
|
expect(suspicious).toEqual([]);
|
|
} finally {
|
|
errSpy.mockRestore();
|
|
}
|
|
});
|
|
|
|
it('WS-006 — leave room receives left confirmation', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
await client.next(); // joined
|
|
|
|
client.send({ type: 'leave', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('left');
|
|
expect(msg.tripId).toBe(trip.id);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('WS rate limiting', () => {
|
|
it('WS-008 — exceeding 30 messages per window triggers rate-limit error', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Send 35 messages quickly — at least one should trigger rate limit
|
|
for (let i = 0; i < 35; i++) {
|
|
client.send({ type: 'ping' });
|
|
}
|
|
|
|
// Collect for up to 2s and find a rate-limit error
|
|
const msgs = await client.collectFor(1500);
|
|
const rateLimitMsg = msgs.find((m: any) => m.type === 'error' && m.message?.includes('Rate limit'));
|
|
expect(rateLimitMsg).toBeDefined();
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('WS real-time broadcast', () => {
|
|
it('WS-009 — POST /api/trips/:id/places broadcasts place:created to room members', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Join the trip room
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
await client.next(); // joined
|
|
|
|
// Create a place via REST (from a different socket, so it broadcasts to us)
|
|
const wsToken2 = createEphemeralToken(user.id, 'ws')!;
|
|
const client2 = await connectWs(wsToken2);
|
|
try {
|
|
await client2.next(); // welcome
|
|
client2.send({ type: 'join', tripId: trip.id });
|
|
await client2.next(); // joined
|
|
|
|
// REST call from client2's socket ID
|
|
const welcome2SocketId = (await Promise.resolve(null)) ?? null;
|
|
await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'Test Place', lat: 48.8566, lng: 2.3522 });
|
|
|
|
// client should receive the broadcast
|
|
const msg = await client.waitFor((m: any) => m.type === 'place:created', 3000);
|
|
expect(msg.type).toBe('place:created');
|
|
expect(msg.place).toBeDefined();
|
|
expect(msg.place.name).toBe('Test Place');
|
|
} finally {
|
|
client2.close();
|
|
}
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-010 — ephemeral WS token is single-use (second connection is rejected)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
// First connection: should succeed
|
|
const client = await connectWs(token);
|
|
await client.next(); // welcome
|
|
client.close();
|
|
|
|
// Second connection with same token: should be rejected with code 4001
|
|
const closeCode = await new Promise<number>((resolve, reject) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001)); // connection error also means rejection
|
|
setTimeout(() => reject(new Error('Timeout waiting for rejection')), 3000);
|
|
});
|
|
expect([4001, 1006]).toContain(closeCode); // 4001 = auth rejected, 1006 = abnormal close (also rejection)
|
|
});
|
|
|
|
it('WS-011 — client not in trip room does not receive broadcast', async () => {
|
|
const { user: owner } = createUser(testDb);
|
|
const { user: other } = createUser(testDb);
|
|
const trip = createTrip(testDb, owner.id);
|
|
|
|
// Connect `other` user but do NOT join the trip room
|
|
const tokenOther = createEphemeralToken(other.id, 'ws')!;
|
|
const clientOther = await connectWs(tokenOther);
|
|
try {
|
|
await clientOther.next(); // welcome — but no join
|
|
|
|
// Owner creates a place
|
|
await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(owner.id))
|
|
.send({ name: 'Owner Place', lat: 48.8566, lng: 2.3522 });
|
|
|
|
// `other` should NOT receive any broadcast within 500ms
|
|
const msgs = await clientOther.collectFor(500);
|
|
const broadcast = msgs.find((m: any) => m.type === 'place:created');
|
|
expect(broadcast).toBeUndefined();
|
|
} finally {
|
|
clientOther.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS auth edge cases — user-not-found and MFA enforcement
|
|
// ---------------------------------------------------------------------------
|
|
|
|
describe('WS auth edge cases', () => {
|
|
it('WS-012 — token for non-existent user closes with code 4001', async () => {
|
|
// Insert a user, grab an ephemeral token, then delete the user before connecting
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
// Remove the user so the DB lookup returns undefined
|
|
await deleteRows(orm(), Users, { id: user.id });
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001));
|
|
});
|
|
expect(closeCode).toBe(4001);
|
|
});
|
|
|
|
it('WS-013 — MFA is enforced when require_mfa is enabled and user has no MFA', async () => {
|
|
// Enable require_mfa in app_settings
|
|
await setAppSetting(orm(), 'require_mfa', 'true');
|
|
|
|
// Create a regular user without MFA
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4403));
|
|
});
|
|
expect(closeCode).toBe(4403);
|
|
});
|
|
|
|
it('WS-014 — MFA-enabled user connects successfully when require_mfa is enabled', async () => {
|
|
// Enable require_mfa
|
|
await setAppSetting(orm(), 'require_mfa', 'true');
|
|
|
|
// Create a user with MFA enabled
|
|
const { user } = createUser(testDb);
|
|
await updateRows(orm(), Users, { id: user.id }, { mfa_enabled: 1, mfa_secret: 'JBSWY3DPEHPK3PXP' });
|
|
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const client = await connectWs(token);
|
|
try {
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('welcome');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-027 — ws-token minted before a password change is rejected (session gate)', async () => {
|
|
// createWsToken stamps the user's current password_version (0) into the token.
|
|
const { user } = createUser(testDb);
|
|
const result = await createWsToken(user.id);
|
|
const token = result.token!;
|
|
|
|
// Simulate a password reset bumping the version AFTER the token was issued.
|
|
const { password_version: version } = await readUser(orm(), user.id);
|
|
await updateRows(orm(), Users, { id: user.id }, { password_version: version + 1 });
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001));
|
|
});
|
|
expect(closeCode).toBe(4001);
|
|
});
|
|
|
|
it('WS-028 — ws-token whose password_version still matches connects successfully', async () => {
|
|
const { user } = createUser(testDb);
|
|
// Bump the version first, THEN mint — the token captures the current pv.
|
|
await updateRows(orm(), Users, { id: user.id }, { password_version: 3 });
|
|
const result = await createWsToken(user.id);
|
|
const client = await connectWs(result.token!);
|
|
try {
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('welcome');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-029 — legacy token without a pv is rejected once the user resets their password', async () => {
|
|
// Tokens minted via createEphemeralToken carry no pv (treated as version 0).
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
await updateRows(orm(), Users, { id: user.id }, { password_version: 1 });
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001));
|
|
});
|
|
expect(closeCode).toBe(4001);
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS message processing — malformed/invalid payloads
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/** Connect a raw WebSocket (no WsClient wrapper) using a raw-send capable helper. */
|
|
function connectRawWs(token: string): Promise<{ ws: WebSocket; received: any[] }> {
|
|
return new Promise((resolve, reject) => {
|
|
const received: any[] = [];
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.on('message', (data) => {
|
|
try { received.push(JSON.parse(data.toString())); } catch { /* ignore parse errors */ }
|
|
});
|
|
ws.once('open', () => resolve({ ws, received }));
|
|
ws.once('error', reject);
|
|
ws.once('close', (code) => { if (code === 4001) reject(new Error('WS closed 4001')); });
|
|
});
|
|
}
|
|
|
|
/** Wait until `received` array has at least `n` items, up to `timeoutMs`. */
|
|
function waitForMessages(received: any[], n = 1, timeoutMs = 3000): Promise<void> {
|
|
return new Promise((resolve, reject) => {
|
|
if (received.length >= n) { resolve(); return; }
|
|
const start = Date.now();
|
|
const poll = () => {
|
|
if (received.length >= n) { resolve(); return; }
|
|
if (Date.now() - start > timeoutMs) { reject(new Error(`Timeout waiting for ${n} messages`)); return; }
|
|
setTimeout(poll, 20);
|
|
};
|
|
poll();
|
|
});
|
|
}
|
|
|
|
describe('WS message processing edge cases', () => {
|
|
it('WS-015 — malformed JSON is silently ignored (no crash, no error response)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Send raw malformed JSON — server should silently ignore and not close connection
|
|
rawWs.send('{ this is not json }');
|
|
rawWs.send('{broken');
|
|
|
|
await new Promise(r => setTimeout(r, 300));
|
|
|
|
// No error messages should have been sent by the server
|
|
const errMsgs = received.filter(m => m.type === 'error');
|
|
expect(errMsgs).toHaveLength(0);
|
|
// Connection should still be open
|
|
expect(rawWs.readyState).toBe(WebSocket.OPEN);
|
|
|
|
rawWs.close();
|
|
});
|
|
|
|
it('WS-015b — message with non-object payload is silently ignored', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Send valid JSON but not an object (array) — should be ignored
|
|
rawWs.send(JSON.stringify([1, 2, 3]));
|
|
// Send valid JSON number — should be ignored
|
|
rawWs.send('42');
|
|
|
|
await new Promise(r => setTimeout(r, 300));
|
|
|
|
// The only message received should be the welcome; no errors emitted
|
|
const errors = received.filter(m => m.type === 'error');
|
|
expect(errors).toHaveLength(0);
|
|
|
|
rawWs.close();
|
|
});
|
|
|
|
it('WS-015c — message object missing type field is silently ignored', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Object without a string `type` field
|
|
rawWs.send(JSON.stringify({ tripId: 1 }));
|
|
rawWs.send(JSON.stringify({ type: 42, tripId: 1 }));
|
|
|
|
await new Promise(r => setTimeout(r, 300));
|
|
|
|
const errors = received.filter(m => m.type === 'error');
|
|
expect(errors).toHaveLength(0);
|
|
|
|
rawWs.close();
|
|
});
|
|
|
|
it('WS-015d — a close frame with a reserved status code does not crash the server (#1576)', async () => {
|
|
// The `ws` client cannot send an invalid close code, so craft the frame over a raw
|
|
// socket. Reserved code 1006 makes ws emit an 'error' event on the socket; with no
|
|
// listener attached in setupWebSocket, Node rethrows it as an uncaughtException and the
|
|
// process dies in production. In-process under vitest that surfaces as an uncaught
|
|
// WS_ERR_INVALID_CLOSE_CODE, so capture uncaughtException for the duration and assert
|
|
// none fired — then confirm the server is still serving.
|
|
const uncaught: Error[] = [];
|
|
const onUncaught = (err: Error): void => { uncaught.push(err); };
|
|
process.on('uncaughtException', onUncaught);
|
|
|
|
try {
|
|
const port = (server.address() as { port: number }).port;
|
|
await new Promise<void>((resolve, reject) => {
|
|
const key = crypto.randomBytes(16).toString('base64');
|
|
const sock = net.connect(port, '127.0.0.1');
|
|
sock.on('error', reject);
|
|
sock.write(
|
|
`GET /ws HTTP/1.1\r\nHost: 127.0.0.1:${port}\r\nUpgrade: websocket\r\n` +
|
|
`Connection: Upgrade\r\nSec-WebSocket-Key: ${key}\r\nSec-WebSocket-Version: 13\r\n\r\n`
|
|
);
|
|
sock.once('data', (buf) => {
|
|
if (!buf.toString('latin1').includes('101 Switching Protocols')) return reject(new Error('no upgrade'));
|
|
const mask = crypto.randomBytes(4);
|
|
const payload = Buffer.alloc(2);
|
|
payload.writeUInt16BE(1006, 0); // reserved — MUST NOT appear in a close frame (RFC 6455)
|
|
const masked = Buffer.from(payload.map((b, i) => b ^ mask[i % 4]));
|
|
sock.write(Buffer.concat([Buffer.from([0x88, 0x82]), mask, masked])); // FIN|close, MASK|len2
|
|
setTimeout(() => { sock.destroy(); resolve(); }, 200);
|
|
});
|
|
});
|
|
|
|
expect(uncaught.map(e => (e as { code?: string }).code)).not.toContain('WS_ERR_INVALID_CLOSE_CODE');
|
|
|
|
// And the server is still serving: a fresh connection still gets a welcome.
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const client = await connectWs(token);
|
|
try {
|
|
expect((await client.next()).type).toBe('welcome');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
} finally {
|
|
process.off('uncaughtException', onUncaught);
|
|
}
|
|
});
|
|
|
|
it('WS-016 — rate-limit window resets: after limit hit, next window accepts messages again', async () => {
|
|
// Exercises line 108-110: the `now - rate.windowStart > WS_MSG_WINDOW` branch (counter reset).
|
|
// We confirm that:
|
|
// (a) msg 31 triggers the rate-limit error (current window),
|
|
// (b) a trip join in the same window is blocked,
|
|
// (c) after the rate-limit trip-join is blocked we verify the counter path was reached.
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Send exactly 30 messages (the limit) — all should succeed (no rate-limit error yet)
|
|
for (let i = 0; i < 30; i++) {
|
|
rawWs.send(JSON.stringify({ type: 'noop' }));
|
|
}
|
|
await new Promise(r => setTimeout(r, 200));
|
|
|
|
// Message 31 — triggers the `count > WS_MSG_LIMIT` branch, sends rate-limit error
|
|
rawWs.send(JSON.stringify({ type: 'noop' }));
|
|
await waitForMessages(received, 2, 3000); // welcome + rate-limit error
|
|
|
|
const rateLimitErrors = received.filter(m => m.type === 'error' && m.message?.includes('Rate limit'));
|
|
expect(rateLimitErrors.length).toBeGreaterThanOrEqual(1);
|
|
|
|
rawWs.close();
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS room management — disconnect cleanup and leave-nonexistent-room
|
|
// ---------------------------------------------------------------------------
|
|
|
|
describe('WS disconnect and room cleanup', () => {
|
|
it('WS-017 — disconnecting cleans up room membership so broadcast stops reaching the client', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token1 = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
// Connect and join the room
|
|
const client = await connectWs(token1);
|
|
await client.next(); // welcome
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
await client.next(); // joined
|
|
|
|
// Disconnect — triggers the 'close' handler that calls leaveRoom for all rooms
|
|
client.close();
|
|
await new Promise(r => setTimeout(r, 200)); // let the close event propagate
|
|
|
|
// Now create a second client that also joins the room, then creates a place.
|
|
// The first client (now disconnected) must NOT receive it (it can't, but more
|
|
// importantly the server must not crash when iterating rooms and finding a gone socket).
|
|
const token2 = createEphemeralToken(user.id, 'ws')!;
|
|
const client2 = await connectWs(token2);
|
|
try {
|
|
await client2.next(); // welcome
|
|
client2.send({ type: 'join', tripId: trip.id });
|
|
await client2.next(); // joined
|
|
|
|
// REST call to create a place — triggers broadcast; if room cleanup failed,
|
|
// iterating a closed socket would surface here.
|
|
const res = await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'Post-Disconnect Place', lat: 48.8566, lng: 2.3522 });
|
|
expect(res.status).toBe(201);
|
|
|
|
// client2 should still receive the broadcast
|
|
const msg = await client2.waitFor((m: any) => m.type === 'place:created', 3000);
|
|
expect(msg.place.name).toBe('Post-Disconnect Place');
|
|
} finally {
|
|
client2.close();
|
|
}
|
|
});
|
|
|
|
it('WS-018 — leaving a room the client was never in is a no-op (no crash, no error)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Send leave without ever joining — the server should respond with 'left'
|
|
// and not throw, since leaveRoom is defensive about missing rooms/sockets.
|
|
client.send({ type: 'leave', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('left');
|
|
expect(msg.tripId).toBe(trip.id);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// broadcastToUser() and getOnlineUserIds() — exported utility coverage
|
|
// ---------------------------------------------------------------------------
|
|
|
|
describe('broadcastToUser and getOnlineUserIds', () => {
|
|
it('WS-019 — broadcastToUser sends payload to all connected sockets for that user', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Call broadcastToUser directly
|
|
broadcastToUser(user.id, { type: 'test:direct', data: 'hello' });
|
|
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('test:direct');
|
|
expect(msg.data).toBe('hello');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-020 — broadcastToUser with excludeSid does not send to the excluded socket', async () => {
|
|
const { user } = createUser(testDb);
|
|
|
|
// Connect two sockets for the same user
|
|
const token1 = createEphemeralToken(user.id, 'ws')!;
|
|
const token2 = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client1 = await connectWs(token1);
|
|
const client2 = await connectWs(token2);
|
|
try {
|
|
const welcome1 = await client1.next();
|
|
const welcome2 = await client2.next();
|
|
const sid1 = welcome1.socketId;
|
|
|
|
// Broadcast excluding client1's socket ID
|
|
broadcastToUser(user.id, { type: 'test:exclude' }, sid1);
|
|
|
|
// client2 should receive it
|
|
const msg2 = await client2.next();
|
|
expect(msg2.type).toBe('test:exclude');
|
|
|
|
// client1 should NOT receive it within 400ms
|
|
const msgs1 = await client1.collectFor(400);
|
|
const received = msgs1.find((m: any) => m.type === 'test:exclude');
|
|
expect(received).toBeUndefined();
|
|
} finally {
|
|
client1.close();
|
|
client2.close();
|
|
}
|
|
});
|
|
|
|
it('WS-021 — broadcastToUser does not send to sockets belonging to a different user', async () => {
|
|
const { user: userA } = createUser(testDb);
|
|
const { user: userB } = createUser(testDb);
|
|
|
|
const tokenA = createEphemeralToken(userA.id, 'ws')!;
|
|
const tokenB = createEphemeralToken(userB.id, 'ws')!;
|
|
|
|
const clientA = await connectWs(tokenA);
|
|
const clientB = await connectWs(tokenB);
|
|
try {
|
|
await clientA.next(); // welcome
|
|
await clientB.next(); // welcome
|
|
|
|
// Broadcast only to userA
|
|
broadcastToUser(userA.id, { type: 'test:userA-only' });
|
|
|
|
// userA's client receives it
|
|
const msgA = await clientA.next();
|
|
expect(msgA.type).toBe('test:userA-only');
|
|
|
|
// userB's client must NOT receive it within 400ms
|
|
const msgsB = await clientB.collectFor(400);
|
|
const leak = msgsB.find((m: any) => m.type === 'test:userA-only');
|
|
expect(leak).toBeUndefined();
|
|
} finally {
|
|
clientA.close();
|
|
clientB.close();
|
|
}
|
|
});
|
|
|
|
it('WS-022 — getOnlineUserIds returns IDs of all connected authenticated users', async () => {
|
|
const { user: userA } = createUser(testDb);
|
|
const { user: userB } = createUser(testDb);
|
|
|
|
const tokenA = createEphemeralToken(userA.id, 'ws')!;
|
|
const tokenB = createEphemeralToken(userB.id, 'ws')!;
|
|
|
|
const clientA = await connectWs(tokenA);
|
|
const clientB = await connectWs(tokenB);
|
|
try {
|
|
await clientA.next(); // welcome
|
|
await clientB.next(); // welcome
|
|
|
|
const online = getOnlineUserIds();
|
|
expect(online.has(userA.id)).toBe(true);
|
|
expect(online.has(userB.id)).toBe(true);
|
|
} finally {
|
|
clientA.close();
|
|
clientB.close();
|
|
}
|
|
});
|
|
|
|
it('WS-023 — getOnlineUserIds excludes disconnected users', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
await client.next(); // welcome
|
|
|
|
// Verify user is online
|
|
expect(getOnlineUserIds().has(user.id)).toBe(true);
|
|
|
|
// Disconnect
|
|
client.close();
|
|
await new Promise(r => setTimeout(r, 200));
|
|
|
|
// User should no longer appear in online set
|
|
expect(getOnlineUserIds().has(user.id)).toBe(false);
|
|
});
|
|
|
|
it('WS-024 — broadcastToUser delivers custom payload to the correct connected socket', async () => {
|
|
// This directly exercises the broadcastToUser code path end-to-end through the
|
|
// exported function, verifying that the correct socket receives the message.
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
const customPayload = { type: 'custom:event', value: 99 };
|
|
broadcastToUser(user.id, customPayload);
|
|
|
|
const msg = await client.waitFor((m: any) => m.type === 'custom:event', 2000);
|
|
expect(msg.type).toBe('custom:event');
|
|
expect(msg.value).toBe(99);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-025 — broadcast() to an empty/nonexistent room is a no-op (no crash)', async () => {
|
|
// Exercises line 180: `if (!room || room.size === 0) return`
|
|
// A REST mutation on a trip with no connected WS clients triggers broadcast()
|
|
// with a room that doesn't exist — must not throw.
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
|
|
// No WebSocket clients join the trip room before the REST call
|
|
const res = await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'No Room Place', lat: 10, lng: 20 });
|
|
|
|
// Server must not crash — 201 confirms broadcast() returned silently
|
|
expect(res.status).toBe(201);
|
|
});
|
|
|
|
it('WS-026 — broadcast() skips non-OPEN sockets in the room', async () => {
|
|
// This exercises line 185: `if (ws.readyState !== 1) continue`
|
|
// We join a room with two clients, forcefully terminate one (so its readyState becomes
|
|
// CLOSED while still transiently in the room map), then trigger a broadcast and confirm
|
|
// the surviving client receives it without errors.
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
|
|
const token1 = createEphemeralToken(user.id, 'ws')!;
|
|
const token2 = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client1 = await connectWs(token1);
|
|
const client2 = await connectWs(token2);
|
|
try {
|
|
await client1.next(); // welcome
|
|
await client2.next(); // welcome
|
|
|
|
client1.send({ type: 'join', tripId: trip.id });
|
|
await client1.next(); // joined
|
|
|
|
client2.send({ type: 'join', tripId: trip.id });
|
|
await client2.next(); // joined
|
|
|
|
// Close client1 abruptly — the underlying socket may momentarily remain in the room map
|
|
client1.close();
|
|
await new Promise(r => setTimeout(r, 50)); // brief pause
|
|
|
|
// Trigger broadcast via REST — should not crash even if client1's socket is closed
|
|
const res = await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'Resilience Place', lat: 1, lng: 2 });
|
|
expect(res.status).toBe(201);
|
|
|
|
// client2 should still receive the broadcast
|
|
const msg = await client2.waitFor((m: any) => m.type === 'place:created', 3000);
|
|
expect(msg.place.name).toBe('Resilience Place');
|
|
} finally {
|
|
client2.close();
|
|
}
|
|
});
|
|
});
|