From 02302131fb5f8752136797e00c3b26f11b714cdb Mon Sep 17 00:00:00 2001 From: Demiurge The Single Date: Tue, 16 Jun 2026 19:15:25 +0300 Subject: [PATCH] fix(live-ws): bridge sidecar events to dashboard (#4004) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.27 — repair LiveWS sidecar (startup, same-origin /live-ws, main→sidecar compression.completed bridge, early-msg queue). Fixed the cookie-parse regex (\s) + added a focused unit test; baseline bumped for the non-blocking chatCore bridge. --- config/quality/file-size-baseline.json | 5 +- open-sse/handlers/chatCore.ts | 27 ++- package.json | 1 + scripts/start-ws-server.mjs | 46 +++-- src/hooks/useLiveDashboard.ts | 12 +- src/lib/db/compressionAnalytics.ts | 29 +++ src/server/ws/liveServer.ts | 260 +++++++++++++++++++++--- src/server/ws/types.ts | 6 +- tests/unit/cli/live-ws-startup.test.ts | 151 ++++++++++++++ tests/unit/live-ws-cookie-parse.test.ts | 42 ++++ 10 files changed, 525 insertions(+), 54 deletions(-) create mode 100644 tests/unit/cli/live-ws-startup.test.ts create mode 100644 tests/unit/live-ws-cookie-parse.test.ts diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 8049c36b7d..7031e19cda 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -36,7 +36,7 @@ "open-sse/executors/muse-spark-web.ts": 1284, "open-sse/executors/perplexity-web.ts": 1013, "open-sse/handlers/audioSpeech.ts": 965, - "open-sse/handlers/chatCore.ts": 5830, + "open-sse/handlers/chatCore.ts": 5851, "open-sse/handlers/imageGeneration.ts": 3777, "open-sse/handlers/responseSanitizer.ts": 1103, "open-sse/handlers/search.ts": 1442, @@ -156,5 +156,6 @@ "_rebaseline_2026_06_15_3941_provider_request_capture": "PR #3941 own growth: chatCore.ts 5823->5830 (+7 by check-file-size counting = run executor attempts inside the unified provider request capture scope), antigravity.ts 1649->1664 (+15) and codex.ts 1439->1447 (+8) = bridge hand-written upstream transports that bypass normal fetch/BaseExecutor capture. Cohesive logging-fidelity refactor; not extractable without hiding the actual dispatch boundary.", "_rebaseline_2026_06_16_3958_qwen_body_check": "PR #3958 own growth: validation.ts 4407->4428 (+21 = validateQwenWebProvider now parses the /api/v2/user 200 body and requires a real user object, since Qwen returns HTTP 200 even for invalid tokens — fixes the validation false-positive, #3931). Cohesive with the existing qwen-web validation branch; not separately extractable.", "_rebaseline_2026_06_16_4001_perplexity_diff_block": "PR #4001 own growth: perplexity-web.ts 939->1013 (+74 = parse the schematized API's RFC-6902 diff_block JSON-patch frames — applyMarkdownDiff + isAnswerTextUsage primary-usage lock + stop only on COMPLETED, not on a still-PENDING final flag — so streamed answers aren't empty, #3938 follow-up). Cohesive single-executor SSE-parsing logic; not separately extractable.", - "_rebaseline_2026_06_16_4005_openai_dynamic_models": "PR #4005 own growth: models/route.ts 2494->2512 (+18 = openai model-discovery derives {customBaseUrl}/v1/models from providerSpecificData.baseUrl, SSRF-guarded via safeOutboundFetch+public-only) and pricing.ts 1529->1581 (+52 = pure-data pricing rows closing $0 gaps for registry-exposed ids: openai gpt-5.4/-mini/-nano, gpt-4.1, gpt-4o-2024-11-20, o3 + codex(cx) gpt-5.4-{xhigh,high,medium,low}, gpt-5.3-codex-spark). Cohesive; pricing is data, route change mirrors the anthropic-compat discovery path." + "_rebaseline_2026_06_16_4005_openai_dynamic_models": "PR #4005 own growth: models/route.ts 2494->2512 (+18 = openai model-discovery derives {customBaseUrl}/v1/models from providerSpecificData.baseUrl, SSRF-guarded via safeOutboundFetch+public-only) and pricing.ts 1529->1581 (+52 = pure-data pricing rows closing $0 gaps for registry-exposed ids: openai gpt-5.4/-mini/-nano, gpt-4.1, gpt-4o-2024-11-20, o3 + codex(cx) gpt-5.4-{xhigh,high,medium,low}, gpt-5.3-codex-spark). Cohesive; pricing is data, route change mirrors the anthropic-compat discovery path.", + "_rebaseline_2026_06_16_4004_livews_bridge": "PR #4004 own growth: chatCore.ts 5830->5851 (+21 = forwardDashboardEventToLiveWs — a best-effort, non-blocking, timeout-bounded POST that bridges compression.completed events from the main process to the LiveWS sidecar so the dashboard updates under a reverse proxy). Cohesive fire-and-forget beacon at the existing compression emit site; not extractable. Structural shrink of chatCore.ts tracked in #3501." } diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index ed6dd48bc4..531b80382e 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -445,6 +445,24 @@ function extractMemoryTextFromRequestBody( return ""; } +async function forwardDashboardEventToLiveWs(event: string, payload: unknown): Promise { + const port = process.env.LIVE_WS_PORT || "20129"; + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 1_500); + try { + await fetch(`http://127.0.0.1:${port}/__omniroute_event`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ event, payload, timestamp: Date.now() }), + signal: controller.signal, + }); + } catch { + // Best-effort sidecar bridge; do not affect the chat hot path. + } finally { + clearTimeout(timeout); + } +} + async function maybeSyncClaudeExtraUsageState({ provider, connectionId, @@ -2654,7 +2672,7 @@ export async function handleChatCore({ // Guard: only emit when compression actually ran and produced stats. if (result.compressed && result.stats) { try { - emit("compression.completed", { + const compressionCompletedPayload = { requestId: traceId, comboId: result.stats.compressionComboId ?? null, mode, @@ -2665,7 +2683,12 @@ export async function handleChatCore({ validationWarnings: result.stats.validationWarnings, fallbackApplied: result.stats.fallbackApplied, timestamp: Date.now(), - }); + }; + emit("compression.completed", compressionCompletedPayload); + void forwardDashboardEventToLiveWs( + "compression.completed", + compressionCompletedPayload + ); } catch (_emitErr) { // never propagate into the hot path — but log like the sibling // fire-and-forget blocks so a throwing event bus isn't fully silent. diff --git a/package.json b/package.json index 01aef17a2f..785db2836f 100644 --- a/package.json +++ b/package.json @@ -244,6 +244,7 @@ "react-reconciler": "^0.33.0", "recharts": "^3.8.1", "selfsigned": "^5.5.0", + "socks": "^2.8.7", "sql.js": "^1.14.1", "sqlite-vec": "^0.1.9", "tsx": "^4.22.3", diff --git a/scripts/start-ws-server.mjs b/scripts/start-ws-server.mjs index 3cc768d789..88e48a9ced 100644 --- a/scripts/start-ws-server.mjs +++ b/scripts/start-ws-server.mjs @@ -4,15 +4,17 @@ * This script starts the live dashboard WebSocket server as a separate * process alongside the Next.js app. Run it with: * - * node scripts/start-ws-server.js - * - * Or use the built-in auto-start in src/server/ws/liveServer.ts. + * node scripts/start-ws-server.mjs * * Environment variables: * LIVE_WS_PORT — WebSocket server port (default: 20129) + * LIVE_WS_HOST — WebSocket server host (default: 127.0.0.1) * OMNIROUTE_DISABLE_LIVE_WS — Set to "1" or "true" to disable */ +import { spawnSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; + if ( process.env.OMNIROUTE_DISABLE_LIVE_WS === "1" || process.env.OMNIROUTE_DISABLE_LIVE_WS === "true" @@ -21,23 +23,39 @@ if ( process.exit(0); } -// Register tsx to handle TypeScript imports -import { register } from "node:module"; -import { pathToFileURL } from "node:url"; +const BOOTSTRAPPED_ENV = "OMNIROUTE_LIVE_WS_BOOTSTRAPPED"; -register("tsx", pathToFileURL("./")); +if (process.env[BOOTSTRAPPED_ENV] !== "1") { + const result = spawnSync(process.execPath, ["--import", "tsx", fileURLToPath(import.meta.url)], { + stdio: "inherit", + env: { + ...process.env, + [BOOTSTRAPPED_ENV]: "1", + // Prevent liveServer.ts from auto-starting on import; this script owns + // process startup so errors propagate to the supervisor/CLI caller. + OMNIROUTE_ENABLE_LIVE_WS: "0", + }, + }); -const { startLiveDashboardServer } = await import("../src/server/ws/liveServer"); + if (result.signal) { + process.kill(process.pid, result.signal); + } + + process.exit(result.status ?? 1); +} + +const { startLiveDashboardServer } = await import("../src/server/ws/liveServer.ts"); const port = parseInt(process.env.LIVE_WS_PORT || "20129", 10); +const host = process.env.LIVE_WS_HOST || "127.0.0.1"; -console.log(`[LiveWS] Starting dashboard WebSocket server on port ${port}...`); +console.log(`[LiveWS] Starting dashboard WebSocket server on ${host}:${port}...`); -startLiveDashboardServer(port) - .then((server) => { - console.log(`[LiveWS] Dashboard WebSocket server listening on ws://0.0.0.0:${port}`); - console.log(`[LiveWS] Connect via: ws://localhost:${port}?token=`); - console.log(`[LiveWS] Channels: requests, combo, credentials`); +startLiveDashboardServer(port, host) + .then(() => { + console.log(`[LiveWS] Dashboard WebSocket server listening on ws://${host}:${port}`); + console.log("[LiveWS] Connect via: ws://localhost:%d?token=", port); + console.log("[LiveWS] Channels: requests, combo, credentials"); }) .catch((err) => { console.error("[LiveWS] Failed to start:", err); diff --git a/src/hooks/useLiveDashboard.ts b/src/hooks/useLiveDashboard.ts index 4a51b4846b..e216b7a584 100644 --- a/src/hooks/useLiveDashboard.ts +++ b/src/hooks/useLiveDashboard.ts @@ -17,7 +17,17 @@ import type { DashboardChannel, DashboardEventName } from "@/lib/events/types"; // ── Config ──────────────────────────────────────────────────────────────── const WS_RECONNECT_DELAYS = [1000, 2000, 4000, 8000, 16000, 30000]; -const DEFAULT_WS_URL = `ws://${typeof window !== "undefined" ? window.location.hostname : "localhost"}:20129`; +function getDefaultWsUrl(): string { + if (typeof window === "undefined") return "ws://localhost:20129"; + const protocol = window.location.protocol === "https:" ? "wss:" : "ws:"; + const { hostname } = window.location; + if (hostname === "localhost" || hostname === "127.0.0.1") { + return `${protocol}//${hostname}:20129`; + } + return `${protocol}//${window.location.host}/live-ws`; +} + +const DEFAULT_WS_URL = getDefaultWsUrl(); // ── Types ───────────────────────────────────────────────────────────────── diff --git a/src/lib/db/compressionAnalytics.ts b/src/lib/db/compressionAnalytics.ts index 8a051bd8a6..2a8fc448e8 100644 --- a/src/lib/db/compressionAnalytics.ts +++ b/src/lib/db/compressionAnalytics.ts @@ -440,3 +440,32 @@ export function getCompressionAnalyticsSummary(since?: string): CompressionAnaly }, }; } + +export interface LatestCompressionAnalyticsRun { + id: number; + timestamp: string; + combo_id: string | null; + compression_combo_id: string | null; + mode: string; + original_tokens: number; + compressed_tokens: number; + tokens_saved: number; + duration_ms: number | null; + request_id: string | null; + engine: string | null; + validation_fallback: number | null; +} + +export function getLatestCompressionAnalyticsRun(): LatestCompressionAnalyticsRun | undefined { + const db = getDbInstance(); + return db + .prepare( + `SELECT id, timestamp, combo_id, compression_combo_id, mode, + original_tokens, compressed_tokens, tokens_saved, duration_ms, + request_id, engine, validation_fallback + FROM compression_analytics + ORDER BY timestamp DESC, id DESC + LIMIT 1` + ) + .get() as LatestCompressionAnalyticsRun | undefined; +} diff --git a/src/server/ws/liveServer.ts b/src/server/ws/liveServer.ts index 74f5e48fd8..e48b4a91b8 100644 --- a/src/server/ws/liveServer.ts +++ b/src/server/ws/liveServer.ts @@ -14,7 +14,8 @@ */ import { WebSocketServer, WebSocket } from "ws"; -import { createServer } from "http"; +import { jwtVerify } from "jose"; +import { createServer, type IncomingMessage, type ServerResponse } from "http"; import { randomUUID } from "crypto"; // ── Types ───────────────────────────────────────────────────────────────── @@ -38,6 +39,8 @@ const HEARTBEAT_INTERVAL_MS = 15_000; const HEARTBEAT_TIMEOUT_MS = 35_000; const MAX_CLIENTS = 500; const MAX_EVENTS_PER_SECOND = 100; +const MAX_PENDING_MESSAGES_PER_CLIENT = 32; +const MAX_PENDING_MESSAGE_BYTES = 16_384; /** * Origins allowed to open a WebSocket. Defaults to the loopback dashboard @@ -91,6 +94,20 @@ const BACKLOG_MAX = 500; // ── Auth ────────────────────────────────────────────────────────────────── +function toWebHeaders(headers: import("http").IncomingMessage["headers"]): Headers { + const webHeaders = new Headers(); + + for (const [name, value] of Object.entries(headers)) { + if (typeof value === "string") { + webHeaders.set(name, value); + } else if (Array.isArray(value)) { + webHeaders.set(name, name.toLowerCase() === "cookie" ? value.join("; ") : value.join(", ")); + } + } + + return webHeaders; +} + async function authorizeConnection(request: import("http").IncomingMessage): Promise { const sessionId = randomUUID().slice(0, 8); @@ -99,16 +116,25 @@ async function authorizeConnection(request: import("http").IncomingMessage): Pro // headers — a single screenshot of the URL bar exposes the API key. const token = extractBearerToken(request) || extractAltTokenHeader(request); + // Browser WebSocket clients cannot set custom Authorization headers. When + // LiveWS is exposed same-origin through a reverse proxy, accept the existing + // dashboard session cookie before falling back to API-key authentication. Keep + // the check local to this sidecar so it does not import Next.js-only modules. if (!token) { + if (await isDashboardCookieAuthenticated(request)) { + return { authorized: true, sessionId }; + } return { authorized: false, sessionId, error: "Missing token" }; } try { - // Validate API key via the existing auth system - const { extractApiKey, isValidApiKey } = await import("../services/auth"); - const apiKey = extractApiKey({ headers: { authorization: `Bearer ${token}` } } as any); + // Validate API key via the existing auth system. + const { extractApiKey, isValidApiKey } = await import("../../sse/services/auth.ts"); + const apiKey = extractApiKey({ headers: { authorization: `Bearer ${token}` } } as any, { + allowUrl: false, + }); - if (!apiKey || !isValidApiKey(apiKey)) { + if (!apiKey || !(await isValidApiKey(apiKey))) { return { authorized: false, sessionId, error: "Invalid API key" }; } @@ -124,6 +150,36 @@ function extractAltTokenHeader(request: import("http").IncomingMessage): string return typeof raw === "string" ? raw : null; } +export function getCookieValueFromHeader( + headers: import("http").IncomingHttpHeaders, + name: string +): string | null { + const raw = headers.cookie; + const cookieHeader = Array.isArray(raw) ? raw.join("; ") : raw; + if (!cookieHeader) return null; + const escaped = name.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); + // NOTE: \\s (not \s) — this is a plain template literal, so \s would collapse to a + // literal "s" and the pattern would only match auth_token when it is the FIRST cookie. + // Browsers serialize the Cookie header as "a=1; b=2", so the leading-cookie case + // (auth_token preceded by another cookie) must match too (#4004 same-origin proxy auth). + const match = cookieHeader.match(new RegExp(`(?:^|;\\s*)${escaped}=([^;]*)`)); + return match ? decodeURIComponent(match[1]) : null; +} + +async function isDashboardCookieAuthenticated( + request: import("http").IncomingMessage +): Promise { + const token = getCookieValueFromHeader(request.headers, "auth_token"); + if (!token || !process.env.JWT_SECRET) return false; + try { + const secret = new TextEncoder().encode(process.env.JWT_SECRET); + await jwtVerify(token, secret); + return true; + } catch { + return false; + } +} + function extractBearerToken(request: import("http").IncomingMessage): string | null { const auth = request.headers["authorization"]; if (!auth || typeof auth !== "string") return null; @@ -202,35 +258,148 @@ function sendTo(ws: WebSocket, msg: WsServerMessage | Record): // ── Event Bus → WebSocket Bridge ────────────────────────────────────────── +function publishDashboardEvent( + event: DashboardEventName, + payload: unknown, + timestamp = Date.now() +): boolean { + const channel = getChannelForEvent(event); + if (!channel) return false; + + // Store in backlog so clients that subscribe just after a run still receive it. + eventHistoryBacklog.push({ event, payload, timestamp }); + if (eventHistoryBacklog.length > BACKLOG_MAX) { + eventHistoryBacklog.shift(); + } + + const msg: WsEventMessage = { + type: "event", + channel, + event, + data: payload, + }; + + for (const [clientId, client] of clients) { + if (client.ws.readyState !== WebSocket.OPEN) { + clients.delete(clientId); + continue; + } + if (client.subscribedChannels.has(channel)) { + sendTo(client.ws, msg); + } + } + + return true; +} + function subscribeToEventBus(): () => void { return onAny((event: DashboardEventName, payload: unknown) => { - const channel = getChannelForEvent(event); - if (!channel) return; + publishDashboardEvent(event, payload); + }); +} - // Store in backlog - eventHistoryBacklog.push({ event, payload, timestamp: Date.now() }); - if (eventHistoryBacklog.length > BACKLOG_MAX) { - eventHistoryBacklog.shift(); - } +function isLoopbackRequest(req: IncomingMessage): boolean { + const addr = req.socket.remoteAddress; + return addr === "127.0.0.1" || addr === "::1" || addr === "::ffff:127.0.0.1"; +} - // Forward to subscribed clients - const msg: WsEventMessage = { - type: "event", - channel, - event, - data: payload, - }; +function handleInternalEventRequest(req: IncomingMessage, res: ServerResponse): void { + if (req.method !== "POST" || req.url !== "/__omniroute_event") { + res.writeHead(404).end(); + return; + } + if (!isLoopbackRequest(req)) { + res.writeHead(403, { "content-type": "application/json" }).end(JSON.stringify({ ok: false })); + return; + } - for (const [clientId, client] of clients) { - if (client.ws.readyState !== WebSocket.OPEN) { - clients.delete(clientId); - continue; - } - if (client.subscribedChannels.has(channel)) { - sendTo(client.ws, msg); - } + let body = ""; + req.setEncoding("utf8"); + req.on("data", (chunk) => { + body += chunk; + if (body.length > 1_000_000) { + req.destroy(new Error("Internal event payload too large")); } }); + req.on("error", () => { + if (!res.headersSent) res.writeHead(400).end(); + }); + req.on("end", () => { + try { + const parsed = JSON.parse(body || "{}"); + const event = parsed.event as DashboardEventName; + if (!Object.values(CHANNEL_EVENTS).some((events) => events.includes(event))) { + res + .writeHead(400, { "content-type": "application/json" }) + .end(JSON.stringify({ ok: false })); + return; + } + const ok = publishDashboardEvent( + event, + parsed.payload, + Number(parsed.timestamp) || Date.now() + ); + res + .writeHead(ok ? 202 : 400, { "content-type": "application/json" }) + .end(JSON.stringify({ ok })); + } catch { + res.writeHead(400, { "content-type": "application/json" }).end(JSON.stringify({ ok: false })); + } + }); +} + +async function seedLatestCompressionRunFromDb(): Promise { + try { + const { getLatestCompressionAnalyticsRun } = await import("@/lib/db/compressionAnalytics"); + const row = getLatestCompressionAnalyticsRun(); + if (!row) return; + + const originalTokens = Number(row.original_tokens) || 0; + const compressedTokens = Number(row.compressed_tokens) || 0; + const savingsPercent = + originalTokens > 0 + ? Math.round(((originalTokens - compressedTokens) / originalTokens) * 100) + : 0; + const timestamp = Number.isFinite(Date.parse(row.timestamp)) + ? Date.parse(row.timestamp) + : Date.now(); + + publishDashboardEvent( + "compression.completed", + { + requestId: row.request_id || `analytics-${row.id}`, + comboId: row.compression_combo_id || row.combo_id || null, + mode: row.mode, + originalTokens, + compressedTokens, + savingsPercent, + engineBreakdown: [ + { + engine: row.engine || row.mode || "compression", + originalTokens, + compressedTokens, + savingsPercent, + techniquesUsed: [], + rulesApplied: [], + durationMs: row.duration_ms ?? undefined, + }, + ], + validationWarnings: [], + fallbackApplied: Boolean(row.validation_fallback), + timestamp, + }, + timestamp + ); + console.log( + "[LiveWS] Seeded latest compression run from analytics: %s", + row.request_id || row.id + ); + } catch (err) { + console.warn( + "[LiveWS] Could not seed compression analytics backlog: %s", + err instanceof Error ? err.message : String(err) + ); + } } // ── Heartbeat ───────────────────────────────────────────────────────────── @@ -270,13 +439,39 @@ export async function startLiveDashboardServer( port = DEFAULT_PORT, host = DEFAULT_HOST ): Promise { - const server = createServer(); + const server = createServer((req, res) => { + handleInternalEventRequest(req, res); + }); const wss = new WebSocketServer({ server }); // Subscribe to EventBus const unsubscribe = subscribeToEventBus(); + await seedLatestCompressionRunFromDb(); wss.on("connection", async (ws, request) => { + const pendingMessages: string[] = []; + let activeClientId: string | null = null; + + // Clients can send the subscribe frame immediately after the WS open event, + // while dashboard cookie/API-key auth is still resolving. Queue those early + // messages so the first subscribe is not dropped. + ws.on("message", (data) => { + const raw = data.toString(); + if (!activeClientId) { + if ( + pendingMessages.length >= MAX_PENDING_MESSAGES_PER_CLIENT || + raw.length > MAX_PENDING_MESSAGE_BYTES + ) { + sendTo(ws, { type: "error", code: "RATE_LIMITED", message: "Too many early messages" }); + ws.close(4008, "Too many early messages"); + return; + } + pendingMessages.push(raw); + return; + } + handleMessage(activeClientId, raw); + }); + // Origin check — browsers always send Origin on the WS upgrade; reject // unknown origins to stop drive-by cross-origin WebSocket from a victim // page. Non-browser clients (CLI / MCP) omit Origin and are accepted @@ -305,6 +500,7 @@ export async function startLiveDashboardServer( } const clientId = auth.sessionId; + activeClientId = clientId; const client: ClientState = { ws, sessionId: clientId, @@ -327,10 +523,10 @@ export async function startLiveDashboardServer( clients.size ); - // Handle messages - ws.on("message", (data) => { - handleMessage(clientId, data.toString()); - }); + // Replay any subscribe/ping frames sent while auth was still pending. + for (const raw of pendingMessages.splice(0)) { + handleMessage(clientId, raw); + } // Handle close ws.on("close", () => { diff --git a/src/server/ws/types.ts b/src/server/ws/types.ts index 759493b46d..2fa678e768 100644 --- a/src/server/ws/types.ts +++ b/src/server/ws/types.ts @@ -8,7 +8,7 @@ export interface WsSubscribeMessage { type: "subscribe"; - channels: Array<"requests" | "combo" | "credentials">; + channels: Array<"requests" | "combo" | "credentials" | "compression">; } export interface WsPingMessage { @@ -21,7 +21,7 @@ export type WsClientMessage = WsSubscribeMessage | WsPingMessage; export interface WsEventMessage { type: "event"; - channel: "requests" | "combo" | "credentials"; + channel: "requests" | "combo" | "credentials" | "compression"; event: string; data: unknown; } @@ -35,7 +35,7 @@ export interface WsWelcomeMessage { version: string; sessionId: string; serverTime: number; - channels: Array<"requests" | "combo" | "credentials">; + channels: Array<"requests" | "combo" | "credentials" | "compression">; /** Number of buffered events since last reconnect */ backlog: number; } diff --git a/tests/unit/cli/live-ws-startup.test.ts b/tests/unit/cli/live-ws-startup.test.ts new file mode 100644 index 0000000000..4a4e799dcb --- /dev/null +++ b/tests/unit/cli/live-ws-startup.test.ts @@ -0,0 +1,151 @@ +import assert from "node:assert/strict"; +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { SignJWT } from "jose"; +import net from "node:net"; +import test from "node:test"; +import WebSocket from "ws"; + +function getFreePort(): Promise { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + server.close(() => { + if (address && typeof address === "object") resolve(address.port); + else reject(new Error("Failed to allocate a local port")); + }); + }); + }); +} + +function terminateTree(child: ChildProcessWithoutNullStreams): void { + if (!child.pid) return; + + try { + process.kill(-child.pid, "SIGTERM"); + } catch { + child.kill("SIGTERM"); + } +} + +function waitForStartup( + child: ChildProcessWithoutNullStreams, + getOutput: () => string +): Promise { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error(`LiveWS startup timed out. Output:\n${getOutput()}`)); + }, 8_000); + + const onData = () => { + const output = getOutput(); + if (output.includes("Dashboard WebSocket server listening")) { + cleanup(); + resolve(); + } + }; + + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + cleanup(); + reject( + new Error(`LiveWS exited before listening: code=${code} signal=${signal}\n${getOutput()}`) + ); + }; + + const cleanup = () => { + clearTimeout(timeout); + child.stdout.off("data", onData); + child.stderr.off("data", onData); + child.off("exit", onExit); + }; + + child.stdout.on("data", onData); + child.stderr.on("data", onData); + child.once("exit", onExit); + onData(); + }); +} + +test( + "LiveWS startup script boots on current Node and accepts API-key WebSocket clients", + { timeout: 15_000 }, + async () => { + const port = await getFreePort(); + const apiKey = "test-live-ws-key"; + const jwtSecret = "test-live-ws-jwt-secret"; + const origin = "http://localhost"; + let output = ""; + + const child = spawn(process.execPath, ["scripts/start-ws-server.mjs"], { + cwd: process.cwd(), + detached: process.platform !== "win32", + env: { + ...process.env, + NODE_ENV: "test", + OMNIROUTE_API_KEY: apiKey, + JWT_SECRET: jwtSecret, + LIVE_WS_HOST: "127.0.0.1", + LIVE_WS_PORT: String(port), + LIVE_WS_ALLOWED_ORIGINS: origin, + }, + }); + + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + output += chunk; + }); + child.stderr.on("data", (chunk) => { + output += chunk; + }); + + try { + await waitForStartup(child, () => output); + + assert.doesNotMatch(output, /tsx must be loaded with --import/i); + assert.doesNotMatch(output, /EADDRINUSE/i); + + async function expectLiveWsOpen(headers: Record): Promise { + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error(`Timed out waiting for LiveWS connection. Output:\n${output}`)); + }, 4_000); + + const ws = new WebSocket(`ws://127.0.0.1:${port}/live-ws`, { headers }); + + ws.once("open", () => { + clearTimeout(timeout); + ws.close(1000); + resolve(); + }); + + ws.once("error", (error) => { + clearTimeout(timeout); + reject(new Error(`LiveWS client failed: ${error.message}. Output:\n${output}`)); + }); + }); + } + + await expectLiveWsOpen({ + Authorization: `Bearer ${apiKey}`, + Origin: origin, + }); + + const dashboardToken = await new SignJWT({ authenticated: true }) + .setProtectedHeader({ alg: "HS256" }) + .setExpirationTime("5m") + .sign(new TextEncoder().encode(jwtSecret)); + + // Send auth_token preceded by another cookie (the real browser case: "a=1; auth_token=…"). + // Guards #4004's cookie-parse regex: a literal-"s" bug (\s vs \\s) only matched auth_token + // when it was the FIRST cookie, silently breaking same-origin reverse-proxy auth otherwise. + await expectLiveWsOpen({ + Cookie: `omni_pref=dark; auth_token=${dashboardToken}`, + Origin: origin, + }); + } finally { + terminateTree(child); + } + } +); diff --git a/tests/unit/live-ws-cookie-parse.test.ts b/tests/unit/live-ws-cookie-parse.test.ts new file mode 100644 index 0000000000..06877f9da9 --- /dev/null +++ b/tests/unit/live-ws-cookie-parse.test.ts @@ -0,0 +1,42 @@ +// Regression for #4004: liveServer's cookie parser used an UNTAGGED template literal +// `(?:^|;\s*)` — \s collapsed to a literal "s", so auth_token only matched when it was +// the FIRST cookie. Browsers serialize "a=1; auth_token=…", so the same-origin +// reverse-proxy dashboard auth silently failed whenever any cookie preceded auth_token. +// Keep the server from auto-starting on import. +process.env.OMNIROUTE_ENABLE_LIVE_WS = "0"; + +import test from "node:test"; +import assert from "node:assert/strict"; +import { getCookieValueFromHeader } from "@/server/ws/liveServer"; + +test("getCookieValueFromHeader reads auth_token when it is the only cookie", () => { + assert.equal(getCookieValueFromHeader({ cookie: "auth_token=abc123" }, "auth_token"), "abc123"); +}); + +test("getCookieValueFromHeader reads auth_token when preceded by other cookies (#4004)", () => { + // The standard browser "; " separator — the case the \s-vs-\\s bug broke. + assert.equal( + getCookieValueFromHeader({ cookie: "omni_pref=dark; auth_token=abc123" }, "auth_token"), + "abc123" + ); + assert.equal( + getCookieValueFromHeader({ cookie: "a=1; b=2; auth_token=xyz" }, "auth_token"), + "xyz" + ); +}); + +test("getCookieValueFromHeader handles a no-space separator too", () => { + assert.equal(getCookieValueFromHeader({ cookie: "a=1;auth_token=tok" }, "auth_token"), "tok"); +}); + +test("getCookieValueFromHeader returns null when the cookie is absent", () => { + assert.equal(getCookieValueFromHeader({ cookie: "other=1; foo=2" }, "auth_token"), null); + assert.equal(getCookieValueFromHeader({}, "auth_token"), null); +}); + +test("getCookieValueFromHeader URL-decodes the value", () => { + assert.equal( + getCookieValueFromHeader({ cookie: "x=1; auth_token=a%20b" }, "auth_token"), + "a b" + ); +});