fix(live-ws): bridge sidecar events to dashboard (#4004)

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.
This commit is contained in:
Demiurge The Single 2026-06-16 19:15:25 +03:00 committed by GitHub
parent 0bc670e414
commit 02302131fb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 525 additions and 54 deletions

View file

@ -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."
}

View file

@ -445,6 +445,24 @@ function extractMemoryTextFromRequestBody(
return "";
}
async function forwardDashboardEventToLiveWs(event: string, payload: unknown): Promise<void> {
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.

View file

@ -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",

View file

@ -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=<api-key>`);
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=<api-key>", port);
console.log("[LiveWS] Channels: requests, combo, credentials");
})
.catch((err) => {
console.error("[LiveWS] Failed to start:", err);

View file

@ -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 ─────────────────────────────────────────────────────────────────

View file

@ -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;
}

View file

@ -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<WsAuthResult> {
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<boolean> {
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<string, unknown>):
// ── 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<void> {
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<import("http").Server> {
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", () => {

View file

@ -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;
}

View file

@ -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<number> {
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<void> {
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<string, string>): Promise<void> {
await new Promise<void>((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);
}
}
);

View file

@ -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"
);
});