fix(ctl): harden proxy and WebSocket handler

Author: vhscom

packages/control/src/bridge/handler.ts

@@ -21,6 +21,7 @@ import {
 	HEARTBEAT_TIMEOUT_MS,
 	IDLE_TIMEOUT_MS,
 	MAX_MESSAGE_BYTES,
+	MAX_PENDING_MESSAGES,
 	type ProxyConnection,
 	RATE_LIMIT_MAX,
 	RATE_LIMIT_WINDOW_MS,
@@ -366,7 +367,12 @@ export function createProxyHandler(
 		onMessage(evt, ws) {
 			conn.lastActivity = Date.now();
 
-			const raw = typeof evt.data === "string" ? evt.data : String(evt.data);
+			if (typeof evt.data !== "string") {
+				closeWith(ws, 1003, "Binary frames not supported");
+				cleanup();
+				return;
+			}
+			const raw = evt.data;
 
 			if (raw.length > MAX_MESSAGE_BYTES) {
 				closeWith(ws, 1009, "Message too large");
@@ -381,6 +387,11 @@ export function createProxyHandler(
 			}
 
 			if (conn.state === "connecting") {
+				if (conn.pendingMessages.length >= MAX_PENDING_MESSAGES) {
+					closeWith(ws, CloseCodes.RATE_LIMITED, "Too many pending messages");
+					cleanup();
+					return;
+				}
 				conn.pendingMessages.push(raw);
 				return;
 			}

packages/control/src/bridge/types.ts

@@ -56,6 +56,8 @@ export const RATE_LIMIT_WINDOW_MS = 1_000;
 export const RATE_LIMIT_MAX = 10;
 /** Maximum inbound message size in bytes. */
 export const MAX_MESSAGE_BYTES = 1024 * 1024;
+/** Maximum messages buffered while backend is connecting. */
+export const MAX_PENDING_MESSAGES = 20;
 
 /** Interval (ms) between heartbeat pings with session re-validation. */
 export const HEARTBEAT_INTERVAL_MS = 25_000;

packages/control/src/proxy.ts

@@ -21,7 +21,16 @@ export async function proxyToGateway(
 		target.port = gateway.port;
 		target.pathname = target.pathname.replace("/ops/control", "");
 
-		const res = await fetch(target.toString(), ctx.req.raw);
+		const headers = new Headers(ctx.req.raw.headers);
+		headers.delete("cookie");
+		headers.delete("authorization");
+
+		const res = await fetch(target.toString(), {
+			method: ctx.req.method,
+			headers,
+			body: ctx.req.raw.body,
+			redirect: "manual",
+		});
 
 		if (!res.ok) {
 			return ctx.json({ error: "Bad Gateway" }, 502);

packages/control/test/handler.test.ts

@@ -108,6 +108,21 @@ describe("createProxyHandler", () => {
 		expect(mock.closeCode).toBe(CloseCodes.BACKEND_UNAVAILABLE);
 	});
 
+	it("rejects binary frames", () => {
+		const handler = createProxyHandler(makePrincipal(), makeDeps());
+		const mock = createMockWs();
+
+		const msgEvent = new MessageEvent("message", {
+			data: new ArrayBuffer(8),
+		});
+
+		// biome-ignore lint/suspicious/noExplicitAny: test mock
+		handler.onMessage?.(msgEvent, mock.ws as any);
+
+		expect(mock.closed).toBe(true);
+		expect(mock.closeCode).toBe(1003);
+	});
+
 	it("handles oversized messages", () => {
 		const handler = createProxyHandler(makePrincipal(), makeDeps());
 		const mock = createMockWs();
@@ -181,6 +196,25 @@ describe("concurrent connection limit", () => {
 
 // --- Rate limiting ---
 
+describe("pending message buffer", () => {
+	it("closes connection when pending buffer overflows", () => {
+		const handler = createProxyHandler(makePrincipal(), makeDeps());
+		const mock = createMockWs();
+
+		// Don't call onOpen — connection stays in "connecting" state
+		for (let i = 0; i <= 20; i++) {
+			const msgEvent = new MessageEvent("message", {
+				data: JSON.stringify({ type: "req", method: "test", i }),
+			});
+			// biome-ignore lint/suspicious/noExplicitAny: test mock
+			handler.onMessage?.(msgEvent, mock.ws as any);
+		}
+
+		expect(mock.closed).toBe(true);
+		expect(mock.closeCode).toBe(CloseCodes.RATE_LIMITED);
+	});
+});
+
 describe("message rate limiting", () => {
 	it("closes connection after exceeding rate limit", () => {
 		const handler = createProxyHandler(makePrincipal(), makeDeps());

packages/control/test/proxy.test.ts

@@ -68,6 +68,25 @@ describe("proxyToGateway", () => {
 		expect(body).toEqual({ error: "Bad Gateway" });
 	});
 
+	it("strips Cookie and Authorization headers before forwarding", async () => {
+		fetchSpy.mockResolvedValueOnce(new Response("ok", { status: 200 }));
+
+		const app = createApp("http://gateway:18789");
+		await app.request("/ops/control/dashboard", {
+			headers: {
+				Cookie: "access_token=secret; refresh_token=secret",
+				Authorization: "Bearer leaked",
+				Accept: "text/html",
+			},
+		});
+
+		const calledInit = fetchSpy.mock.calls[0]?.[1] as RequestInit;
+		const fwdHeaders = new Headers(calledInit.headers);
+		expect(fwdHeaders.get("cookie")).toBeNull();
+		expect(fwdHeaders.get("authorization")).toBeNull();
+		expect(fwdHeaders.get("accept")).toBe("text/html");
+	});
+
 	it("does not leak gateway error details", async () => {
 		fetchSpy.mockResolvedValueOnce(
 			new Response(