From b944cb2f21b31ac3c562c546c90da4ed2b49820a Mon Sep 17 00:00:00 2001 From: Krishna Date: Mon, 5 Oct 2026 00:46:08 +0530 Subject: [PATCH 1/4] fix(cli): authenticate remote hosts with API keys --- .../cli/src/lib/host-target/exchangeApiKey.ts | 39 ++++++ .../lib/host-target/resolveHostTarget.test.ts | 117 ++++++++++++++++++ .../src/lib/host-target/resolveHostTarget.ts | 6 +- 3 files changed, 160 insertions(+), 2 deletions(-) create mode 100644 packages/cli/src/lib/host-target/exchangeApiKey.ts create mode 100644 packages/cli/src/lib/host-target/resolveHostTarget.test.ts diff --git a/packages/cli/src/lib/host-target/exchangeApiKey.ts b/packages/cli/src/lib/host-target/exchangeApiKey.ts new file mode 100644 index 00000000000..31735a68468 --- /dev/null +++ b/packages/cli/src/lib/host-target/exchangeApiKey.ts @@ -0,0 +1,39 @@ +import { CLIError } from "@superset/cli-framework"; +import { z } from "zod"; +import { env } from "../env"; + +const tokenResponse = z.object({ token: z.string().trim().min(1) }); + +export async function exchangeApiKey(bearer: string): Promise { + if (!bearer.startsWith("sk_live_")) return bearer; + + let response: Response; + try { + response = await fetch(`${env.SUPERSET_API_URL}/api/auth/token`, { + headers: { "x-api-key": bearer }, + signal: AbortSignal.timeout(10_000), + redirect: "error", + }); + } catch { + throw new CLIError( + "Could not exchange API key for remote host access", + "Check connectivity to the Superset API and try again", + ); + } + if (!response.ok) { + throw new CLIError( + `Could not exchange API key for remote host access (HTTP ${response.status})`, + response.status === 401 || response.status === 403 + ? "Check --api-key or SUPERSET_API_KEY" + : "Check the Superset API and try again", + ); + } + + const parsed = tokenResponse.safeParse( + await response.json().catch(() => null), + ); + if (!parsed.success) { + throw new CLIError("Superset API returned an invalid remote host token"); + } + return parsed.data.token; +} diff --git a/packages/cli/src/lib/host-target/resolveHostTarget.test.ts b/packages/cli/src/lib/host-target/resolveHostTarget.test.ts new file mode 100644 index 00000000000..7fdfd7bab67 --- /dev/null +++ b/packages/cli/src/lib/host-target/resolveHostTarget.test.ts @@ -0,0 +1,117 @@ +import { afterAll, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { createApiClient } from "../api-client"; +import { env } from "../env"; +import { resolveHostTarget } from "./resolveHostTarget"; + +const apiKey = ["sk", "live", "remote", "host", "fixture"].join("_"); +const jwt = "header.payload.signature"; +const requests: { url: string; headers: Headers }[] = []; +let tokenResponse = () => Response.json({ token: jwt }); +const server = Bun.serve({ + port: 0, + fetch(request, server): Response { + requests.push({ url: request.url, headers: new Headers(request.headers) }); + const url = new URL(request.url); + if (url.pathname === "/api/auth/token") return tokenResponse(); + if (url.pathname === "/api/trpc/host.relayEndpoint") { + return Response.json([ + { result: { data: { json: { url: server.url.origin } } } }, + ]); + } + return Response.json([{ result: { data: { json: [] } } }]); + }, +}); +const originalApiUrl = env.SUPERSET_API_URL; + +beforeEach(() => { + env.SUPERSET_API_URL = server.url.origin; + requests.length = 0; + tokenResponse = () => Response.json({ token: jwt }); +}); + +afterAll(() => { + env.SUPERSET_API_URL = originalApiUrl; + server.stop(true); +}); + +function resolve(bearer = apiKey) { + return resolveHostTarget({ + requestedHostId: "remote-host-fixture", + organizationId: "org-fixture", + userJwt: bearer, + api: createApiClient({ bearer, organizationId: "org-fixture" }), + }); +} + +describe("resolveHostTarget remote authentication", () => { + test("exchanges an API key for HTTP and WebSocket credentials", async () => { + const target = await resolve(); + expect(await target.client.workspace.list.query()).toEqual([]); + const exchange = requests.find((request) => + request.url.endsWith("/api/auth/token"), + ); + expect(exchange?.headers.get("x-api-key")).toBe(apiKey); + expect(exchange?.headers.get("authorization")).toBeNull(); + const hostRequest = requests.find((request) => + new URL(request.url).pathname.startsWith("/hosts/"), + ); + expect(hostRequest?.headers.get("authorization")).toBe(`Bearer ${jwt}`); + expect(hostRequest?.headers.get("x-api-key")).toBeNull(); + expect(target.ws.token).toBe(jwt); + }); + + test("passes OAuth JWTs through without an exchange", async () => { + const target = await resolve(jwt); + await target.client.workspace.list.query(); + expect( + requests.some((request) => request.url.endsWith("/api/auth/token")), + ).toBe(false); + expect(requests.at(-1)?.headers.get("authorization")).toBe(`Bearer ${jwt}`); + expect(target.ws.token).toBe(jwt); + }); + + test("reports rejected keys before contacting the relay", async () => { + tokenResponse = () => + new Response("private error details", { status: 401 }); + await expect(resolve()).rejects.toThrow( + "Could not exchange API key for remote host access (HTTP 401)", + ); + expect(requests).toHaveLength(1); + }); + + test.each([ + {}, + { token: 123 }, + { token: " " }, + null, + ])("rejects an invalid token response: %j", async (body) => { + tokenResponse = () => Response.json(body); + await expect(resolve()).rejects.toThrow( + "Superset API returned an invalid remote host token", + ); + expect(requests).toHaveLength(1); + }); + + test("does not forward the API key through a redirect", async () => { + tokenResponse = () => Response.redirect(`${server.url.origin}/other`, 302); + await expect(resolve()).rejects.toThrow( + "Could not exchange API key for remote host access", + ); + expect(requests).toHaveLength(1); + }); + + test("bounds token exchange with an abort signal", async () => { + const timeout = spyOn(AbortSignal, "timeout").mockReturnValue( + AbortSignal.abort(new DOMException("Timed out", "TimeoutError")), + ); + try { + await expect(resolve()).rejects.toThrow( + "Could not exchange API key for remote host access", + ); + expect(timeout).toHaveBeenCalledWith(10_000); + expect(requests).toHaveLength(0); + } finally { + timeout.mockRestore(); + } + }); +}); diff --git a/packages/cli/src/lib/host-target/resolveHostTarget.ts b/packages/cli/src/lib/host-target/resolveHostTarget.ts index 6d1d397ec4c..a1cb3038fb5 100644 --- a/packages/cli/src/lib/host-target/resolveHostTarget.ts +++ b/packages/cli/src/lib/host-target/resolveHostTarget.ts @@ -10,6 +10,7 @@ import SuperJSON from "superjson"; import type { ApiClient } from "../api-client"; import { isProcessAlive, readManifest } from "../host/manifest"; import { getRelayUrl } from "../host/relay-url"; +import { exchangeApiKey } from "./exchangeApiKey"; import { readJwtSubject } from "./readJwtSubject"; export type HostServiceClient = ReturnType< @@ -92,6 +93,7 @@ export async function resolveHostTarget( } const routingKey = buildHostRoutingKey(options.organizationId, targetHostId); + const userJwt = await exchangeApiKey(options.userJwt); const relayUrl = await getRelayUrl(options.api); return { kind: "remote", @@ -102,7 +104,7 @@ export async function resolveHostTarget( url: `${relayUrl}/hosts/${routingKey}/trpc`, transformer: SuperJSON, headers: { - Authorization: `Bearer ${options.userJwt}`, + Authorization: `Bearer ${userJwt}`, "x-superset-client-machine-id": localHostId, }, }), @@ -110,7 +112,7 @@ export async function resolveHostTarget( }), ws: { baseWsUrl: `${relayUrl.replace(/^http/, "ws")}/hosts/${routingKey}`, - token: options.userJwt, + token: userJwt, }, }; } From 9721f00f9e3c255458a26cb84bd8ff6d9c868a7e Mon Sep 17 00:00:00 2001 From: Krishna Date: Mon, 5 Oct 2026 02:29:25 +0530 Subject: [PATCH 2/4] fix(cli): refresh remote host credentials --- .../cli/src/commands/browser/cdp/command.ts | 3 +- .../cli/src/lib/host-target/exchangeApiKey.ts | 53 +++++++++++- .../lib/host-target/resolveHostTarget.test.ts | 84 ++++++++++++++++++- .../src/lib/host-target/resolveHostTarget.ts | 13 +-- 4 files changed, 142 insertions(+), 11 deletions(-) diff --git a/packages/cli/src/commands/browser/cdp/command.ts b/packages/cli/src/commands/browser/cdp/command.ts index 7fe99a37877..96450983537 100644 --- a/packages/cli/src/commands/browser/cdp/command.ts +++ b/packages/cli/src/commands/browser/cdp/command.ts @@ -23,9 +23,10 @@ export default command({ "Run: superset browser list --workspace ", ); } + const token = ws.getToken ? await ws.getToken() : ws.token; const url = `${ws.baseWsUrl}/browser/${encodeURIComponent( options.pane, - )}/cdp?workspaceId=${encodeURIComponent(options.workspace)}&token=${encodeURIComponent(ws.token)}`; + )}/cdp?workspaceId=${encodeURIComponent(options.workspace)}&token=${encodeURIComponent(token)}`; return { data: { url }, // The URL embeds a bearer token — treat it as a credential (keep it out diff --git a/packages/cli/src/lib/host-target/exchangeApiKey.ts b/packages/cli/src/lib/host-target/exchangeApiKey.ts index 31735a68468..6fd3d6302e9 100644 --- a/packages/cli/src/lib/host-target/exchangeApiKey.ts +++ b/packages/cli/src/lib/host-target/exchangeApiKey.ts @@ -4,8 +4,40 @@ import { env } from "../env"; const tokenResponse = z.object({ token: z.string().trim().min(1) }); +function isApiKey(bearer: string): boolean { + return bearer.startsWith("sk_live_") || bearer.startsWith("sk_test_"); +} + +export function createHostTokenProvider(bearer: string): () => Promise { + let cached: { token: string; expiresAt: number } | undefined; + let pending: Promise | undefined; + return async () => { + if (!isApiKey(bearer)) return bearer; + if (cached && Date.now() < cached.expiresAt - 30_000) return cached.token; + pending ??= exchangeApiKey(bearer) + .then((token) => { + let expiresAt = 0; + try { + const payload: unknown = JSON.parse( + Buffer.from(token.split(".")[1] ?? "", "base64url").toString(), + ); + const parsed = z + .object({ exp: z.number().finite() }) + .safeParse(payload); + if (parsed.success) expiresAt = parsed.data.exp * 1000; + } catch {} + cached = { token, expiresAt }; + return token; + }) + .finally(() => { + pending = undefined; + }); + return pending; + }; +} + export async function exchangeApiKey(bearer: string): Promise { - if (!bearer.startsWith("sk_live_")) return bearer; + if (!isApiKey(bearer)) return bearer; let response: Response; try { @@ -29,9 +61,22 @@ export async function exchangeApiKey(bearer: string): Promise { ); } - const parsed = tokenResponse.safeParse( - await response.json().catch(() => null), - ); + let body: string; + try { + body = await response.text(); + } catch { + throw new CLIError( + "Could not exchange API key for remote host access", + "Check connectivity to the Superset API and try again", + ); + } + let data: unknown; + try { + data = JSON.parse(body); + } catch { + throw new CLIError("Superset API returned an invalid remote host token"); + } + const parsed = tokenResponse.safeParse(data); if (!parsed.success) { throw new CLIError("Superset API returned an invalid remote host token"); } diff --git a/packages/cli/src/lib/host-target/resolveHostTarget.test.ts b/packages/cli/src/lib/host-target/resolveHostTarget.test.ts index 7fdfd7bab67..6f46e0f3dad 100644 --- a/packages/cli/src/lib/host-target/resolveHostTarget.test.ts +++ b/packages/cli/src/lib/host-target/resolveHostTarget.test.ts @@ -4,7 +4,7 @@ import { env } from "../env"; import { resolveHostTarget } from "./resolveHostTarget"; const apiKey = ["sk", "live", "remote", "host", "fixture"].join("_"); -const jwt = "header.payload.signature"; +const jwt = `header.${Buffer.from(JSON.stringify({ exp: 4_000_000_000 })).toString("base64url")}.signature`; const requests: { url: string; headers: Headers }[] = []; let tokenResponse = () => Response.json({ token: jwt }); const server = Bun.serve({ @@ -44,6 +44,88 @@ function resolve(bearer = apiKey) { } describe("resolveHostTarget remote authentication", () => { + test("exchanges test API keys", async () => { + const key = ["sk", "test", "fixture"].join("_"); + const target = await resolve(key); + await target.client.workspace.list.query(); + expect(requests[0]?.headers.get("x-api-key")).toBe(key); + expect(requests.at(-1)?.headers.get("authorization")).toBe(`Bearer ${jwt}`); + }); + + test.each([ + "HTTP", + "WebSocket", + ])("refreshes expiring credentials for %s connections", async (transport) => { + const clock = spyOn(Date, "now").mockReturnValue(1_000_000); + const first = `header.${Buffer.from(JSON.stringify({ exp: 4600 })).toString("base64url")}.signature`; + tokenResponse = () => Response.json({ token: first }); + try { + const target = await resolve(); + await target.client.workspace.list.query(); + expect( + requests.filter((r) => r.url.endsWith("/api/auth/token")), + ).toHaveLength(1); + clock.mockReturnValue(4_600_000); + tokenResponse = () => Response.json({ token: jwt }); + if (transport === "WebSocket") + expect(await target.ws.getToken?.()).toBe(jwt); + await target.client.workspace.list.query(); + expect(requests.at(-1)?.headers.get("authorization")).toBe( + `Bearer ${jwt}`, + ); + expect(await target.ws.getToken?.()).toBe(jwt); + expect( + requests.filter((r) => r.url.endsWith("/api/auth/token")), + ).toHaveLength(2); + } finally { + clock.mockRestore(); + } + }); + + test("shares concurrent refreshes and retries a failed exchange", async () => { + const clock = spyOn(Date, "now").mockReturnValue(1_000_000); + const first = `header.${Buffer.from(JSON.stringify({ exp: 4600 })).toString("base64url")}.signature`; + tokenResponse = () => Response.json({ token: first }); + try { + const target = await resolve(); + clock.mockReturnValue(4_600_000); + tokenResponse = () => new Response(null, { status: 503 }); + await expect(target.ws.getToken?.()).rejects.toThrow("HTTP 503"); + tokenResponse = () => Response.json({ token: jwt }); + expect( + await Promise.all([target.ws.getToken?.(), target.ws.getToken?.()]), + ).toEqual([jwt, jwt]); + expect( + requests.filter((r) => r.url.endsWith("/api/auth/token")), + ).toHaveLength(3); + } finally { + clock.mockRestore(); + } + }); + + test("reports body transport failures as connectivity errors", async () => { + const read = spyOn(Response.prototype, "text").mockRejectedValue( + new TypeError("connection closed"), + ); + const json = spyOn(Response.prototype, "json").mockRejectedValue( + new TypeError("connection closed"), + ); + try { + await expect(resolve()).rejects.toThrow( + "Could not exchange API key for remote host access", + ); + } finally { + read.mockRestore(); + json.mockRestore(); + } + }); + + test("reports completed malformed JSON as an invalid token", async () => { + tokenResponse = () => new Response("invalid json"); + await expect(resolve()).rejects.toThrow( + "Superset API returned an invalid remote host token", + ); + }); test("exchanges an API key for HTTP and WebSocket credentials", async () => { const target = await resolve(); expect(await target.client.workspace.list.query()).toEqual([]); diff --git a/packages/cli/src/lib/host-target/resolveHostTarget.ts b/packages/cli/src/lib/host-target/resolveHostTarget.ts index a1cb3038fb5..2cfe2d63313 100644 --- a/packages/cli/src/lib/host-target/resolveHostTarget.ts +++ b/packages/cli/src/lib/host-target/resolveHostTarget.ts @@ -10,7 +10,7 @@ import SuperJSON from "superjson"; import type { ApiClient } from "../api-client"; import { isProcessAlive, readManifest } from "../host/manifest"; import { getRelayUrl } from "../host/relay-url"; -import { exchangeApiKey } from "./exchangeApiKey"; +import { createHostTokenProvider } from "./exchangeApiKey"; import { readJwtSubject } from "./readJwtSubject"; export type HostServiceClient = ReturnType< @@ -23,6 +23,7 @@ export interface HostWsEndpoint { baseWsUrl: string; /** Passed as the `?token=` query param on WS routes. */ token: string; + getToken?: () => Promise; } export type ResolvedHostTarget = { @@ -93,7 +94,8 @@ export async function resolveHostTarget( } const routingKey = buildHostRoutingKey(options.organizationId, targetHostId); - const userJwt = await exchangeApiKey(options.userJwt); + const getToken = createHostTokenProvider(options.userJwt); + const userJwt = await getToken(); const relayUrl = await getRelayUrl(options.api); return { kind: "remote", @@ -103,16 +105,17 @@ export async function resolveHostTarget( httpBatchLink({ url: `${relayUrl}/hosts/${routingKey}/trpc`, transformer: SuperJSON, - headers: { - Authorization: `Bearer ${userJwt}`, + headers: async () => ({ + Authorization: `Bearer ${await getToken()}`, "x-superset-client-machine-id": localHostId, - }, + }), }), ], }), ws: { baseWsUrl: `${relayUrl.replace(/^http/, "ws")}/hosts/${routingKey}`, token: userJwt, + getToken, }, }; } From bf8baf8245772cb39d3748fd1d1efbe89bee2ae3 Mon Sep 17 00:00:00 2001 From: Krishna Date: Mon, 5 Oct 2026 10:17:44 +0530 Subject: [PATCH 3/4] fix(cli): keep exported remote CDP URLs usable --- apps/docs/content/docs/browser.mdx | 10 + apps/docs/content/docs/cli/cli-reference.mdx | 19 +- .../src/commands/browser/cdp-proxy/command.ts | 17 + .../src/commands/browser/cdp-stop/command.ts | 16 + .../src/commands/browser/cdp/command.test.ts | 182 ++++++++++ .../cli/src/commands/browser/cdp/command.ts | 30 +- .../commands/browser/cdp/fixtures/export.ts | 21 ++ .../src/commands/browser/cdp/fixtures/stop.ts | 8 + packages/cli/src/commands/browser/shared.ts | 9 +- packages/cli/src/lib/cdp-proxy/client.ts | 212 +++++++++++ .../cli/src/lib/cdp-proxy/fixtures/lock.ts | 6 + .../cli/src/lib/cdp-proxy/registry.test.ts | 137 +++++++ packages/cli/src/lib/cdp-proxy/registry.ts | 121 +++++++ packages/cli/src/lib/cdp-proxy/server.test.ts | 339 ++++++++++++++++++ packages/cli/src/lib/cdp-proxy/server.ts | 174 +++++++++ packages/cli/src/lib/cdp-proxy/worker-main.ts | 11 + packages/cli/src/lib/cdp-proxy/worker.test.ts | 109 ++++++ packages/cli/src/lib/cdp-proxy/worker.ts | 88 +++++ .../cli/src/lib/host-target/exchangeApiKey.ts | 20 +- 19 files changed, 1516 insertions(+), 13 deletions(-) create mode 100644 packages/cli/src/commands/browser/cdp-proxy/command.ts create mode 100644 packages/cli/src/commands/browser/cdp-stop/command.ts create mode 100644 packages/cli/src/commands/browser/cdp/command.test.ts create mode 100644 packages/cli/src/commands/browser/cdp/fixtures/export.ts create mode 100644 packages/cli/src/commands/browser/cdp/fixtures/stop.ts create mode 100644 packages/cli/src/lib/cdp-proxy/client.ts create mode 100644 packages/cli/src/lib/cdp-proxy/fixtures/lock.ts create mode 100644 packages/cli/src/lib/cdp-proxy/registry.test.ts create mode 100644 packages/cli/src/lib/cdp-proxy/registry.ts create mode 100644 packages/cli/src/lib/cdp-proxy/server.test.ts create mode 100644 packages/cli/src/lib/cdp-proxy/server.ts create mode 100644 packages/cli/src/lib/cdp-proxy/worker-main.ts create mode 100644 packages/cli/src/lib/cdp-proxy/worker.test.ts create mode 100644 packages/cli/src/lib/cdp-proxy/worker.ts diff --git a/apps/docs/content/docs/browser.mdx b/apps/docs/content/docs/browser.mdx index a15beb0b638..13789ed9c4e 100644 --- a/apps/docs/content/docs/browser.mdx +++ b/apps/docs/content/docs/browser.mdx @@ -66,6 +66,16 @@ For full interaction (clicking, typing, scrolling, form flows), `superset browse Agents pick this up automatically through the [`superset:browser` skill](/skills). A pane runs against the browser's real, logged-in session; agents are instructed to treat it accordingly and to confirm before consequential actions. +For a remote host authenticated with an API key, `browser cdp` starts a local proxy that keeps the exported URL usable after the command exits. It refreshes the remote token when a client connects. Repeating the command with the same API key, API address, organization, host, workspace and pane reuses that proxy while the remote endpoint is unchanged. + +The URL is a secret capability for that pane. It listens only on the CLI machine's loopback interface; clients on another machine need deliberate port forwarding. The proxy keeps the API key in memory and stops after 24 hours without an active connection. Active connections keep it running. Stop it explicitly with the `proxyId` returned by `--json`: + +```bash +superset browser cdp-stop --id +``` + +Stopping does not require the original API key. Local-host and OAuth-authenticated endpoints retain their existing token lifetime. + ## Pane or standalone browser? The `superset:browser` skill is engine-generic with the in-app pane as the default: when a page is (or can be) open in a workspace pane (previewing a dev server, verifying UI you're watching, clicking through a flow), agents use the `superset browser` commands above. diff --git a/apps/docs/content/docs/cli/cli-reference.mdx b/apps/docs/content/docs/cli/cli-reference.mdx index ccfea92094c..52b2cd5a09b 100644 --- a/apps/docs/content/docs/cli/cli-reference.mdx +++ b/apps/docs/content/docs/cli/cli-reference.mdx @@ -1272,11 +1272,28 @@ Read the pane's captured console output. { flag: "--pane ", required: true, description: "Pane ID." }, { flag: "--host ", description: "Host the workspace lives on (default: this machine)." }, ]} - output={`{ url: string } // a CDP WebSocket URL`} + output={`{ url: string, proxyId?: string, idleTimeoutSeconds?: number }`} > Print a raw Chrome DevTools Protocol WebSocket endpoint for the pane, for browser-use / Playwright-class tools (mouse, keyboard, scroll, DOM, network). The URL embeds an auth token; treat it as a secret. +For remote hosts authenticated with an API key, the URL connects through a local +proxy that refreshes remote credentials after this command exits. It works on +the CLI machine; remote consumers need deliberate port forwarding. Repeating +the same target and credentials reuses the proxy. It exits after 24 hours without +an active connection. Use `proxyId` to stop it sooner. + + +", required: true, description: "Proxy ID returned by browser cdp." }, + ]} + output={`{ proxyId: string, stopped: boolean }`} +> +Stop a local exported CDP proxy and disconnect its clients. This command uses +the local proxy's management credential so it works even if the original API +key has been removed or revoked. Returns `stopped: false` if no proxy was found. --- diff --git a/packages/cli/src/commands/browser/cdp-proxy/command.ts b/packages/cli/src/commands/browser/cdp-proxy/command.ts new file mode 100644 index 00000000000..e2314954d54 --- /dev/null +++ b/packages/cli/src/commands/browser/cdp-proxy/command.ts @@ -0,0 +1,17 @@ +import { string } from "@superset/cli-framework"; +import { runCdpProxyWorker } from "../../../lib/cdp-proxy/worker"; +import { command } from "../../../lib/command"; + +export default command({ + description: "Run the local CDP proxy worker", + hidden: true, + skipMiddleware: true, + options: { instance: string().required() }, + run: async ({ signal, options }) => { + try { + await runCdpProxyWorker(signal, options.instance); + } catch { + throw new Error("CDP proxy worker failed"); + } + }, +}); diff --git a/packages/cli/src/commands/browser/cdp-stop/command.ts b/packages/cli/src/commands/browser/cdp-stop/command.ts new file mode 100644 index 00000000000..e5604dd2e3e --- /dev/null +++ b/packages/cli/src/commands/browser/cdp-stop/command.ts @@ -0,0 +1,16 @@ +import { string } from "@superset/cli-framework"; +import { stopCdpProxy } from "../../../lib/cdp-proxy/client"; +import { command } from "../../../lib/command"; + +export default command({ + description: "Stop an exported local CDP proxy", + skipMiddleware: true, + options: { id: string().required().desc("Proxy ID from browser cdp") }, + run: async ({ options }) => { + const stopped = await stopCdpProxy(options.id); + return { + data: { proxyId: options.id, stopped }, + message: stopped ? "CDP proxy stopped" : "No CDP proxy found", + }; + }, +}); diff --git a/packages/cli/src/commands/browser/cdp/command.test.ts b/packages/cli/src/commands/browser/cdp/command.test.ts new file mode 100644 index 00000000000..8b451e716cf --- /dev/null +++ b/packages/cli/src/commands/browser/cdp/command.test.ts @@ -0,0 +1,182 @@ +import { afterEach, expect, test } from "bun:test"; +import { + mkdtempSync, + readdirSync, + readFileSync, + rmSync, + statSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const homes: string[] = []; +const stops: Array<() => Promise> = []; + +afterEach(async () => { + for (const stop of stops.splice(0)) await stop(); + for (const home of homes.splice(0)) + rmSync(home, { recursive: true, force: true }); +}); + +const launchers = [ + { + name: "source", + argv: [ + process.execPath, + "run", + join(import.meta.dir, "fixtures/export.ts"), + ], + direct: true, + }, + { + name: "dev", + argv: [ + process.execPath, + join(import.meta.dir, "../../../../../cli-framework/src/bin.ts"), + "dev", + ], + direct: false, + }, + ...(process.env.CDP_TEST_BINARY + ? [{ name: "compiled", argv: [process.env.CDP_TEST_BINARY], direct: false }] + : []), +]; + +test.each( + launchers, +)("$name exported CDP URL survives CLI exit and the original JWT expiry", async (launcher) => { + let clock = Math.floor(Date.now() / 1000); + const originalClock = clock; + let exchanges = 0; + const server = Bun.serve({ + port: 0, + fetch(request, server) { + const url = new URL(request.url); + if (url.pathname === "/api/auth/token") { + exchanges++; + return Response.json({ + token: `header.${Buffer.from(JSON.stringify({ exp: clock + 3600 })).toString("base64url")}.signature`, + }); + } + if (url.pathname === "/api/trpc/host.relayEndpoint") + return Response.json([ + { result: { data: { json: { url: server.url.origin } } } }, + ]); + if (url.pathname.endsWith("/trpc/browser.list")) + return Response.json([ + { + result: { data: { json: { panes: [{ paneId: "pane-fixture" }] } } }, + }, + ]); + if (url.pathname.endsWith("/cdp")) { + const token = url.searchParams.get("token") ?? ""; + const payload = JSON.parse( + Buffer.from(token.split(".")[1] ?? "", "base64url").toString(), + ); + if (payload.exp <= clock) return new Response(null, { status: 401 }); + if (server.upgrade(request)) return; + } + return new Response(null, { status: 404 }); + }, + websocket: { + message(ws, message) { + ws.send(message); + }, + }, + }); + stops.push(async () => { + server.stop(true); + }); + const home = mkdtempSync(join(tmpdir(), "superset-cdp-export-")); + homes.push(home); + const child = Bun.spawn( + [ + ...launcher.argv, + ...(launcher.direct + ? [] + : [ + "browser", + "cdp", + "--host", + "remote-fixture", + "--workspace", + "workspace-fixture", + "--pane", + "pane-fixture", + "--json", + ]), + ], + { + cwd: join(import.meta.dir, "../../../.."), + env: { + ...process.env, + SUPERSET_HOME_DIR: home, + SUPERSET_API_URL: server.url.origin, + SUPERSET_API_KEY: ["sk", "test", "cdp", "fixture"].join("_"), + SUPERSET_ORGANIZATION_ID: "org-fixture", + }, + stdout: "pipe", + stderr: "pipe", + }, + ); + const output = await new Response(child.stdout).text(); + expect(await child.exited).toBe(0); + const parsed = JSON.parse(output); + const result = (launcher.direct ? parsed : { data: parsed }) as { + data: { url: string; proxyId?: string }; + }; + if (result.data.proxyId) { + const manifestPath = join( + home, + "cdp-proxies", + `${result.data.proxyId}.json`, + ); + const manifest = JSON.parse(readFileSync(manifestPath, "utf8")); + stops.unshift(async () => { + const stop = Bun.spawn( + [ + process.execPath, + "run", + join(import.meta.dir, "fixtures/stop.ts"), + manifest.id, + ], + { + env: { + ...process.env, + SUPERSET_HOME_DIR: home, + SUPERSET_API_KEY: undefined, + }, + stdout: "pipe", + stderr: "pipe", + }, + ); + expect(await stop.exited).toBe(0); + await expect(fetch(`${manifest.endpoint}/health`)).rejects.toThrow(); + }); + expect(statSync(manifestPath).mode & 0o777).toBe(0o600); + expect(readFileSync(manifestPath, "utf8")).not.toContain( + ["sk", "test", "cdp", "fixture"].join("_"), + ); + } + clock = originalClock + 3601; + const ws = new WebSocket(result.data.url); + const received = await new Promise((resolve, reject) => { + ws.onopen = () => ws.send('{"id":1,"method":"Runtime.enable"}'); + ws.onmessage = (event) => { + resolve(String(event.data)); + ws.close(); + }; + ws.onerror = () => + reject( + new Error("Exported CDP URL rejected after original token expiry"), + ); + }); + expect(received).toBe('{"id":1,"method":"Runtime.enable"}'); + expect(exchanges).toBeGreaterThan(1); + expect(result.data.url).toStartWith("ws://127.0.0.1:"); + expect( + readdirSync(join(home, "cdp-proxies")).filter((name) => + name.endsWith(".json"), + ), + ).toHaveLength(1); +}); diff --git a/packages/cli/src/commands/browser/cdp/command.ts b/packages/cli/src/commands/browser/cdp/command.ts index 96450983537..72f13be4257 100644 --- a/packages/cli/src/commands/browser/cdp/command.ts +++ b/packages/cli/src/commands/browser/cdp/command.ts @@ -1,4 +1,5 @@ import { CLIError, string } from "@superset/cli-framework"; +import { ensureCdpProxy } from "../../../lib/cdp-proxy/client"; import { command } from "../../../lib/command"; import { resolveBrowserTarget } from "../shared"; @@ -11,7 +12,10 @@ export default command({ pane: string().required().desc("Pane ID (from `superset browser list`)"), }, run: async ({ ctx, options }) => { - const { client, ws } = await resolveBrowserTarget(ctx, options); + const { client, ws, kind, hostId } = await resolveBrowserTarget( + ctx, + options, + ); // Verify the pane exists in this workspace before handing out a URL — a // dead/foreign pane id would otherwise fail only once the tool dials in. const { panes } = await client.browser.list.query({ @@ -23,10 +27,28 @@ export default command({ "Run: superset browser list --workspace ", ); } - const token = ws.getToken ? await ws.getToken() : ws.token; - const url = `${ws.baseWsUrl}/browser/${encodeURIComponent( + const endpoint = `${ws.baseWsUrl}/browser/${encodeURIComponent( options.pane, - )}/cdp?workspaceId=${encodeURIComponent(options.workspace)}&token=${encodeURIComponent(token)}`; + )}/cdp?workspaceId=${encodeURIComponent(options.workspace)}`; + if ( + kind === "remote" && + (ctx.bearer.startsWith("sk_live_") || ctx.bearer.startsWith("sk_test_")) + ) { + const proxy = await ensureCdpProxy({ + apiKey: ctx.bearer, + organizationId: ctx.config.organizationId!, + hostId, + workspaceId: options.workspace, + paneId: options.pane, + upstreamUrl: endpoint, + }); + return { + data: proxy, + message: proxy.url, + }; + } + const token = ws.getToken ? await ws.getToken() : ws.token; + const url = `${endpoint}&token=${encodeURIComponent(token)}`; return { data: { url }, // The URL embeds a bearer token — treat it as a credential (keep it out diff --git a/packages/cli/src/commands/browser/cdp/fixtures/export.ts b/packages/cli/src/commands/browser/cdp/fixtures/export.ts new file mode 100644 index 00000000000..49c3af06681 --- /dev/null +++ b/packages/cli/src/commands/browser/cdp/fixtures/export.ts @@ -0,0 +1,21 @@ +import { createApiClient } from "../../../../lib/api-client"; +import cdp from "../command"; + +const bearer = ["sk", "test", "cdp", "fixture"].join("_"); +const organizationId = "org-fixture"; +const result = await cdp.run({ + ctx: { + bearer, + authSource: "override", + config: { organizationId }, + api: createApiClient({ bearer, organizationId }), + }, + options: { + workspace: "workspace-fixture", + host: "remote-fixture", + pane: "pane-fixture", + }, + args: {}, + signal: new AbortController().signal, +}); +process.stdout.write(JSON.stringify(result)); diff --git a/packages/cli/src/commands/browser/cdp/fixtures/stop.ts b/packages/cli/src/commands/browser/cdp/fixtures/stop.ts new file mode 100644 index 00000000000..210b9a0348b --- /dev/null +++ b/packages/cli/src/commands/browser/cdp/fixtures/stop.ts @@ -0,0 +1,8 @@ +import stop from "../../cdp-stop/command"; + +await stop.run({ + ctx: {}, + options: { id: process.argv[2] ?? "" }, + args: {}, + signal: new AbortController().signal, +}); diff --git a/packages/cli/src/commands/browser/shared.ts b/packages/cli/src/commands/browser/shared.ts index a6897e0df1a..b46d735b028 100644 --- a/packages/cli/src/commands/browser/shared.ts +++ b/packages/cli/src/commands/browser/shared.ts @@ -14,7 +14,12 @@ import { export async function resolveBrowserTarget( ctx: CliContext, options: { workspace: string; host?: string | null }, -): Promise<{ client: HostServiceClient; hostId: string; ws: HostWsEndpoint }> { +): Promise<{ + client: HostServiceClient; + hostId: string; + ws: HostWsEndpoint; + kind: "local" | "remote" | "cloud"; +}> { const organizationId = ctx.config.organizationId; if (!organizationId) { throw new CLIError("No active organization", "Run: superset auth login"); @@ -26,5 +31,5 @@ export async function resolveBrowserTarget( userJwt: ctx.bearer, api: ctx.api, }); - return { client: target.client, hostId, ws: target.ws }; + return { client: target.client, hostId, ws: target.ws, kind: target.kind }; } diff --git a/packages/cli/src/lib/cdp-proxy/client.ts b/packages/cli/src/lib/cdp-proxy/client.ts new file mode 100644 index 00000000000..4444284e7ee --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/client.ts @@ -0,0 +1,212 @@ +import { spawn } from "node:child_process"; +import { createHmac, randomBytes } from "node:crypto"; +import { fileURLToPath } from "node:url"; +import { CLIError } from "@superset/cli-framework"; +import { env } from "../env"; +import { + type ProxyManifest, + proxyDirectory, + proxyIsLive, + proxyManifest, + proxyProcessIsAlive, + readProxy, + removeProxy, + tryProxyLock, + writeProxy, +} from "./registry"; +import { CDP_PROXY_IDLE_MS } from "./server"; + +export interface CdpProxyOptions { + apiKey: string; + organizationId: string; + hostId: string; + workspaceId: string; + paneId: string; + upstreamUrl: string; +} + +export function cdpProxyId(options: CdpProxyOptions): string { + return createHmac("sha256", options.apiKey) + .update( + JSON.stringify([ + "cdp-proxy-v1", + new URL(env.SUPERSET_API_URL).href.replace(/\/$/, ""), + options.organizationId, + options.hostId, + options.workspaceId, + options.paneId, + new URL(options.upstreamUrl).href, + ]), + ) + .digest("hex"); +} + +async function lockProxy(id: string) { + const deadline = Date.now() + 12_000; + do { + const release = tryProxyLock(id); + if (release) return release; + await new Promise((resolve) => setTimeout(resolve, 25)); + } while (Date.now() < deadline); + throw new CLIError( + "CDP proxy startup is locked", + "Retry once the other invocation finishes", + ); +} + +async function launchProxy( + options: CdpProxyOptions, + id: string, +): Promise { + const instance = randomBytes(32).toString("hex"); + const child = spawn( + process.execPath, + Bun.main.includes("$bunfs") + ? ["browser", "cdp-proxy", "--instance", instance] + : [ + fileURLToPath(new URL("./worker-main.ts", import.meta.url)), + "--instance", + instance, + ], + { + detached: true, + cwd: proxyDirectory, + stdio: ["pipe", "pipe", "ignore"], + env: { + PATH: process.env.PATH, + HOME: process.env.HOME, + USERPROFILE: process.env.USERPROFILE, + SYSTEMROOT: process.env.SYSTEMROOT, + SUPERSET_HOME_DIR: process.env.SUPERSET_HOME_DIR, + SUPERSET_API_URL: env.SUPERSET_API_URL, + }, + }, + ); + const exited = new Promise((resolve) => + child.once("exit", () => resolve()), + ); + let manifest: ProxyManifest | undefined; + try { + manifest = await new Promise((resolve, reject) => { + const timeout = setTimeout( + () => reject(new Error("Startup timeout")), + 10_000, + ); + const fail = () => { + clearTimeout(timeout); + reject(new Error("Proxy startup failed")); + }; + let output = ""; + child.once("error", fail); + child.once("exit", fail); + child.stdin.on("error", fail); + child.stdout.on("data", (data: Buffer) => { + output += data.toString(); + if (output.length > 16_384) { + fail(); + return; + } + if (!output.includes("\n")) return; + try { + const parsed = proxyManifest.parse( + JSON.parse(output.slice(0, output.indexOf("\n"))), + ); + if ( + parsed.id !== id || + parsed.pid !== child.pid || + parsed.instance !== instance + ) { + fail(); + return; + } + clearTimeout(timeout); + resolve(parsed); + } catch { + fail(); + } + }); + child.stdin.write( + `${JSON.stringify({ id, apiKey: options.apiKey, upstreamUrl: options.upstreamUrl })}\n`, + ); + }); + writeProxy(manifest); + await new Promise((resolve, reject) => + child.stdin.end("ready\n", (error?: Error) => + error ? reject(error) : resolve(), + ), + ); + child.stdout.destroy(); + child.unref(); + return manifest; + } catch { + child.stdin.destroy(); + if (child.pid && child.exitCode === null && child.signalCode === null) { + child.kill("SIGTERM"); + const force = setTimeout(() => child.kill("SIGKILL"), 1000); + await exited; + clearTimeout(force); + } + if (manifest) removeProxy(manifest); + throw new CLIError( + "Could not start the local CDP proxy", + "Check local CLI permissions and try again", + ); + } finally { + child.stdin.destroy(); + child.stdout.destroy(); + } +} + +export async function ensureCdpProxy(options: CdpProxyOptions) { + const id = cdpProxyId(options); + const release = await lockProxy(id); + try { + const existing = readProxy(id); + const live = existing && (await proxyIsLive(existing)); + if (existing && !live && (await proxyProcessIsAlive(existing))) + throw new CLIError( + "The existing CDP proxy is not responding", + `Retry or stop it with: superset browser cdp-stop --id ${id}`, + ); + const manifest = + existing && live ? existing : await launchProxy(options, id); + return { + url: `${manifest.endpoint.replace(/^http/, "ws")}/cdp?token=${manifest.token}`, + proxyId: id, + idleTimeoutSeconds: CDP_PROXY_IDLE_MS / 1000, + }; + } finally { + release(); + } +} + +export async function stopCdpProxy(id: string): Promise { + const release = await lockProxy(id); + try { + const manifest = readProxy(id); + if (!manifest) return false; + if (await proxyProcessIsAlive(manifest)) { + const response = await fetch(`${manifest.endpoint}/stop`, { + method: "POST", + headers: { Authorization: `Bearer ${manifest.stopToken}` }, + signal: AbortSignal.timeout(1000), + redirect: "error", + }); + if (!response.ok) + throw new CLIError("Could not stop the local CDP proxy"); + const deadline = Date.now() + 2000; + while (await proxyProcessIsAlive(manifest)) { + if (Date.now() >= deadline) + throw new CLIError( + "The CDP proxy has not stopped yet", + "Retry the stop command", + ); + await new Promise((resolve) => setTimeout(resolve, 10)); + } + } + removeProxy(manifest); + return true; + } finally { + release(); + } +} diff --git a/packages/cli/src/lib/cdp-proxy/fixtures/lock.ts b/packages/cli/src/lib/cdp-proxy/fixtures/lock.ts new file mode 100644 index 00000000000..f63ab541f95 --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/fixtures/lock.ts @@ -0,0 +1,6 @@ +import { tryProxyLock } from "../registry"; + +const release = tryProxyLock("a".repeat(64)); +if (!release) throw new Error("Lock unavailable"); +process.stdout.write("locked\n"); +setInterval(() => {}, 1000); diff --git a/packages/cli/src/lib/cdp-proxy/registry.test.ts b/packages/cli/src/lib/cdp-proxy/registry.test.ts new file mode 100644 index 00000000000..93ae13d0c3d --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/registry.test.ts @@ -0,0 +1,137 @@ +import { afterAll, expect, test } from "bun:test"; +import { + mkdtempSync, + readFileSync, + rmSync, + statSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const home = mkdtempSync(join(tmpdir(), "superset-cdp-registry-")); +const previousHome = process.env.SUPERSET_HOME_DIR; +process.env.SUPERSET_HOME_DIR = home; +const { + proxyDirectory, + tryProxyLock, + readProxy, + writeProxy, + proxyProcessIsAlive, +} = await import("./registry"); +const { cdpProxyId, ensureCdpProxy, stopCdpProxy } = await import("./client"); +const { env } = await import("../env"); +const options = { + apiKey: ["sk", "test", "registry", "fixture"].join("_"), + organizationId: "org", + hostId: "host", + workspaceId: "workspace", + paneId: "pane", + upstreamUrl: "ws://127.0.0.1:1/cdp?workspaceId=workspace", +}; + +afterAll(() => { + process.env.SUPERSET_HOME_DIR = previousHome; + rmSync(home, { recursive: true, force: true }); +}); + +test("a crashed lock owner releases the startup transaction", async () => { + const child = Bun.spawn( + [process.execPath, join(import.meta.dir, "fixtures/lock.ts")], + { + env: { ...process.env, SUPERSET_HOME_DIR: home }, + stdout: "pipe", + stderr: "pipe", + }, + ); + try { + await child.stdout.getReader().read(); + expect(tryProxyLock("a".repeat(64))).toBeNull(); + child.kill("SIGKILL"); + await child.exited; + const release = tryProxyLock("a".repeat(64)); + expect(release).not.toBeNull(); + release?.(); + expect( + statSync(join(proxyDirectory, "launch-lock.sqlite")).mode & 0o777, + ).toBe(0o600); + } finally { + if (child.exitCode === null) child.kill("SIGKILL"); + await child.exited; + } +}); + +test("reuse identity isolates credentials, target routes and path-mounted API environments", () => { + const original = env.SUPERSET_API_URL; + try { + env.SUPERSET_API_URL = "https://example.com/one/"; + const id = cdpProxyId(options); + env.SUPERSET_API_URL = "https://example.com/one"; + expect(cdpProxyId(options)).toBe(id); + env.SUPERSET_API_URL = "https://example.com/two"; + expect(cdpProxyId(options)).not.toBe(id); + env.SUPERSET_API_URL = "https://example.com/one"; + for (const key of [ + "apiKey", + "organizationId", + "hostId", + "workspaceId", + "paneId", + "upstreamUrl", + ] as const) { + expect( + cdpProxyId({ ...options, [key]: `${options[key]}other` }), + ).not.toBe(id); + } + } finally { + env.SUPERSET_API_URL = original; + } +}); + +test("concurrent exports reuse one daemon and uncertain health retains its management handle", async () => { + const [first, second] = await Promise.all([ + ensureCdpProxy(options), + ensureCdpProxy(options), + ]); + const manifest = readProxy(first.proxyId)!; + try { + expect(first).toEqual(second); + expect(await proxyProcessIsAlive(manifest)).toBe(true); + expect( + statSync(join(proxyDirectory, `${manifest.id}.json`)).mode & 0o777, + ).toBe(0o600); + writeProxy({ ...manifest, stopToken: "e".repeat(64) }); + await expect(ensureCdpProxy(options)).rejects.toThrow("not responding"); + await expect(stopCdpProxy(manifest.id)).rejects.toThrow("Could not stop"); + expect(readProxy(manifest.id)?.pid).toBe(manifest.pid); + writeProxy(manifest); + expect( + readFileSync(join(proxyDirectory, `${manifest.id}.json`), "utf8"), + ).not.toContain(options.apiKey); + } finally { + writeProxy(manifest); + await stopCdpProxy(manifest.id); + } + expect(await proxyProcessIsAlive(manifest)).toBe(false); + expect(readProxy(manifest.id)).toBeNull(); +}); + +test("manifest identity is bound to its filename and stale PIDs cannot stop unrelated processes", async () => { + const id = cdpProxyId(options); + const manifest = { + id, + pid: process.pid, + instance: "d".repeat(64), + endpoint: "http://127.0.0.1:1", + token: "b".repeat(64), + stopToken: "c".repeat(64), + }; + writeProxy(manifest); + expect(await proxyProcessIsAlive(manifest)).toBe(false); + expect(await stopCdpProxy(id)).toBe(true); + writeFileSync( + join(proxyDirectory, `${id}.json`), + JSON.stringify({ ...manifest, id: "e".repeat(64) }), + ); + expect(readProxy(id)).toBeNull(); +}); diff --git a/packages/cli/src/lib/cdp-proxy/registry.ts b/packages/cli/src/lib/cdp-proxy/registry.ts new file mode 100644 index 00000000000..89ce33202f0 --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/registry.ts @@ -0,0 +1,121 @@ +import { Database } from "bun:sqlite"; +import { randomBytes } from "node:crypto"; +import { + chmodSync, + mkdirSync, + readFileSync, + renameSync, + unlinkSync, + writeFileSync, +} from "node:fs"; +import { join } from "node:path"; +import { z } from "zod"; +import { SUPERSET_HOME_DIR } from "../config"; +import { isProcessAlive } from "../host/manifest"; +import { inspectProcessCommand } from "../host/process-identity"; + +export const proxyId = z.string().regex(/^[a-f0-9]{64}$/); +const capability = z.string().regex(/^[a-f0-9]{64}$/); +export const proxyManifest = z.object({ + id: proxyId, + pid: z.number().int().positive(), + instance: capability, + endpoint: z.string().refine((value) => { + try { + const url = new URL(value); + return ( + url.protocol === "http:" && + url.hostname === "127.0.0.1" && + url.port !== "" && + url.origin === value + ); + } catch { + return false; + } + }), + token: capability, + stopToken: capability, +}); +export type ProxyManifest = z.infer; +export const proxyDirectory = join(SUPERSET_HOME_DIR, "cdp-proxies"); + +export function readProxy(id: string): ProxyManifest | null { + proxyId.parse(id); + try { + const manifest = proxyManifest.parse( + JSON.parse(readFileSync(join(proxyDirectory, `${id}.json`), "utf8")), + ); + return manifest.id === id ? manifest : null; + } catch { + return null; + } +} + +export function tryProxyLock(id: string): (() => void) | null { + proxyId.parse(id); + mkdirSync(proxyDirectory, { recursive: true, mode: 0o700 }); + chmodSync(proxyDirectory, 0o700); + const path = join(proxyDirectory, "launch-lock.sqlite"); + const database = new Database(path, { create: true }); + chmodSync(path, 0o600); + try { + database.exec("BEGIN IMMEDIATE"); + } catch (error) { + database.close(); + if ((error as { code?: string }).code === "SQLITE_BUSY") return null; + throw error; + } + return () => { + try { + database.exec("ROLLBACK"); + } finally { + database.close(); + } + }; +} + +export function writeProxy(manifest: ProxyManifest) { + const temporary = join( + proxyDirectory, + `.${manifest.id}.${randomBytes(8).toString("hex")}.tmp`, + ); + try { + writeFileSync(temporary, JSON.stringify(manifest), { mode: 0o600 }); + renameSync(temporary, join(proxyDirectory, `${manifest.id}.json`)); + } catch (error) { + try { + unlinkSync(temporary); + } catch {} + throw error; + } +} + +export async function proxyProcessIsAlive( + manifest: ProxyManifest, +): Promise { + if (!isProcessAlive(manifest.pid)) return false; + const command = await inspectProcessCommand(manifest.pid); + return ( + command === null || command.includes(`--instance ${manifest.instance}`) + ); +} + +export function removeProxy(manifest: ProxyManifest) { + const current = readProxy(manifest.id); + if (current?.pid === manifest.pid && current.stopToken === manifest.stopToken) + unlinkSync(join(proxyDirectory, `${manifest.id}.json`)); +} + +export async function proxyIsLive(manifest: ProxyManifest): Promise { + try { + const response = await fetch(`${manifest.endpoint}/health`, { + headers: { Authorization: `Bearer ${manifest.stopToken}` }, + signal: AbortSignal.timeout(500), + redirect: "error", + }); + const body = (await response.json()) as { id?: string; pid?: number }; + return response.ok && body.id === manifest.id && body.pid === manifest.pid; + } catch { + return false; + } +} diff --git a/packages/cli/src/lib/cdp-proxy/server.test.ts b/packages/cli/src/lib/cdp-proxy/server.test.ts new file mode 100644 index 00000000000..643c037b58f --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/server.test.ts @@ -0,0 +1,339 @@ +import { afterEach, expect, test } from "bun:test"; +import { createHash } from "node:crypto"; +import { createServer, type Socket } from "node:net"; +import { env } from "../env"; +import { createHostTokenProvider } from "../host-target/exchangeApiKey"; +import { startCdpProxy } from "./server"; + +const token = "a".repeat(64); +const stopToken = "b".repeat(64); +const originalApiUrl = env.SUPERSET_API_URL; +const cleanup: Array<() => void> = []; +afterEach(() => { + for (const stop of cleanup.splice(0)) stop(); + env.SUPERSET_API_URL = originalApiUrl; +}); + +function echo() { + const server = Bun.serve({ + port: 0, + fetch(request, server) { + if (server.upgrade(request)) return; + return new Response(null, { status: 400 }); + }, + websocket: { + message(ws, message) { + ws.send(message); + }, + }, + }); + cleanup.push(() => server.stop(true)); + return `${server.url.origin.replace("http", "ws")}/browser/pane/cdp?workspaceId=workspace`; +} + +function proxy(overrides: Partial[0]> = {}) { + const service = startCdpProxy({ + id: "test", + token, + stopToken, + upstreamUrl: echo(), + getToken: async () => "upstream", + ...overrides, + }); + cleanup.unshift(service.stop); + return { + ...service, + url: `${service.endpoint.replace("http", "ws")}/cdp?token=${token}`, + }; +} + +function connect(url: string) { + const ws = new WebSocket(url); + cleanup.unshift(() => ws.close()); + return ws; +} + +function exchange(url: string, message = "message") { + const ws = connect(url); + return new Promise((resolve, reject) => { + ws.onopen = () => ws.send(message); + ws.onmessage = (event) => { + resolve(String(event.data)); + ws.close(); + }; + ws.onerror = () => reject(new Error("WebSocket connection failed")); + }); +} + +test("refreshes the cached upstream JWT on a later connection to the same URL", async () => { + let clock = 1_000_000; + const received: string[] = []; + let exchanges = 0; + const api = Bun.serve({ + port: 0, + fetch() { + exchanges++; + return Response.json({ + token: `header.${Buffer.from(JSON.stringify({ exp: clock / 1000 + 3600 })).toString("base64url")}.signature`, + }); + }, + }); + cleanup.push(() => api.stop(true)); + env.SUPERSET_API_URL = api.url.origin; + const relay = Bun.serve({ + port: 0, + fetch(request, server) { + const url = new URL(request.url); + const jwt = url.searchParams.get("token") ?? ""; + received.push(jwt); + const payload = JSON.parse( + Buffer.from(jwt.split(".")[1] ?? "", "base64url").toString(), + ); + if (payload.exp <= clock / 1000) + return new Response(null, { status: 401 }); + expect(url.pathname).toBe("/browser/pane/cdp"); + expect(url.searchParams.get("workspaceId")).toBe("workspace"); + if (server.upgrade(request)) return; + return new Response(null, { status: 400 }); + }, + websocket: { + message(ws, message) { + ws.send(message); + }, + }, + }); + cleanup.push(() => relay.stop(true)); + const service = proxy({ + upstreamUrl: `${relay.url.origin.replace("http", "ws")}/browser/pane/cdp?workspaceId=workspace`, + getToken: createHostTokenProvider( + ["sk", "test", "fixture"].join("_"), + () => clock, + ), + }); + expect(await exchange(service.url)).toBe("message"); + clock += 3_601_000; + expect( + await exchange(`${service.url}&workspaceId=another&upstreamUrl=elsewhere`), + ).toBe("message"); + expect(exchanges).toBe(2); + expect(received[0]).not.toBe(received[1]); +}); + +test("rejects wrong capabilities, browser origins and foreign Host headers", async () => { + const service = proxy(); + expect((await fetch(`${service.endpoint}/cdp?token=wrong`)).status).toBe(401); + expect( + ( + await fetch(service.url.replace("ws", "http"), { + headers: { Origin: "https://example.com" }, + }) + ).status, + ).toBe(403); + expect( + ( + await fetch(service.url.replace("ws", "http"), { + headers: { Host: "example.com" }, + }) + ).status, + ).toBe(403); + expect( + ( + await fetch(`${service.endpoint}/stop`, { + method: "POST", + headers: { Authorization: `Bearer ${token}` }, + }) + ).status, + ).toBe(401); + expect( + ( + await fetch(`${service.endpoint}/health`, { + headers: { Authorization: `Bearer ${stopToken}` }, + }) + ).status, + ).toBe(200); + expect( + ( + await fetch(`${service.endpoint}/stop`, { + method: "POST", + headers: { Authorization: `Bearer ${stopToken}` }, + }) + ).status, + ).toBe(200); + await service.closed; + await expect(fetch(`${service.endpoint}/health`)).rejects.toThrow(); +}); + +test("redacts failed upstream authentication and closes the client", async () => { + const api = Bun.serve({ + port: 0, + fetch: () => new Response("private key details", { status: 401 }), + }); + cleanup.push(() => api.stop(true)); + env.SUPERSET_API_URL = api.url.origin; + const service = proxy({ + getToken: createHostTokenProvider( + ["sk", "test", "rejected", "fixture"].join("_"), + ), + }); + const ws = connect(service.url); + const event = await new Promise((resolve) => { + ws.onclose = resolve; + }); + expect(event.code).toBe(1011); + expect(event.reason).toBe("CDP authentication failed"); +}); + +test.each([ + "client", + "upstream", +])("propagates %s closure without illegal close codes", async (side) => { + let ready: () => void = () => {}; + let ended: () => void = () => {}; + const opened = new Promise((resolve) => { + ready = resolve; + }); + const upstreamClosed = new Promise((resolve) => { + ended = resolve; + }); + const upstream = Bun.serve({ + port: 0, + fetch(request, server) { + if (server.upgrade(request)) return; + return new Response(null, { status: 400 }); + }, + websocket: { + open() { + ready(); + }, + message(ws) { + ws.terminate(); + }, + close() { + ended(); + }, + }, + }); + cleanup.push(() => upstream.stop(true)); + const service = proxy({ + upstreamUrl: upstream.url.origin.replace("http", "ws"), + }); + const ws = connect(service.url); + const closed = new Promise((resolve) => { + ws.onclose = resolve; + }); + await opened; + if (side === "client") ws.close(); + else ws.send("close"); + expect((await closed).code).toBe(1000); + await upstreamClosed; +}); + +test("bounds frames queued while obtaining a token and ignores late completion", async () => { + let release: (value: string) => void = () => {}; + const service = proxy({ + getToken: () => + new Promise((resolve) => { + release = resolve; + }), + maxPendingBytes: 8, + }); + const ws = connect(service.url); + const closed = new Promise((resolve) => { + ws.onclose = resolve; + }); + ws.onopen = () => { + ws.send("12345"); + ws.send("67890"); + }; + expect((await closed).code).toBe(1009); + release("upstream"); +}); + +test("does not apply the preconnect queue limit to a connected large CDP frame", async () => { + const service = proxy({ maxPendingBytes: 8 }); + const ws = connect(service.url); + const result = await new Promise((resolve) => { + ws.onopen = () => ws.send("ready"); + ws.onmessage = (event) => { + if (event.data === "ready") ws.send("x".repeat(4 * 1024 * 1024 + 1)); + else resolve(String(event.data)); + }; + }); + expect(result.length).toBe(4 * 1024 * 1024 + 1); +}); + +test("idle cleanup waits for active clients to disconnect", async () => { + const service = proxy({ idleTimeoutMs: 30 }); + const ws = connect(service.url); + await new Promise((resolve) => { + ws.onopen = () => resolve(); + }); + await Bun.sleep(40); + expect( + ( + await fetch(`${service.endpoint}/health`, { + headers: { Authorization: `Bearer ${stopToken}` }, + }) + ).status, + ).toBe(200); + ws.close(); + await service.closed; + await expect(fetch(`${service.endpoint}/health`)).rejects.toThrow(); +}); + +test("a connection timeout releases the idle lifetime", async () => { + const service = proxy({ + getToken: () => new Promise(() => {}), + connectTimeoutMs: 10, + idleTimeoutMs: 10, + }); + const ws = connect(service.url); + const closed = await new Promise((resolve) => { + ws.onclose = resolve; + }); + expect(closed.code).toBe(1011); + await service.closed; +}); + +test("stopping terminates an upstream that never acknowledges WebSocket close", async () => { + let upstream: Socket | undefined; + let accepted: () => void = () => {}; + const connected = new Promise((resolve) => { + accepted = resolve; + }); + const tcp = createServer((socket) => { + upstream = socket; + socket.once("data", (data) => { + const key = + data.toString().match(/Sec-WebSocket-Key: (.+)\r\n/i)?.[1] ?? ""; + const accept = createHash("sha1") + .update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`) + .digest("base64"); + socket.write( + `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`, + ); + accepted(); + }); + }); + await new Promise((resolve) => tcp.listen(0, "127.0.0.1", resolve)); + try { + const address = tcp.address(); + if (!address || typeof address === "string") + throw new Error("Missing TCP address"); + const service = proxy({ + upstreamUrl: `ws://127.0.0.1:${address.port}/cdp`, + }); + connect(service.url); + await connected; + const closed = new Promise((resolve) => + upstream!.once("close", () => resolve(true)), + ); + service.stop(); + expect(await Promise.race([closed, Bun.sleep(250).then(() => false)])).toBe( + true, + ); + } finally { + upstream?.destroy(); + tcp.close(); + } +}); diff --git a/packages/cli/src/lib/cdp-proxy/server.ts b/packages/cli/src/lib/cdp-proxy/server.ts new file mode 100644 index 00000000000..05134c272f9 --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/server.ts @@ -0,0 +1,174 @@ +import { timingSafeEqual } from "node:crypto"; +import type { ServerWebSocket } from "bun"; + +export const CDP_PROXY_IDLE_MS = 24 * 60 * 60 * 1000; +const MAX_PENDING_BYTES = 4 * 1024 * 1024; + +interface Connection { + upstream?: WebSocket; + pending: Array; + bytes: number; + closed: boolean; + timer?: ReturnType; +} + +export function startCdpProxy(options: { + id: string; + token: string; + stopToken: string; + upstreamUrl: string; + getToken: (signal: AbortSignal) => Promise; + idleTimeoutMs?: number; + connectTimeoutMs?: number; + maxPendingBytes?: number; + maxBufferedBytes?: number; +}) { + const connections = new Set>(); + const lifetime = new AbortController(); + const maxBytes = options.maxPendingBytes ?? MAX_PENDING_BYTES; + const maxBufferedBytes = options.maxBufferedBytes ?? 64 * 1024 * 1024; + let idleTimer: ReturnType; + let stopped = false; + let resolveClosed: () => void; + const closed = new Promise((resolve) => { + resolveClosed = resolve; + }); + + function authenticated(actual: string | null, expected: string) { + return ( + actual !== null && + Buffer.byteLength(actual) === Buffer.byteLength(expected) && + timingSafeEqual(Buffer.from(actual), Buffer.from(expected)) + ); + } + + function close(ws: ServerWebSocket, code = 1000, reason = "") { + if (ws.data.closed) return; + ws.data.closed = true; + clearTimeout(ws.data.timer); + ws.data.pending.length = 0; + ws.data.upstream?.terminate(); + ws.close(code, reason); + connections.delete(ws); + if (!stopped && connections.size === 0) armIdle(); + } + + function armIdle() { + clearTimeout(idleTimer); + idleTimer = setTimeout(stop, options.idleTimeoutMs ?? CDP_PROXY_IDLE_MS); + } + + function stop() { + if (stopped) return; + stopped = true; + lifetime.abort(); + clearTimeout(idleTimer); + for (const ws of connections) close(ws, 1001, "CDP proxy stopped"); + server.stop(true); + resolveClosed(); + } + + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request, server) { + if ( + request.headers.get("origin") !== null || + request.headers.get("host") !== `127.0.0.1:${server.port}` + ) + return new Response(null, { status: 403 }); + const url = new URL(request.url); + if (url.pathname === "/health" || url.pathname === "/stop") { + if ( + !authenticated( + request.headers.get("authorization"), + `Bearer ${options.stopToken}`, + ) + ) + return new Response(null, { status: 401 }); + if (url.pathname === "/health" && request.method === "GET") + return Response.json({ id: options.id, pid: process.pid }); + if (url.pathname === "/stop" && request.method === "POST") { + setTimeout(stop, 0); + return Response.json({ stopping: true }); + } + return new Response(null, { status: 405 }); + } + if (url.pathname !== "/cdp" || request.method !== "GET") + return new Response(null, { status: 404 }); + if (!authenticated(url.searchParams.get("token"), options.token)) + return new Response(null, { status: 401 }); + if (connections.size >= 16) return new Response(null, { status: 503 }); + if ( + server.upgrade(request, { + data: { pending: [], bytes: 0, closed: false }, + }) + ) + return; + return new Response(null, { status: 426 }); + }, + websocket: { + maxPayloadLength: maxBufferedBytes, + backpressureLimit: maxBufferedBytes, + closeOnBackpressureLimit: true, + idleTimeout: 0, + open(ws) { + connections.add(ws); + clearTimeout(idleTimer); + ws.data.timer = setTimeout( + () => close(ws, 1011, "CDP connection timed out"), + options.connectTimeoutMs ?? 10_000, + ); + void options + .getToken(lifetime.signal) + .then((token) => { + if (ws.data.closed || stopped) return; + const url = new URL(options.upstreamUrl); + url.searchParams.set("token", token); + const upstream = new WebSocket(url); + ws.data.upstream = upstream; + upstream.binaryType = "arraybuffer"; + upstream.onopen = () => { + clearTimeout(ws.data.timer); + if (ws.data.closed) { + upstream.terminate(); + return; + } + for (const message of ws.data.pending) upstream.send(message); + ws.data.pending.length = 0; + ws.data.bytes = 0; + }; + upstream.onmessage = (event) => { + if (!ws.data.closed) ws.send(event.data); + }; + upstream.onclose = () => close(ws); + upstream.onerror = () => + close(ws, 1011, "CDP upstream connection failed"); + }) + .catch(() => close(ws, 1011, "CDP authentication failed")); + }, + message(ws, message) { + const size = Buffer.byteLength(message); + if (ws.data.upstream?.readyState === WebSocket.OPEN) { + if (ws.data.upstream.bufferedAmount + size > maxBufferedBytes) { + close(ws, 1009, "CDP buffer limit exceeded"); + return; + } + ws.data.upstream.send(message); + return; + } + if (ws.data.pending.length >= 64 || ws.data.bytes + size > maxBytes) { + close(ws, 1009, "CDP buffer limit exceeded"); + return; + } + ws.data.pending.push(message); + ws.data.bytes += size; + }, + close(ws) { + close(ws); + }, + }, + }); + armIdle(); + return { endpoint: server.url.origin, stop, closed }; +} diff --git a/packages/cli/src/lib/cdp-proxy/worker-main.ts b/packages/cli/src/lib/cdp-proxy/worker-main.ts new file mode 100644 index 00000000000..f3702b8cfd3 --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/worker-main.ts @@ -0,0 +1,11 @@ +import { runCdpProxyWorker } from "./worker"; + +const controller = new AbortController(); +const stop = () => controller.abort(); +process.on("SIGTERM", stop); +process.on("SIGINT", stop); +try { + await runCdpProxyWorker(controller.signal, process.argv[3] ?? ""); +} catch { + process.exitCode = 1; +} diff --git a/packages/cli/src/lib/cdp-proxy/worker.test.ts b/packages/cli/src/lib/cdp-proxy/worker.test.ts new file mode 100644 index 00000000000..7af136ac6e7 --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/worker.test.ts @@ -0,0 +1,109 @@ +import { expect, test } from "bun:test"; +import { mkdtempSync, readdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +test("stopping the detached worker aborts a stalled token exchange", async () => { + let started: () => void = () => {}; + const requested = new Promise((resolve) => { + started = resolve; + }); + const api = Bun.serve({ + port: 0, + fetch() { + started(); + return new Promise(() => {}); + }, + }); + const home = mkdtempSync(join(tmpdir(), "superset-cdp-stop-")); + const child = Bun.spawn( + [ + process.execPath, + join(import.meta.dir, "worker-main.ts"), + "--instance", + "a".repeat(64), + ], + { + env: { + ...process.env, + SUPERSET_HOME_DIR: home, + SUPERSET_API_URL: api.url.origin, + }, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }, + ); + try { + child.stdin.write( + `${JSON.stringify({ id: "b".repeat(64), apiKey: ["sk", "test", "fixture"].join("_"), upstreamUrl: "ws://127.0.0.1:1/cdp" })}\n`, + ); + const reader = child.stdout.getReader(); + const { value } = await reader.read(); + const manifest = JSON.parse(new TextDecoder().decode(value)); + child.stdin.write("ready\n"); + child.stdin.end(); + const ws = new WebSocket( + `${manifest.endpoint.replace("http", "ws")}/cdp?token=${manifest.token}`, + ); + try { + await requested; + const response = await fetch(`${manifest.endpoint}/stop`, { + method: "POST", + headers: { Authorization: `Bearer ${manifest.stopToken}` }, + }); + expect(response.status).toBe(200); + expect( + await Promise.race([ + child.exited, + Bun.sleep(500).then(() => "still running"), + ]), + ).toBe(0); + await expect(fetch(`${manifest.endpoint}/health`)).rejects.toThrow(); + } finally { + ws.close(); + } + } finally { + if (child.exitCode === null) child.kill("SIGKILL"); + await child.exited; + api.stop(true); + rmSync(home, { recursive: true, force: true }); + } +}); + +test("a worker exits without a manifest when its launcher disappears before acknowledgement", async () => { + const home = mkdtempSync(join(tmpdir(), "superset-cdp-startup-")); + const child = Bun.spawn( + [ + process.execPath, + join(import.meta.dir, "worker-main.ts"), + "--instance", + "a".repeat(64), + ], + { + env: { ...process.env, SUPERSET_HOME_DIR: home }, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }, + ); + try { + child.stdin.write( + `${JSON.stringify({ id: "b".repeat(64), apiKey: ["sk", "test", "fixture"].join("_"), upstreamUrl: "ws://127.0.0.1:1/cdp" })}\n`, + ); + const { value } = await child.stdout.getReader().read(); + const manifest = JSON.parse(new TextDecoder().decode(value)); + child.stdin.end(); + expect(await child.exited).toBe(0); + await expect(fetch(`${manifest.endpoint}/health`)).rejects.toThrow(); + expect( + readdirSync(join(home, "cdp-proxies")).filter((name) => + name.endsWith(".json"), + ), + ).toHaveLength(0); + } finally { + if (child.exitCode === null) child.kill("SIGKILL"); + await child.exited; + rmSync(home, { recursive: true, force: true }); + } +}); diff --git a/packages/cli/src/lib/cdp-proxy/worker.ts b/packages/cli/src/lib/cdp-proxy/worker.ts new file mode 100644 index 00000000000..b3571f7e504 --- /dev/null +++ b/packages/cli/src/lib/cdp-proxy/worker.ts @@ -0,0 +1,88 @@ +import { randomBytes } from "node:crypto"; +import { createInterface } from "node:readline"; +import { z } from "zod"; +import { createHostTokenProvider } from "../host-target/exchangeApiKey"; +import { + type ProxyManifest, + proxyId, + removeProxy, + tryProxyLock, +} from "./registry"; +import { startCdpProxy } from "./server"; + +const configuration = z.object({ + id: proxyId, + apiKey: z.string().regex(/^sk_(live|test)_/), + upstreamUrl: z.url().refine((value) => { + const url = new URL(value); + return ( + (url.protocol === "ws:" || url.protocol === "wss:") && + !url.username && + !url.password && + !url.searchParams.has("token") + ); + }), +}); + +export async function runCdpProxyWorker( + signal: AbortSignal, + instance: string, +): Promise { + proxyId.parse(instance); + const lines = createInterface({ input: process.stdin })[ + Symbol.asyncIterator + ](); + const startup = setTimeout(() => { + process.stdin.destroy(); + }, 10_000); + let proxy: ReturnType | undefined; + let manifest: ProxyManifest | undefined; + const stop = () => { + proxy?.stop(); + process.stdin.destroy(); + }; + signal.addEventListener("abort", stop, { once: true }); + try { + const first = await lines.next(); + if (first.done || first.value.length > 16_384) + throw new Error("Invalid CDP proxy configuration"); + const config = configuration.parse(JSON.parse(first.value)); + const token = randomBytes(32).toString("hex"); + const stopToken = randomBytes(32).toString("hex"); + proxy = startCdpProxy({ + ...config, + token, + stopToken, + getToken: createHostTokenProvider(config.apiKey), + }); + manifest = { + id: config.id, + pid: process.pid, + instance, + endpoint: proxy.endpoint, + token, + stopToken, + }; + if (signal.aborted) stop(); + process.stdout.write(`${JSON.stringify(manifest)}\n`); + const acknowledgement = await lines.next(); + clearTimeout(startup); + if (acknowledgement.done || acknowledgement.value !== "ready") stop(); + await proxy.closed; + } finally { + clearTimeout(startup); + proxy?.stop(); + signal.removeEventListener("abort", stop); + if (manifest) { + const release = tryProxyLock(manifest.id); + if (release) { + try { + removeProxy(manifest); + } finally { + release(); + } + } + } + await lines.return?.(); + } +} diff --git a/packages/cli/src/lib/host-target/exchangeApiKey.ts b/packages/cli/src/lib/host-target/exchangeApiKey.ts index 6fd3d6302e9..a9b93e3a19a 100644 --- a/packages/cli/src/lib/host-target/exchangeApiKey.ts +++ b/packages/cli/src/lib/host-target/exchangeApiKey.ts @@ -8,13 +8,16 @@ function isApiKey(bearer: string): boolean { return bearer.startsWith("sk_live_") || bearer.startsWith("sk_test_"); } -export function createHostTokenProvider(bearer: string): () => Promise { +export function createHostTokenProvider( + bearer: string, + now: () => number = Date.now, +): (signal?: AbortSignal) => Promise { let cached: { token: string; expiresAt: number } | undefined; let pending: Promise | undefined; - return async () => { + return async (signal) => { if (!isApiKey(bearer)) return bearer; - if (cached && Date.now() < cached.expiresAt - 30_000) return cached.token; - pending ??= exchangeApiKey(bearer) + if (cached && now() < cached.expiresAt - 30_000) return cached.token; + pending ??= exchangeApiKey(bearer, signal) .then((token) => { let expiresAt = 0; try { @@ -36,14 +39,19 @@ export function createHostTokenProvider(bearer: string): () => Promise { }; } -export async function exchangeApiKey(bearer: string): Promise { +export async function exchangeApiKey( + bearer: string, + signal?: AbortSignal, +): Promise { if (!isApiKey(bearer)) return bearer; let response: Response; try { response = await fetch(`${env.SUPERSET_API_URL}/api/auth/token`, { headers: { "x-api-key": bearer }, - signal: AbortSignal.timeout(10_000), + signal: signal + ? AbortSignal.any([signal, AbortSignal.timeout(10_000)]) + : AbortSignal.timeout(10_000), redirect: "error", }); } catch { From 710d9ad8be483242981a431df3565b4c8e022369 Mon Sep 17 00:00:00 2001 From: Krishna Date: Mon, 5 Oct 2026 10:35:42 +0530 Subject: [PATCH 4/4] fix(cli): recover stale CDP proxy registrations --- .../src/commands/browser/cdp/command.test.ts | 7 + .../cli/src/commands/browser/cdp/command.ts | 3 + packages/cli/src/lib/cdp-proxy/client.ts | 60 +++++-- .../cli/src/lib/cdp-proxy/registry.test.ts | 159 +++++++++++++++++- packages/cli/src/lib/cdp-proxy/registry.ts | 47 +++++- 5 files changed, 256 insertions(+), 20 deletions(-) diff --git a/packages/cli/src/commands/browser/cdp/command.test.ts b/packages/cli/src/commands/browser/cdp/command.test.ts index 8b451e716cf..67b70f3d3f7 100644 --- a/packages/cli/src/commands/browser/cdp/command.test.ts +++ b/packages/cli/src/commands/browser/cdp/command.test.ts @@ -153,6 +153,13 @@ test.each( expect(await stop.exited).toBe(0); await expect(fetch(`${manifest.endpoint}/health`)).rejects.toThrow(); }); + const warning = await new Response(child.stderr).text(); + if (launcher.direct) expect(parsed.message).toBe(result.data.url); + expect(warning).toContain("Treat it as a secret"); + expect(warning).toContain(`superset browser cdp-stop --id ${manifest.id}`); + expect(warning).not.toContain(manifest.token); + expect(warning).not.toContain(result.data.url); + expect(warning).not.toContain(["sk", "test", "cdp", "fixture"].join("_")); expect(statSync(manifestPath).mode & 0o777).toBe(0o600); expect(readFileSync(manifestPath, "utf8")).not.toContain( ["sk", "test", "cdp", "fixture"].join("_"), diff --git a/packages/cli/src/commands/browser/cdp/command.ts b/packages/cli/src/commands/browser/cdp/command.ts index 72f13be4257..4e01c9f4eda 100644 --- a/packages/cli/src/commands/browser/cdp/command.ts +++ b/packages/cli/src/commands/browser/cdp/command.ts @@ -42,6 +42,9 @@ export default command({ paneId: options.pane, upstreamUrl: endpoint, }); + process.stderr.write( + `This local URL grants CDP control of the pane. Treat it as a secret.\nStop the proxy with: superset browser cdp-stop --id ${proxy.proxyId}\n`, + ); return { data: proxy, message: proxy.url, diff --git a/packages/cli/src/lib/cdp-proxy/client.ts b/packages/cli/src/lib/cdp-proxy/client.ts index 4444284e7ee..b236970371a 100644 --- a/packages/cli/src/lib/cdp-proxy/client.ts +++ b/packages/cli/src/lib/cdp-proxy/client.ts @@ -4,11 +4,13 @@ import { fileURLToPath } from "node:url"; import { CLIError } from "@superset/cli-framework"; import { env } from "../env"; import { + connectionWasRefused, type ProxyManifest, proxyDirectory, - proxyIsLive, + proxyHealth, + proxyIsGone, proxyManifest, - proxyProcessIsAlive, + proxyProcessState, readProxy, removeProxy, tryProxyLock, @@ -162,8 +164,13 @@ export async function ensureCdpProxy(options: CdpProxyOptions) { const release = await lockProxy(id); try { const existing = readProxy(id); - const live = existing && (await proxyIsLive(existing)); - if (existing && !live && (await proxyProcessIsAlive(existing))) + const health = existing && (await proxyHealth(existing)); + const live = health === "live"; + if ( + existing && + !live && + !(await proxyIsGone(existing, health ?? undefined)) + ) throw new CLIError( "The existing CDP proxy is not responding", `Retry or stop it with: superset browser cdp-stop --id ${id}`, @@ -185,17 +192,44 @@ export async function stopCdpProxy(id: string): Promise { try { const manifest = readProxy(id); if (!manifest) return false; - if (await proxyProcessIsAlive(manifest)) { - const response = await fetch(`${manifest.endpoint}/stop`, { - method: "POST", - headers: { Authorization: `Bearer ${manifest.stopToken}` }, - signal: AbortSignal.timeout(1000), - redirect: "error", - }); - if (!response.ok) + const processState = await proxyProcessState(manifest); + if (processState === "unknown") { + const health = await proxyHealth(manifest); + if (health === "refused") { + removeProxy(manifest); + return true; + } + if (health !== "live") + throw new CLIError( + "Could not verify the local CDP proxy", + "The proxy may still be running. Retry the stop command", + ); + } + if (processState !== "gone") { + let response: Response | undefined; + try { + response = await fetch(`${manifest.endpoint}/stop`, { + method: "POST", + headers: { Authorization: `Bearer ${manifest.stopToken}` }, + signal: AbortSignal.timeout(1000), + redirect: "error", + }); + } catch (error) { + if ( + !(await proxyIsGone( + manifest, + connectionWasRefused(error) ? "refused" : "unknown", + )) + ) + throw new CLIError( + "Could not stop the local CDP proxy", + "The proxy may still be running. Retry the stop command", + ); + } + if (response && !response.ok) throw new CLIError("Could not stop the local CDP proxy"); const deadline = Date.now() + 2000; - while (await proxyProcessIsAlive(manifest)) { + while (response && !(await proxyIsGone(manifest))) { if (Date.now() >= deadline) throw new CLIError( "The CDP proxy has not stopped yet", diff --git a/packages/cli/src/lib/cdp-proxy/registry.test.ts b/packages/cli/src/lib/cdp-proxy/registry.test.ts index 93ae13d0c3d..10ab1a6540a 100644 --- a/packages/cli/src/lib/cdp-proxy/registry.test.ts +++ b/packages/cli/src/lib/cdp-proxy/registry.test.ts @@ -1,13 +1,16 @@ import { afterAll, expect, test } from "bun:test"; import { + mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync, } from "node:fs"; +import { createServer } from "node:net"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { CLIError } from "@superset/cli-framework"; const home = mkdtempSync(join(tmpdir(), "superset-cdp-registry-")); const previousHome = process.env.SUPERSET_HOME_DIR; @@ -21,6 +24,7 @@ const { } = await import("./registry"); const { cdpProxyId, ensureCdpProxy, stopCdpProxy } = await import("./client"); const { env } = await import("../env"); +mkdirSync(proxyDirectory, { recursive: true, mode: 0o700 }); const options = { apiKey: ["sk", "test", "registry", "fixture"].join("_"), organizationId: "org", @@ -31,8 +35,147 @@ const options = { }; afterAll(() => { - process.env.SUPERSET_HOME_DIR = previousHome; + if (previousHome === undefined) delete process.env.SUPERSET_HOME_DIR; + else process.env.SUPERSET_HOME_DIR = previousHome; rmSync(home, { recursive: true, force: true }); + expect(process.env.SUPERSET_HOME_DIR).toBe(previousHome); +}); + +test.each([ + "ensure", + "stop", +])("%s recovers a refused endpoint with unknown recycled-PID identity", async (action) => { + const endpoint = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => new Response(), + }); + const manifest = { + id: cdpProxyId(options), + pid: process.pid, + instance: "d".repeat(64), + endpoint: endpoint.url.origin, + token: "b".repeat(64), + stopToken: "c".repeat(64), + }; + endpoint.stop(true); + writeProxy(manifest); + const previousPath = process.env.PATH; + process.env.PATH = home; + try { + if (action === "ensure") { + const proxy = await ensureCdpProxy(options); + expect(readProxy(proxy.proxyId)?.pid).not.toBe(process.pid); + } else { + expect(await stopCdpProxy(manifest.id)).toBe(true); + expect(readProxy(manifest.id)).toBeNull(); + } + expect(() => process.kill(process.pid, 0)).not.toThrow(); + } finally { + if (previousPath === undefined) delete process.env.PATH; + else process.env.PATH = previousPath; + await stopCdpProxy(manifest.id); + } +}); + +test("uncertain transport failures keep the manifest and produce a controlled stop error", async () => { + const endpoint = createServer((socket) => socket.destroy()); + await new Promise((resolve) => + endpoint.listen(0, "127.0.0.1", resolve), + ); + const address = endpoint.address(); + if (!address || typeof address === "string") + throw new Error("Missing TCP address"); + const manifest = { + id: cdpProxyId(options), + pid: process.pid, + instance: "d".repeat(64), + endpoint: `http://127.0.0.1:${address.port}`, + token: "b".repeat(64), + stopToken: "c".repeat(64), + }; + writeProxy(manifest); + const previousPath = process.env.PATH; + process.env.PATH = home; + try { + await expect(ensureCdpProxy(options)).rejects.toBeInstanceOf(CLIError); + await expect(stopCdpProxy(manifest.id)).rejects.toBeInstanceOf(CLIError); + expect(readProxy(manifest.id)).toEqual(manifest); + } finally { + if (previousPath === undefined) delete process.env.PATH; + else process.env.PATH = previousPath; + endpoint.close(); + await stopCdpProxy(manifest.id); + } +}); + +test("an authenticated stop can confirm closure when PID inspection is unavailable", async () => { + const endpoint = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + expect(request.headers.get("Authorization")).toBe( + `Bearer ${"c".repeat(64)}`, + ); + if (new URL(request.url).pathname === "/health") + return Response.json({ id: cdpProxyId(options), pid: process.pid }); + setTimeout(() => endpoint.stop(true), 0); + return Response.json({ stopping: true }); + }, + }); + const manifest = { + id: cdpProxyId(options), + pid: process.pid, + instance: "d".repeat(64), + endpoint: endpoint.url.origin, + token: "b".repeat(64), + stopToken: "c".repeat(64), + }; + writeProxy(manifest); + const previousPath = process.env.PATH; + process.env.PATH = home; + try { + expect(await stopCdpProxy(manifest.id)).toBe(true); + expect(readProxy(manifest.id)).toBeNull(); + } finally { + if (previousPath === undefined) delete process.env.PATH; + else process.env.PATH = previousPath; + endpoint.stop(true); + await stopCdpProxy(manifest.id); + } +}); + +test("unknown ownership never sends a stop request to a foreign local server", async () => { + let stopRequests = 0; + const endpoint = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + if (new URL(request.url).pathname === "/stop") stopRequests++; + return Response.json({ id: "foreign", pid: process.pid }); + }, + }); + const manifest = { + id: cdpProxyId(options), + pid: process.pid, + instance: "d".repeat(64), + endpoint: endpoint.url.origin, + token: "b".repeat(64), + stopToken: "c".repeat(64), + }; + writeProxy(manifest); + const previousPath = process.env.PATH; + process.env.PATH = home; + try { + await expect(stopCdpProxy(manifest.id)).rejects.toBeInstanceOf(CLIError); + expect(stopRequests).toBe(0); + expect(readProxy(manifest.id)).toEqual(manifest); + } finally { + if (previousPath === undefined) delete process.env.PATH; + else process.env.PATH = previousPath; + endpoint.stop(true); + await stopCdpProxy(manifest.id); + } }); test("a crashed lock owner releases the startup transaction", async () => { @@ -104,6 +247,20 @@ test("concurrent exports reuse one daemon and uncertain health retains its manag await expect(ensureCdpProxy(options)).rejects.toThrow("not responding"); await expect(stopCdpProxy(manifest.id)).rejects.toThrow("Could not stop"); expect(readProxy(manifest.id)?.pid).toBe(manifest.pid); + const closedEndpoint = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch: () => new Response(), + }); + const refusedManifest = { + ...manifest, + endpoint: closedEndpoint.url.origin, + }; + closedEndpoint.stop(true); + writeProxy(refusedManifest); + await expect(ensureCdpProxy(options)).rejects.toThrow("not responding"); + await expect(stopCdpProxy(manifest.id)).rejects.toBeInstanceOf(CLIError); + expect(readProxy(manifest.id)).toEqual(refusedManifest); writeProxy(manifest); expect( readFileSync(join(proxyDirectory, `${manifest.id}.json`), "utf8"), diff --git a/packages/cli/src/lib/cdp-proxy/registry.ts b/packages/cli/src/lib/cdp-proxy/registry.ts index 89ce33202f0..863f327a4c8 100644 --- a/packages/cli/src/lib/cdp-proxy/registry.ts +++ b/packages/cli/src/lib/cdp-proxy/registry.ts @@ -93,10 +93,27 @@ export function writeProxy(manifest: ProxyManifest) { export async function proxyProcessIsAlive( manifest: ProxyManifest, ): Promise { - if (!isProcessAlive(manifest.pid)) return false; + return (await proxyProcessState(manifest)) !== "gone"; +} + +export async function proxyProcessState( + manifest: ProxyManifest, +): Promise<"owned" | "gone" | "unknown"> { + if (!isProcessAlive(manifest.pid)) return "gone"; const command = await inspectProcessCommand(manifest.pid); + if (command === null) return "unknown"; + return command.includes(`--instance ${manifest.instance}`) ? "owned" : "gone"; +} + +export async function proxyIsGone( + manifest: ProxyManifest, + health?: ProxyHealth, +): Promise { + const processState = await proxyProcessState(manifest); + if (processState === "gone") return true; return ( - command === null || command.includes(`--instance ${manifest.instance}`) + processState === "unknown" && + (health ?? (await proxyHealth(manifest))) === "refused" ); } @@ -106,7 +123,23 @@ export function removeProxy(manifest: ProxyManifest) { unlinkSync(join(proxyDirectory, `${manifest.id}.json`)); } -export async function proxyIsLive(manifest: ProxyManifest): Promise { +type ProxyHealth = "live" | "refused" | "unknown"; + +export function connectionWasRefused(error: unknown): boolean { + if (typeof error !== "object" || error === null) return false; + const code = "code" in error ? error.code : undefined; + return ( + code === "ConnectionRefused" || + code === "ECONNREFUSED" || + ("cause" in error && + error.cause !== error && + connectionWasRefused(error.cause)) + ); +} + +export async function proxyHealth( + manifest: ProxyManifest, +): Promise { try { const response = await fetch(`${manifest.endpoint}/health`, { headers: { Authorization: `Bearer ${manifest.stopToken}` }, @@ -114,8 +147,10 @@ export async function proxyIsLive(manifest: ProxyManifest): Promise { redirect: "error", }); const body = (await response.json()) as { id?: string; pid?: number }; - return response.ok && body.id === manifest.id && body.pid === manifest.pid; - } catch { - return false; + return response.ok && body.id === manifest.id && body.pid === manifest.pid + ? "live" + : "unknown"; + } catch (error) { + return connectionWasRefused(error) ? "refused" : "unknown"; } }