From f2e93bba02734461b4e40a1a2684372e6f2437c7 Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Wed, 14 Oct 2026 12:08:17 +0100 Subject: broker: a standalone bridge for the browser's host Upstream's bridge lives inside whichever pi session started first on the machine running the browser, and only serves that machine. The broker takes its place permanently: upstream's HTTP contract to the extension on loopback, and /command and /status on a second, private address for other hosts, behind a bearer token (PI_CHROME_TOKEN, or PI_CHROME_TOKEN_COMMAND). --- broker/broker.mjs | 388 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 388 insertions(+) create mode 100644 broker/broker.mjs diff --git a/broker/broker.mjs b/broker/broker.mjs new file mode 100644 index 0000000..4af8e9d --- /dev/null +++ b/broker/broker.mjs @@ -0,0 +1,388 @@ +#!/usr/bin/env node +// Standalone broker for a pi-chrome companion extension on a shared host (see FORK.md). +// +// Upstream's bridge lives inside whichever pi session started first on the machine running the +// browser: it owns 127.0.0.1:17318, the companion extension long-polls it, and later sessions on +// that same machine forward through it. That cannot serve pi sessions on other hosts - the +// extension only ever polls its own loopback, and the owner disappears with its session. +// +// This process takes the owner's place, permanently, on the browser's host: +// +// extension listener EXT_HOST:PORT (loopback) /next, /result, /status, /command +// peer listener PEER_HOST:PORT (private) /command, /status - nothing else +// +// Keeping /next and /result off the peer listener means nothing on the network can drain the +// extension's command queue or forge its results. /command needs the shared bearer token on +// both listeners, because loopback here is shared with everything else running as the user and +// the browser behind it is signed into real accounts. +// +// The HTTP contract towards the extension is upstream's, unchanged: the extension is loaded as +// shipped and has no idea it is talking to anything but a pi session. +// +// Node builtins only, so it runs straight from a checkout with no install step. + +import { createServer } from "node:http"; +import { readFileSync } from "node:fs"; +import { execSync } from "node:child_process"; +import { createHash, timingSafeEqual } from "node:crypto"; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const PKG_PATH = resolve(dirname(fileURLToPath(import.meta.url)), "..", "package.json"); + +const PORT = Number(process.env.PI_CHROME_BROKER_PORT ?? "17318"); +const EXT_HOST = process.env.PI_CHROME_BROKER_EXT_HOST ?? "127.0.0.1"; +const PEER_HOST = process.env.PI_CHROME_BROKER_PEER_HOST?.trim() || undefined; + +const DEFAULT_TIMEOUT_MS = 30_000; +const MAX_TIMEOUT_MS = 10 * 60_000; +const LONG_POLL_MS = 25_000; +// Results carry screenshots and full-page tiles as base64, so this is sized for those. +const MAX_BODY_BYTES = 64 * 1024 * 1024; + +function log(level, msg, extra = {}) { + const line = JSON.stringify({ t: new Date().toISOString(), level, msg, ...extra }); + if (level === "error") console.error(line); + else console.log(line); +} + +function readVersion() { + try { + return JSON.parse(readFileSync(PKG_PATH, "utf8")).version ?? "0.0.0-dev"; + } catch { + return "0.0.0-dev"; + } +} + +// PI_CHROME_TOKEN, or the output of PI_CHROME_TOKEN_COMMAND for a secret store - the same token +// every client sends. There is no default: the browser behind this is signed into real accounts. +function loadToken() { + const fromEnv = process.env.PI_CHROME_TOKEN?.trim(); + if (fromEnv) return fromEnv; + const command = process.env.PI_CHROME_TOKEN_COMMAND?.trim(); + if (!command) throw new Error("the broker needs a token: set PI_CHROME_TOKEN, or PI_CHROME_TOKEN_COMMAND to read one from a secret store"); + const token = execSync(command, { encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] }).trim(); + if (!token) throw new Error(`PI_CHROME_TOKEN_COMMAND produced nothing: ${command}`); + return token; +} + +const TOKEN_DIGEST = createHash("sha256").update(loadToken()).digest(); + +// Compare digests rather than the raw strings: timingSafeEqual needs equal lengths, and the +// length of a guess must not leak either. +function hasValidToken(request) { + const raw = String(request.headers.authorization ?? ""); + const match = /^Bearer\s+(.+)$/i.exec(raw); + if (!match) return false; + return timingSafeEqual(createHash("sha256").update(match[1].trim()).digest(), TOKEN_DIGEST); +} + +// Same origin rules as upstream's bridge (index.ts), so the extension sees identical behaviour. +function corsHeadersFor(request) { + const origin = String(request.headers.origin ?? ""); + if (!origin.startsWith("chrome-extension://")) return {}; + return { + "access-control-allow-origin": origin, + "access-control-allow-methods": "GET,POST,OPTIONS", + "access-control-allow-headers": "content-type", + "access-control-expose-headers": "x-pi-chrome-version", + "vary": "origin", + }; +} + +function isBrowserOriginAllowed(request) { + const origin = String(request.headers.origin ?? ""); + if (origin) return origin.startsWith("chrome-extension://"); + const secFetchSite = String(request.headers["sec-fetch-site"] ?? ""); + return !secFetchSite || secFetchSite === "none" || secFetchSite === "same-origin"; +} + +function isLocalProcessRequest(request) { + return !request.headers.origin && !request.headers["sec-fetch-site"]; +} + +function sendJson(response, status, body, extraHeaders = {}) { + if (response.writableEnded) return; + response.writeHead(status, { + "content-type": "application/json; charset=utf-8", + "cache-control": "no-store", + ...extraHeaders, + }); + response.end(JSON.stringify(body)); +} + +function readBody(request) { + return new Promise((resolveBody, rejectBody) => { + const chunks = []; + let size = 0; + request.on("data", (chunk) => { + size += chunk.length; + if (size > MAX_BODY_BYTES) { + rejectBody(new Error(`request body over ${MAX_BODY_BYTES} bytes`)); + request.destroy(); + return; + } + chunks.push(Buffer.from(chunk)); + }); + request.on("end", () => resolveBody(Buffer.concat(chunks).toString("utf8"))); + request.on("error", rejectBody); + }); +} + +// --- command queue: a port of upstream's ChromeProfileBridge server half ---------------------- + +const queue = []; +const waiters = []; +const pending = new Map(); +let lastSeenAt; +let clientName; +const startedAt = Date.now(); + +function enqueue(command) { + const waiter = waiters.shift(); + if (waiter) waiter(command); + else queue.push(command); +} + +function dropQueued(id) { + const index = queue.findIndex((queued) => queued.id === id); + if (index !== -1) queue.splice(index, 1); +} + +function waitForCommand(timeoutMs, registerWaiter) { + return new Promise((resolveWait) => { + let settled = false; + const waiter = (command) => { + if (settled) return; + settled = true; + clearTimeout(timer); + const index = waiters.indexOf(waiter); + if (index !== -1) waiters.splice(index, 1); + resolveWait(command); + }; + const timer = setTimeout(() => waiter(undefined), timeoutMs); + waiters.push(waiter); + registerWaiter(waiter); + }); +} + +// Upstream's three-way diagnosis, so a client sees the same actionable timeout text. +function timeoutMessage(entry, timeoutMs) { + const pollAgeMs = lastSeenAt === undefined ? undefined : Date.now() - lastSeenAt; + if (entry?.deliveredAt) { + return `Timed out after ${timeoutMs}ms: the browser extension received the command but never returned a result. The action may be long-running, or the result post failed. Retry; if it persists, reload 'Pi Chrome Connector' on the browser host.`; + } + if (pollAgeMs === undefined || pollAgeMs > 60_000) { + return `Timed out after ${timeoutMs}ms: the browser extension is not polling the broker (last seen ${pollAgeMs === undefined ? "never" : Math.round(pollAgeMs / 1000) + "s ago"}). The browser on the broker's host is closed, or 'Pi Chrome Connector' is not loaded in it.`; + } + return `Timed out after ${timeoutMs}ms: the browser extension is polling (last seen ${Math.round(pollAgeMs / 1000)}s ago) but did not pick up this command in time. Retry; if it persists, reload 'Pi Chrome Connector' on the browser host.`; +} + +function runCommand(action, params, timeoutMs, onAbort) { + const id = `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; + const command = { id, action, params }; + return new Promise((resolveCommand, rejectCommand) => { + const timer = setTimeout(() => { + const entry = pending.get(id); + pending.delete(id); + dropQueued(id); + rejectCommand(new Error(timeoutMessage(entry, timeoutMs))); + }, timeoutMs); + pending.set(id, { command, resolve: resolveCommand, reject: rejectCommand, timer }); + // A client that hangs up before delivery should not leave its command to run later + // against a browser nobody is waiting on. Once delivered, it is the extension's. + onAbort(() => { + const entry = pending.get(id); + if (!entry || entry.deliveredAt) return; + clearTimeout(entry.timer); + pending.delete(id); + dropQueued(id); + rejectCommand(new Error("client went away")); + }); + enqueue(command); + }); +} + +function status() { + return { + mode: "broker", + version: readVersion(), + startedAt, + connected: lastSeenAt !== undefined && Date.now() - lastSeenAt < 5 * 60_000, + lastSeenAt, + clientName, + queuedCommands: queue.length, + pendingCommands: pending.size, + }; +} + +async function handleCommand(request, response) { + if (!isLocalProcessRequest(request)) { + sendJson(response, 403, { ok: false, error: "Chrome commands are accepted only from local Pi processes" }); + return; + } + if (!hasValidToken(request)) { + sendJson(response, 401, { ok: false, error: "pi-chrome broker rejected the bearer token (check PI_CHROME_TOKEN on both ends)" }); + return; + } + let body; + try { + body = JSON.parse(await readBody(request)); + } catch (error) { + sendJson(response, 400, { ok: false, error: `Bad command body: ${error.message}` }); + return; + } + if (!body?.action) { + sendJson(response, 400, { ok: false, error: "Missing command action" }); + return; + } + const timeoutMs = Math.min(Math.max(Number(body.timeoutMs) || DEFAULT_TIMEOUT_MS, 1_000), MAX_TIMEOUT_MS); + const started = Date.now(); + const peer = request.socket.remoteAddress; + try { + const result = await runCommand(body.action, body.params ?? {}, timeoutMs, (abort) => { + response.once("close", () => { + if (!response.writableEnded) abort(); + }); + }); + // Log the action, never its params: those carry typed text, form values and page data. + log("info", "command", { action: body.action, peer, ok: true, ms: Date.now() - started }); + sendJson(response, 200, { ok: true, result }); + } catch (error) { + log("warn", "command", { action: body.action, peer, ok: false, ms: Date.now() - started, error: error.message }); + sendJson(response, 504, { ok: false, error: error.message }); + } +} + +async function handleNext(request, response, url) { + if (!isBrowserOriginAllowed(request)) { + sendJson(response, 403, { ok: false, error: "browser origin not allowed" }); + return; + } + if (lastSeenAt === undefined || Date.now() - lastSeenAt > 60_000) { + log("info", "extension polling", { name: url.searchParams.get("name") ?? undefined }); + } + lastSeenAt = Date.now(); + clientName = url.searchParams.get("name") ?? undefined; + let aborted = false; + let activeWaiter; + response.once("close", () => { + if (response.writableEnded) return; + aborted = true; + const index = activeWaiter ? waiters.indexOf(activeWaiter) : -1; + if (index !== -1) waiters.splice(index, 1); + }); + let command = queue.shift(); + if (!command) { + command = await waitForCommand(LONG_POLL_MS, (waiter) => { + activeWaiter = waiter; + }); + } + if (aborted) { + // The long-poll died before delivery: put the command back for the next poll. + if (command) queue.unshift(command); + return; + } + if (command) { + const entry = pending.get(command.id); + if (entry) entry.deliveredAt = Date.now(); + } + const version = readVersion(); + sendJson( + response, + 200, + command ? { type: "command", command, expectedExtensionVersion: version } : { type: "none", expectedExtensionVersion: version }, + { ...corsHeadersFor(request), "x-pi-chrome-version": version }, + ); +} + +async function handleResult(request, response) { + if (!isBrowserOriginAllowed(request)) { + sendJson(response, 403, { ok: false, error: "browser origin not allowed" }); + return; + } + lastSeenAt = Date.now(); + const corsHeaders = corsHeadersFor(request); + let result; + try { + result = JSON.parse(await readBody(request)); + } catch (error) { + sendJson(response, 400, { ok: false, error: `Bad result body: ${error.message}` }, corsHeaders); + return; + } + const entry = pending.get(result?.id); + if (!entry) { + sendJson(response, 404, { ok: false, error: "unknown command id" }, corsHeaders); + return; + } + clearTimeout(entry.timer); + pending.delete(result.id); + if (result.ok) entry.resolve(result.result); + else entry.reject(new Error(result.error ?? "Chrome extension command failed")); + sendJson(response, 200, { ok: true }, corsHeaders); +} + +async function handle(listener, request, response) { + const url = new URL(request.url ?? "/", "http://broker"); + const route = `${request.method} ${url.pathname}`; + + if (route === "POST /command") return handleCommand(request, response); + + if (route === "GET /status") { + if (listener === "peer" && !hasValidToken(request)) { + sendJson(response, 401, { ok: false, error: "bearer token required" }); + return; + } + sendJson(response, 200, status()); + return; + } + + if (listener === "ext") { + if (request.method === "OPTIONS") { + if (!isBrowserOriginAllowed(request)) { + sendJson(response, 403, { ok: false, error: "browser origin not allowed" }); + return; + } + sendJson(response, 200, { ok: true }, corsHeadersFor(request)); + return; + } + if (route === "GET /next") return handleNext(request, response, url); + if (route === "POST /result") return handleResult(request, response); + } + + sendJson(response, 404, { error: "not found" }); +} + +function listen(listener, host) { + const server = createServer((request, response) => { + handle(listener, request, response).catch((error) => { + log("error", "request failed", { listener, error: error.message }); + sendJson(response, 500, { ok: false, error: error.message }); + }); + }); + server.on("error", (error) => { + // Most likely the private address is not up yet. Exit and let the supervisor retry, rather than + // run half a broker that the other hosts cannot reach. + log("error", "listen failed", { listener, host, port: PORT, error: error.message }); + process.exit(1); + }); + server.listen(PORT, host, () => log("info", "listening", { listener, host, port: PORT, version: readVersion() })); + return server; +} + +const servers = [listen("ext", EXT_HOST)]; +if (PEER_HOST && PEER_HOST !== EXT_HOST) servers.push(listen("peer", PEER_HOST)); + +function shutdown(signal) { + log("info", "shutting down", { signal, pending: pending.size }); + for (const entry of pending.values()) { + clearTimeout(entry.timer); + entry.reject(new Error("pi-chrome broker stopped")); + } + pending.clear(); + for (const waiter of [...waiters]) waiter(undefined); + for (const server of servers) server.close(); + setTimeout(() => process.exit(0), 1_000).unref(); +} +process.on("SIGTERM", () => shutdown("SIGTERM")); +process.on("SIGINT", () => shutdown("SIGINT")); -- cgit v1.2.3