From 0bb02c40140d048f3874919273589f76bd1a1854 Mon Sep 17 00:00:00 2001 From: Jonathan Hinkel Date: Thu, 27 Aug 2026 11:36:22 +0200 Subject: [PATCH] Validate realtime browser events --- docs/REALTIME_SESSION.md | 2 + src/lib/core/protocol.ts | 39 ++++++++++++++ src/lib/server/core/session.spec.ts | 81 ++++++++++++++++++++++++++++- src/lib/server/core/session.ts | 59 +++++++++++++++++++-- 4 files changed, 175 insertions(+), 6 deletions(-) diff --git a/docs/REALTIME_SESSION.md b/docs/REALTIME_SESSION.md index dcae9b4..b9e00f9 100644 --- a/docs/REALTIME_SESSION.md +++ b/docs/REALTIME_SESSION.md @@ -104,6 +104,8 @@ See [`protocol.ts`](../src/lib/core/protocol.ts) for the authoritative, current - **Client → server**: `session.start`, `input_audio.chunk`, `input_audio.commit`, `input_text.send`, `response.cancel`, `session.end`, `debug.subscribe` - **Server → client**: `session.ready`, `output_audio.chunk`, `output_audio.done`, `transcript.delta`, `transcript.done`, `response.done`, `error`, `session.closed`, `debug.subscribe.result`, `debug.log` +Parsed browser JSON is runtime-validated against the client-event contract before it reaches analytics or a provider adapter; the TypeScript union alone is not a wire-level guarantee. Missing or wrongly typed required fields are rejected with `invalid_client_event`, while an unexpected throw in the async message path closes only that connection with `internal_error`. This containment is process-critical: one Node process serves every realtime session. + `debug.subscribe`/`debug.subscribe.result`/`debug.log` are a separate concern from the rest of this vocabulary — ephemeral developer debug-log forwarding, not session content. See [LOGGING.md](LOGGING.md). ## OpenAI Realtime API gotcha (verified live, Aug 2026) diff --git a/src/lib/core/protocol.ts b/src/lib/core/protocol.ts index dd44fb5..8b56e35 100644 --- a/src/lib/core/protocol.ts +++ b/src/lib/core/protocol.ts @@ -40,6 +40,45 @@ export type ClientEvent = // DEBUG_ACCESS_KEY) and ignored in dev, where forwarding is automatic. | { type: 'debug.subscribe'; key?: string }; +/** Events accepted from the browser websocket (excludes server-internal control events). */ +export type BrowserClientEvent = Exclude; + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +/** + * Runtime validation for the browser websocket trust boundary. TypeScript's + * `ClientEvent` union disappears at runtime, so parsed JSON must be checked + * before it reaches analytics storage or a provider adapter. + */ +export function isBrowserClientEvent(value: unknown): value is BrowserClientEvent { + if (!isRecord(value)) return false; + + switch (value.type) { + case 'session.start': + return ( + value.config === undefined || + (isRecord(value.config) && + (value.config.voice === undefined || typeof value.config.voice === 'string')) + ); + case 'input_audio.chunk': + return typeof value.audio === 'string'; + case 'input_text.send': + return typeof value.text === 'string'; + case 'tool.result': + return typeof value.callId === 'string' && typeof value.output === 'string'; + case 'debug.subscribe': + return value.key === undefined || typeof value.key === 'string'; + case 'input_audio.commit': + case 'response.cancel': + case 'session.end': + return true; + default: + return false; + } +} + export type RealtimeErrorCode = | 'auth_expired' | 'invalid_client_event' diff --git a/src/lib/server/core/session.spec.ts b/src/lib/server/core/session.spec.ts index 8e0057f..39ed7ad 100644 --- a/src/lib/server/core/session.spec.ts +++ b/src/lib/server/core/session.spec.ts @@ -3,7 +3,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type WebSocket from 'ws'; import type { ClientEvent, ServerEvent } from '../../core/protocol.js'; import type { ToolDefinition } from '../../core/tools.js'; -import type { AnalyticsRecorder } from './analytics.js'; +import { SqliteAnalyticsRecorder, type AnalyticsRecorder } from './analytics.js'; import type { ProviderSession, RealtimeProvider } from './provider.js'; import { createRealtimeSession, resolveSessionConfig } from './session.js'; import { createToolRegistry } from './tools.js'; @@ -36,7 +36,7 @@ class FakeBrowserSocket extends EventEmitter { return this.sentRaw.map((raw) => JSON.parse(raw)); } - receive(event: ClientEvent) { + receive(event: unknown) { this.emit('message', Buffer.from(JSON.stringify(event)), false); } } @@ -210,6 +210,83 @@ describe('RealtimeSession', () => { ); }); + it('rejects input_text.send without text before it can violate SQLite constraints', async () => { + const provider = new FakeProvider(); + const recorder = new SqliteAnalyticsRecorder(':memory:', { + messages: true, + toolCalls: true, + events: true + }); + const { socket, asWs } = makeSocket(); + createRealtimeSession(asWs, provider, DEFAULTS, recorder, TEST_TOOL_TIMEOUT_MS); + + socket.receive({ type: 'session.start' }); + await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); + + socket.receive({ type: 'input_text.send' }); + await vi.waitFor(() => + expect(socket.sent).toContainEqual({ + type: 'error', + code: 'invalid_client_event', + message: 'Invalid client event.' + }) + ); + expect(provider.sessions[0].sent).not.toContainEqual( + expect.objectContaining({ type: 'input_text.send' }) + ); + expect(recorder.rawDb.prepare('SELECT * FROM messages').all()).toEqual([]); + + // The malformed frame is isolated: this same session remains usable. + socket.receive({ type: 'input_text.send', text: 'still alive' }); + await vi.waitFor(() => + expect(provider.sessions[0].sent).toContainEqual({ + type: 'input_text.send', + text: 'still alive' + }) + ); + expect(recorder.rawDb.prepare('SELECT text FROM messages').all()).toEqual([ + { text: 'still alive' } + ]); + + socket.emit('close'); + recorder.close(); + }); + + it('contains unexpected async message failures and leaves other sessions usable', async () => { + const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); + recorder.messageLogged = () => { + throw new Error('analytics unavailable'); + }; + const affected = makeSocket(); + const other = makeSocket(); + createRealtimeSession(affected.asWs, provider, DEFAULTS, recorder, TEST_TOOL_TIMEOUT_MS); + createRealtimeSession(other.asWs, provider, DEFAULTS, recorder, TEST_TOOL_TIMEOUT_MS); + + affected.socket.receive({ type: 'session.start' }); + await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); + other.socket.receive({ type: 'session.start' }); + await vi.waitFor(() => expect(provider.sessions).toHaveLength(2)); + affected.socket.receive({ type: 'input_text.send', text: 'hello' }); + + await vi.waitFor(() => + expect(affected.socket.sent).toContainEqual({ + type: 'error', + code: 'internal_error', + message: 'Failed to handle client event.' + }) + ); + expect(provider.sessions[0].closed).toBe(true); + expect(provider.sessions[0].closeReason).toBe('internal_error'); + expect(affected.socket.closeCalls).toBe(1); + expect(provider.sessions[1].closed).toBe(false); + + other.socket.receive({ type: 'response.cancel' }); + await vi.waitFor(() => + expect(provider.sessions[1].sent).toContainEqual({ type: 'response.cancel' }) + ); + }); + it('relays provider server events back to the browser socket', async () => { const provider = new FakeProvider(); const recorder = new FakeAnalyticsRecorder(); diff --git a/src/lib/server/core/session.ts b/src/lib/server/core/session.ts index b6a4c07..d0236b4 100644 --- a/src/lib/server/core/session.ts +++ b/src/lib/server/core/session.ts @@ -1,7 +1,13 @@ import { randomUUID } from 'node:crypto'; import WebSocket from 'ws'; import type { LogEntry, Logger } from '../../core/logger.js'; -import type { ClientEvent, ServerEvent, SessionStartConfig } from '../../core/protocol.js'; +import { + isBrowserClientEvent, + type BrowserClientEvent, + type ClientEvent, + type ServerEvent, + type SessionStartConfig +} from '../../core/protocol.js'; import { formatToolError } from '../../core/tools.js'; import type { AnalyticsRecorder } from './analytics.js'; import { createLogger, safeSerialize, verifyDebugAccessKey } from './logger.js'; @@ -79,7 +85,9 @@ export class RealtimeSession { this.debugSubscribed = isDev; this.logger = createLogger((entry) => this.forwardDebugLog(entry), { logMessages }); this.browserSocket.on('message', (data, isBinary) => { - void this.handleBrowserMessage(data, isBinary); + void this.handleBrowserMessage(data, isBinary).catch((error) => { + this.handleBrowserMessageFailure(error); + }); }); this.browserSocket.on('close', () => this.close('client_disconnected')); this.browserSocket.on('error', () => this.close('client_disconnected')); @@ -100,9 +108,9 @@ export class RealtimeSession { return; } - let event: ClientEvent; + let parsed: unknown; try { - event = JSON.parse(data.toString()); + parsed = JSON.parse(data.toString()); } catch { this.logger.warn('ws', 'Rejected malformed JSON from browser.'); this.sendToBrowser({ @@ -112,6 +120,16 @@ export class RealtimeSession { }); return; } + if (!isBrowserClientEvent(parsed)) { + this.logger.warn('ws', 'Rejected invalid client event.'); + this.sendToBrowser({ + type: 'error', + code: 'invalid_client_event', + message: 'Invalid client event.' + }); + return; + } + const event: BrowserClientEvent = parsed; if (event.type === 'session.start') { await this.start(event.config); @@ -156,6 +174,39 @@ export class RealtimeSession { this.providerSession.send(event); } + /** + * Last-resort containment for unexpected failures in the async websocket + * message path. A single connection must never create an unhandled + * rejection in the process shared by every realtime session. + */ + private handleBrowserMessageFailure(error: unknown): void { + let message = 'Unknown error'; + try { + message = error instanceof Error ? error.message : String(error); + } catch { + // Even a hostile thrown value with a broken toString stays contained. + } + try { + this.logger.error('ws', 'Unexpected failure handling browser message.', { message }); + } catch { + // Error reporting itself must not escape this rejection boundary. + } + try { + this.sendToBrowser({ + type: 'error', + code: 'internal_error', + message: 'Failed to handle client event.' + }); + } catch { + // A failing analytics/logging sink may be the original cause. + } + try { + this.close('internal_error'); + } catch { + // Keep the failure scoped to this connection even if teardown throws. + } + } + private async start(clientConfig?: SessionStartConfig): Promise { if (this.state !== 'idle') { this.sendToBrowser({