diff --git a/.changeset/calm-egress-policies.md b/.changeset/calm-egress-policies.md new file mode 100644 index 00000000..139daa0d --- /dev/null +++ b/.changeset/calm-egress-policies.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +Configure ambient network access consistently across execution backends. diff --git a/examples/container/src/index.ts b/examples/container/src/index.ts index bbbadf1a..28510bd1 100644 --- a/examples/container/src/index.ts +++ b/examples/container/src/index.ts @@ -52,6 +52,7 @@ class ContainerBase extends withWorkspaceContainer(class extends DurableObject this, workspace: { binding: "ContainerExample", id: this.ctx.id.toString() }, + egress: { mode: "direct" }, }); } diff --git a/examples/think-compare-runtimes/worker/think/agents.ts b/examples/think-compare-runtimes/worker/think/agents.ts index ef980d89..04425f0f 100644 --- a/examples/think-compare-runtimes/worker/think/agents.ts +++ b/examples/think-compare-runtimes/worker/think/agents.ts @@ -330,6 +330,7 @@ export class WorkspaceThinkAgent extends RuntimeThinkAgent { ); }, workspace: workspaceRef, + egress: { mode: "direct" }, containerEnv: this.env.FUSE_MOUNT ? { FUSE_MOUNT: this.env.FUSE_MOUNT } : undefined, }); const workspace = new Workspace({ diff --git a/examples/think/src/agent.ts b/examples/think/src/agent.ts index e9ae6122..7a82b725 100644 --- a/examples/think/src/agent.ts +++ b/examples/think/src/agent.ts @@ -80,6 +80,7 @@ export class Assistant extends withWorkspaceContainer(AssistantBase) { id: "container", container: () => this, workspace: workspaceRef(this.ctx), + egress: { mode: "direct" }, }); /** diff --git a/examples/tutorial/src/index.ts b/examples/tutorial/src/index.ts index 6e86bf2c..9a0aae33 100644 --- a/examples/tutorial/src/index.ts +++ b/examples/tutorial/src/index.ts @@ -48,6 +48,7 @@ export class RecipeAgent extends withWorkspaceContainer(RecipeBase) { readonly #backend = new CloudflareContainerBackend({ container: () => this, workspace: { binding: "RecipeAgent", id: this.ctx.id.toString() }, + egress: { mode: "direct" }, }); override workspace = new Workspace({ diff --git a/packages/computer/src/backends/container/cloudflare-container.test.ts b/packages/computer/src/backends/container/cloudflare-container.test.ts index 70c86b95..08de8265 100644 --- a/packages/computer/src/backends/container/cloudflare-container.test.ts +++ b/packages/computer/src/backends/container/cloudflare-container.test.ts @@ -32,8 +32,11 @@ interface FakeHost { host: IWorkspaceContainerAPI; calls: { name: string; args: unknown[] }[]; startEnv?: Record; + enableInternet?: boolean; interceptedHost?: string; interceptedWorkspace?: WorkspaceRef; + gatewayWorkspace?: WorkspaceRef; + gatewayToken?: string; running: boolean; exit: { exitedAt: number; reason: string } | null; simulateExit(reason: string): void; @@ -62,9 +65,10 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { } state.host = { - async start(env) { - calls.push({ name: "start", args: [env] }); + async start(env, enableInternet) { + calls.push({ name: "start", args: [env, enableInternet] }); state.startEnv = env; + state.enableInternet = enableInternet; state.running = true; // A successful start clears any prior exit, matching // WorkspaceContainerAPI.start. @@ -75,6 +79,11 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { state.interceptedHost = host; state.interceptedWorkspace = ref; }, + async interceptAllOutboundHttp(ref, token) { + calls.push({ name: "interceptAllOutboundHttp", args: [ref, token] }); + state.gatewayWorkspace = ref; + state.gatewayToken = token; + }, async fetchPort(port, input, init) { const request = input instanceof Request ? input : new Request(input, init); const url = new URL(request.url); @@ -94,8 +103,8 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { port() { throw new Error("cross-boundary Fetchers should not be used by CloudflareContainerBackend"); }, - async restart(env) { - calls.push({ name: "restart", args: [env] }); + async restart(env, enableInternet) { + calls.push({ name: "restart", args: [env, enableInternet] }); if (opts.restart) { await opts.restart(); } @@ -135,6 +144,79 @@ describe("CloudflareContainerBackend", () => { expect(fake.interceptedWorkspace).toEqual(fakeWorkspace); }); + test("blocks ambient egress by default", async () => { + const fake = makeFakeHost({ healthy: false }); + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 300, + }); + + await expect(backend.connect()).rejects.toThrow(); + + expect(fake.enableInternet).toBe(false); + }); + + test("enables direct ambient egress", async () => { + const fake = makeFakeHost({ healthy: false }); + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 300, + egress: { mode: "direct" }, + }); + + await expect(backend.connect()).rejects.toThrow(); + + expect(fake.enableInternet).toBe(true); + }); + + test("routes HTTP egress through the configured gateway", async () => { + const fake = makeFakeHost({ healthy: false }); + const gateway = { + fetch: vi.fn(async (request: Request) => new Response(request.url)), + } as unknown as Fetcher; + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 300, + egress: { mode: "http-gateway", gateway }, + }); + await expect(backend.connect()).rejects.toThrow(); + const request = new Request("https://api.example.test/data", { + headers: { "x-workspace-egress-token": fake.gatewayToken ?? "" }, + }); + + const response = await backend.handleFetch(request); + + expect(fake.gatewayWorkspace).toEqual(fakeWorkspace); + expect(await response.text()).toBe("https://api.example.test/data"); + expect(gateway.fetch).toHaveBeenCalledOnce(); + }); + + test("rejects container egress callbacks with the wrong token", async () => { + const fake = makeFakeHost({ healthy: false }); + const gateway = { + fetch: vi.fn(async () => new Response("forwarded")), + } as unknown as Fetcher; + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 300, + egress: { mode: "http-gateway", gateway }, + }); + await expect(backend.connect()).rejects.toThrow(); + + const response = await backend.handleFetch( + new Request("https://api.example.test/data", { + headers: { "x-workspace-egress-token": "wrong" }, + }), + ); + + expect(response.status).toBe(404); + expect(gateway.fetch).not.toHaveBeenCalled(); + }); + test("egressHost option overrides the default", async () => { const fake = makeFakeHost({ healthy: false }); const backend = new CloudflareContainerBackend({ diff --git a/packages/computer/src/backends/container/cloudflare-container.ts b/packages/computer/src/backends/container/cloudflare-container.ts index aa43a62d..da84cb8e 100644 --- a/packages/computer/src/backends/container/cloudflare-container.ts +++ b/packages/computer/src/backends/container/cloudflare-container.ts @@ -49,6 +49,7 @@ import { newWebSocketRpcSession, type RpcStub } from "capnweb"; import type { BackendHandle, WorkspaceBackend } from "../../backend.js"; import { startHeartbeat } from "../../heartbeat.js"; +import { WORKSPACE_EGRESS_TOKEN_HEADER, type WorkspaceEgressPolicy } from "../../runtime/egress.js"; import type { IWorkspaceContainerAPI, WorkspaceRef } from "./container-host.js"; import { probeComputerdHealth } from "./health-probe.js"; @@ -83,6 +84,8 @@ export interface CloudflareContainerBackendOptions { // sharing the same container host. egressHost?: string; + egress?: WorkspaceEgressPolicy; + // TCP port computerd listens on inside the container. Default 8080, // matching the Dockerfile shipped with examples/container. containerPort?: number; @@ -143,9 +146,14 @@ export class CloudflareContainerBackend implements WorkspaceBackend { readonly id: string; readonly #options: Required< - Omit + Omit< + CloudflareContainerBackendOptions, + "container" | "workspace" | "containerEnv" | "egress" | "id" + > > & Pick; + readonly #egress: WorkspaceEgressPolicy; + readonly #egressToken: string | undefined; // State for the in-flight /ws upgrade. handleFetch() resolves // #pendingUpgrade; connect() awaits it. @@ -159,6 +167,8 @@ export class CloudflareContainerBackend implements WorkspaceBackend { constructor(options: CloudflareContainerBackendOptions) { this.id = options.id ?? "container-shell"; + this.#egress = options.egress ?? { mode: "none" }; + this.#egressToken = this.#egress.mode === "http-gateway" ? crypto.randomUUID() : undefined; this.#options = { container: options.container, workspace: options.workspace, @@ -195,8 +205,11 @@ export class CloudflareContainerBackend implements WorkspaceBackend { MOUNT_POINT: "/workspace", ...this.#options.containerEnv, }; - await host.start(env); + await host.start(env, this.#egress.mode === "direct"); await host.interceptOutboundHttp(this.#options.egressHost, this.#options.workspace); + if (this.#egress.mode === "http-gateway" && this.#egressToken !== undefined) { + await host.interceptAllOutboundHttp(this.#options.workspace, this.#egressToken); + } // Arm the upgrade promise before posting /connect — computerd // dials back as soon as /health on the egress answers, so @@ -286,6 +299,15 @@ export class CloudflareContainerBackend implements WorkspaceBackend { // Returns the 101 response that the WorkspaceProxy fetch handler // forwards back to the container. async handleFetch(req: Request): Promise { + if ( + this.#egress.mode === "http-gateway" && + this.#egressToken !== undefined && + req.headers.get(WORKSPACE_EGRESS_TOKEN_HEADER) === this.#egressToken + ) { + const headers = new Headers(req.headers); + headers.delete(WORKSPACE_EGRESS_TOKEN_HEADER); + return this.#egress.gateway.fetch(new Request(req, { headers })); + } const url = new URL(req.url); if (url.pathname !== "/ws") { return new Response("not found", { status: 404 }); @@ -365,7 +387,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { if (attempt < maxAttempts) { try { - await host.restart(env); + await host.restart(env, this.#egress.mode === "direct"); restarts++; } catch (error) { this.#rejectUpgrade?.(error); diff --git a/packages/computer/src/backends/container/container-host.ts b/packages/computer/src/backends/container/container-host.ts index aabcd354..0cc2f519 100644 --- a/packages/computer/src/backends/container/container-host.ts +++ b/packages/computer/src/backends/container/container-host.ts @@ -46,13 +46,14 @@ export interface IWorkspaceContainerAPI { // Idempotent start. Returns once the runtime has accepted the // start command; readiness is verified by the backend through // probeComputerdHealth against port(). - start(env: Record): Promise; + start(env: Record, enableInternet: boolean): Promise; // Wire `host` → workspace inside the container's egress table. // Called once per backend connect(). The implementation // constructs the loopback Fetcher locally from {binding, id}, // because Fetchers can't survive a Workers RPC hop. interceptOutboundHttp(host: string, workspace: WorkspaceRef): Promise; + interceptAllOutboundHttp(workspace: WorkspaceRef, token: string): Promise; // Fetch against a named TCP port inside the container. The fetch // runs in the container-owning Durable Object, so callers across @@ -68,7 +69,7 @@ export interface IWorkspaceContainerAPI { // current generation dead. Implementation: destroy() the // container, then start({ env }). Callers bound the number of // restart attempts — this method does no looping of its own. - restart(env: Record): Promise; + restart(env: Record, enableInternet: boolean): Promise; // Coarse diagnostic state. The `running` flag reports whether // the platform still has a container instance attached; it does @@ -104,7 +105,7 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai this.#ctx = ctx; } - async start(env: Record) { + async start(env: Record, enableInternet: boolean) { // If a prior generation has died, commit to a fresh one: the // destroy clears any platform-side carcass, and the start that // follows is unconditional. We cannot rely on @@ -119,14 +120,14 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai // best-effort — the next start() will surface any real // platform-side failure. } - this.#container.start({ enableInternet: true, env }); + this.#container.start({ enableInternet, env }); } else if (!this.#container.running) { - this.#container.start({ enableInternet: true, env }); + this.#container.start({ enableInternet, env }); } installContainerMonitor(this.#ctx, this.#container); } - async restart(env: Record) { + async restart(env: Record, enableInternet: boolean) { // destroy() resolves once the platform has torn down the // attached container. A subsequent start() launches a fresh // generation — ports re-bind, the computerd daemon comes up clean. @@ -139,7 +140,7 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai // succeed against a fresh generation or surface its own // failure. } - this.#container.start({ enableInternet: true, env }); + this.#container.start({ enableInternet, env }); installContainerMonitor(this.#ctx, this.#container); } @@ -151,6 +152,17 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai return containerExitInfo(this.#ctx); } + async interceptAllOutboundHttp(ref: WorkspaceRef, token: string) { + const exports = (this.#ctx as unknown as { exports: Record }).exports as { + WorkspaceProxy: (opts: { props: WorkspaceRef & { egressToken: string } }) => Fetcher; + }; + const proxy = exports.WorkspaceProxy({ props: { ...ref, egressToken: token } }); + await Promise.all([ + this.#container.interceptAllOutboundHttp(proxy), + this.#container.interceptOutboundHttps("*", proxy), + ]); + } + async interceptOutboundHttp(host: string, ref: WorkspaceRef) { // ctx.exports.WorkspaceProxy is bound by name in the // consumer's Worker (they re-export WorkspaceProxy from this diff --git a/packages/computer/src/backends/container/index.ts b/packages/computer/src/backends/container/index.ts index df9b0055..2aa20d7a 100644 --- a/packages/computer/src/backends/container/index.ts +++ b/packages/computer/src/backends/container/index.ts @@ -12,6 +12,7 @@ // withWorkspaceContainer, // } from "@cloudflare/computer/backends/container"; +export type { WorkspaceEgressPolicy } from "../../runtime/egress.js"; export { CloudflareContainerBackend, type CloudflareContainerBackendOptions, diff --git a/packages/computer/src/backends/worker-javascript/index.ts b/packages/computer/src/backends/worker-javascript/index.ts index a3729142..41a4e81f 100644 --- a/packages/computer/src/backends/worker-javascript/index.ts +++ b/packages/computer/src/backends/worker-javascript/index.ts @@ -1,3 +1,4 @@ +export type { WorkspaceEgressPolicy } from "../../runtime/egress.js"; export { WorkerJavaScriptBackend, type WorkerJavaScriptBackendOptions, diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts index 0439478e..46c22076 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts @@ -43,6 +43,103 @@ async function evaluateResult( } describe("WorkerJavaScriptBackend", () => { + it("blocks ambient egress by default", async () => { + const load = vi.fn(() => ({ + getEntrypoint() { + return { + evaluate: ( + _input: unknown, + host: { + assertResult(value: unknown): Promise; + attachOutput(readable: ReadableStream): Promise; + }, + ) => evaluateResult(host, null), + }; + }, + })); + const workspace = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [new WorkerJavaScriptBackend({ loader: { load } })], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + + await (await workspace.runtime.exec("export default null")).result(); + + expect(load.mock.calls[0]?.[0]).toMatchObject({ globalOutbound: null }); + }); + + it("omits globalOutbound for direct egress", async () => { + const load = vi.fn(() => ({ + getEntrypoint() { + return { + evaluate: ( + _input: unknown, + host: { + assertResult(value: unknown): Promise; + attachOutput(readable: ReadableStream): Promise; + }, + ) => evaluateResult(host, null), + }; + }, + })); + const workspace = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [ + new WorkerJavaScriptBackend({ + loader: { load }, + egress: { mode: "direct" }, + }), + ], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + + await (await workspace.runtime.exec("export default null")).result(); + + expect(load.mock.calls[0]?.[0]).not.toHaveProperty("globalOutbound"); + }); + + it("routes ambient egress through an HTTP gateway", async () => { + const gateway = { fetch: vi.fn() } as unknown as Fetcher; + const load = vi.fn(() => ({ + getEntrypoint() { + return { + evaluate: ( + _input: unknown, + host: { + assertResult(value: unknown): Promise; + attachOutput(readable: ReadableStream): Promise; + }, + ) => evaluateResult(host, null), + }; + }, + })); + const workspace = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [ + new WorkerJavaScriptBackend({ + loader: { load }, + egress: { mode: "http-gateway", gateway }, + }), + ], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + + await (await workspace.runtime.exec("export default null")).result(); + + expect(load.mock.calls[0]?.[0]).toMatchObject({ globalOutbound: gateway }); + }); + + it("rejects globalOutbound together with egress", () => { + expect( + () => + new WorkerJavaScriptBackend({ + loader: throwingLoader("unused"), + globalOutbound: null, + egress: { mode: "none" }, + }), + ).toThrow(/globalOutbound.*egress/); + }); + it("validates timeout configuration", () => { expect( () => diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index 9023c2fe..4bd5fcfe 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -1,5 +1,6 @@ import { WorkspaceRuntimeBridge } from "../../runtime/bridge.js"; import { assertRuntimeValue, WorkspaceRuntimeCapability } from "../../runtime/capability.js"; +import { dynamicWorkerEgress, type WorkspaceEgressPolicy } from "../../runtime/egress.js"; import type { ModuleExecutionEnvelope, ModuleExecutionInput, @@ -53,6 +54,7 @@ export interface WorkerJavaScriptBackendOptions { maxRetainedExecutions?: number; compatibilityDate?: string; compatibilityFlags?: string[]; + egress?: WorkspaceEgressPolicy; globalOutbound?: Fetcher | null; /** Allow ws:git operations that can perform host-side network requests. */ allowGitNetwork?: boolean; @@ -88,7 +90,9 @@ type ResolvedWorkerJavaScriptBackendOptions = Required< | "compatibilityFlags" > > & - WorkerJavaScriptBackendOptions; + Omit & { + egress: WorkspaceEgressPolicy; + }; interface WorkspaceExecutionContext { env: Record; @@ -140,6 +144,9 @@ export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { constructor(options: WorkerJavaScriptBackendOptions) { this.id = options.id ?? "worker-javascript"; + if (options.egress !== undefined && options.globalOutbound !== undefined) { + throw new Error("WorkerJavaScriptBackend cannot use globalOutbound together with egress."); + } const maxTimeoutMs = options.maxTimeoutMs ?? 180_000; const defaultTimeoutMs = options.defaultTimeoutMs ?? Math.min(60_000, maxTimeoutMs); assertPositiveFinite(maxTimeoutMs, "maxTimeoutMs"); @@ -180,8 +187,17 @@ export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { if (defaultTimeoutMs > maxTimeoutMs) { throw new Error("WorkerJavaScriptBackend defaultTimeoutMs cannot exceed maxTimeoutMs."); } + const { globalOutbound, egress, ...backendOptions } = options; + const resolvedEgress = + egress ?? + (globalOutbound === undefined + ? { mode: "none" as const } + : globalOutbound === null + ? { mode: "none" as const } + : { mode: "http-gateway" as const, gateway: globalOutbound }); this.#options = { - ...options, + ...backendOptions, + egress: resolvedEgress, root: options.root ?? "/workspace", access: options.access ?? "read-write", defaultTimeoutMs, @@ -205,7 +221,6 @@ export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { maxRetainedExecutions: options.maxRetainedExecutions ?? 100, compatibilityDate, compatibilityFlags: options.compatibilityFlags ?? ["nodejs_compat"], - globalOutbound: options.globalOutbound ?? null, }; } @@ -401,7 +416,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { }, bridge, timeoutMs, - globalOutbound: this.#options.globalOutbound ?? null, + egress: this.#options.egress, compatibilityDate: this.#options.compatibilityDate, compatibilityFlags: this.#options.compatibilityFlags, maxStdioBytes: this.#options.maxStdioBytes, @@ -916,7 +931,7 @@ function startJavaScriptExecution(options: { context: WorkspaceExecutionContext; bridge: WorkspaceRuntimeBridge; timeoutMs: number; - globalOutbound: Fetcher | null; + egress: WorkspaceEgressPolicy; compatibilityDate: string; compatibilityFlags: string[]; maxStdioBytes: number; @@ -935,7 +950,7 @@ function startJavaScriptExecution(options: { limits: { cpuMs: options.timeoutMs }, mainModule: "workspace-runtime-runner.js", modules, - globalOutbound: options.globalOutbound, + ...dynamicWorkerEgress(options.egress), }); let entrypoint: JavaScriptEntrypoint; try { diff --git a/packages/computer/src/backends/worker-shell/index.ts b/packages/computer/src/backends/worker-shell/index.ts index c4cbf561..c9db0551 100644 --- a/packages/computer/src/backends/worker-shell/index.ts +++ b/packages/computer/src/backends/worker-shell/index.ts @@ -26,6 +26,7 @@ // factory to WorkerShellBackend instead of `loader` + `workspace` + // `ctx`). +export type { WorkspaceEgressPolicy } from "../../runtime/egress.js"; export { type WorkspaceFs, WorkspaceFsAdapter } from "./adapter.js"; export { type ArtifactsCommandHost, defineArtifactsCommand } from "./artifacts-command.js"; export { type AssetsCommandHost, defineAssetsCommand } from "./assets-command.js"; @@ -40,5 +41,7 @@ export { export { WorkerShellBackend, type WorkerShellBackendOptions, - type WorkerShellFetcher, + type WorkerShellLoader, + type WorkerShellRuntime, + type WorkerShellSource, } from "./worker-shell.js"; diff --git a/packages/computer/src/backends/worker-shell/worker-shell.test.ts b/packages/computer/src/backends/worker-shell/worker-shell.test.ts index af021566..203557ec 100644 --- a/packages/computer/src/backends/worker-shell/worker-shell.test.ts +++ b/packages/computer/src/backends/worker-shell/worker-shell.test.ts @@ -118,7 +118,9 @@ describe("WorkerShellBackend", () => { backends: [noopFsBackend()], }); await ws.ready(); - const backend = new WorkerShellBackend({ fetcher: () => fetcher }); + const backend = new WorkerShellBackend({ + source: { type: "external-runtime", connect: () => fetcher }, + }); const handle = await backend.connect(); expect(handle.sync).toBe("none"); await handle.close(); @@ -139,7 +141,9 @@ describe("WorkerShellBackend", () => { backends: [noopFsBackend()], }); await ws.ready(); - const backend = new WorkerShellBackend({ fetcher: () => fetcher }); + const backend = new WorkerShellBackend({ + source: { type: "external-runtime", connect: () => fetcher }, + }); const handle = await backend.connect(); const envelope = await handle.rpc.shell.exec({ source: "echo hello" }); @@ -169,7 +173,9 @@ describe("WorkerShellBackend", () => { events: framedStream([{ id: "env", seq: 1, name: "exit", value: 0 }]), }; }); - const backend = new WorkerShellBackend({ fetcher: () => fetcher }); + const backend = new WorkerShellBackend({ + source: { type: "external-runtime", connect: () => fetcher }, + }); const handle = await backend.connect(); const envelope = await handle.rpc.shell.exec({ source: "printenv TOKEN", @@ -193,7 +199,9 @@ describe("WorkerShellBackend", () => { }, }), })); - const handle = await new WorkerShellBackend({ fetcher: () => fetcher }).connect(); + const handle = await new WorkerShellBackend({ + source: { type: "external-runtime", connect: () => fetcher }, + }).connect(); const envelope = await handle.rpc.shell.exec({ source: "bad" }); await expect(envelope.events.getReader().read()).rejects.toMatchObject({ code: "EPROTOCOL" }); }); @@ -212,7 +220,9 @@ describe("WorkerShellBackend", () => { backends: [noopFsBackend()], }); await ws.ready(); - const backend = new WorkerShellBackend({ fetcher: () => fetcher }); + const backend = new WorkerShellBackend({ + source: { type: "external-runtime", connect: () => fetcher }, + }); const handle = await backend.connect(); await handle.rpc.shell.exec({ source: "x", cwd: "/workspace/src", id: "fixed" }); expect(observed?.cwd).toBe("/workspace/src"); @@ -232,7 +242,9 @@ describe("WorkerShellBackend", () => { })); const ws = new Workspace({ storage: new SQLiteTestStorage() as never, - backends: [new WorkerShellBackend({ fetcher: () => fetcher })], + backends: [ + new WorkerShellBackend({ source: { type: "external-runtime", connect: () => fetcher } }), + ], }); await ws.ready(); const handle = await ws.runtime.exec("echo world", { encoding: "utf8" }); @@ -280,6 +292,115 @@ describe("WorkerShellBackend", () => { expect(observedFlags).toEqual(["nodejs_compat"]); }); + it("blocks ambient egress by default", async () => { + let loaderId: string | undefined; + let workerCode: Record | undefined; + const loader = { + get(name: string, getCode: () => Record) { + loaderId = name; + workerCode = getCode(); + return { + getEntrypoint: () => + fakeFetcher(() => ({ + id: "x", + events: framedStream([{ id: "x", seq: 1, name: "exit", value: 0 }]), + })), + }; + }, + }; + const backend = new WorkerShellBackend({ + loader, + workspace: { binding: "WorkspaceHost", id: "abc" }, + ctx: { exports: { WorkspaceServiceProxy: () => ({}) } }, + }); + + await backend.connect(); + + expect(loaderId).toBe("workspace-shell:abc:egress-none"); + expect(workerCode).toMatchObject({ globalOutbound: null }); + }); + + it("omits globalOutbound for direct egress", async () => { + let loaderId: string | undefined; + let workerCode: Record | undefined; + const loader = { + get(name: string, getCode: () => Record) { + loaderId = name; + workerCode = getCode(); + return { + getEntrypoint: () => + fakeFetcher(() => ({ + id: "x", + events: framedStream([{ id: "x", seq: 1, name: "exit", value: 0 }]), + })), + }; + }, + }; + const backend = new WorkerShellBackend({ + loader, + workspace: { binding: "WorkspaceHost", id: "abc" }, + ctx: { exports: { WorkspaceServiceProxy: () => ({}) } }, + egress: { mode: "direct" }, + }); + + await backend.connect(); + + expect(loaderId).toBe("workspace-shell:abc:egress-direct"); + expect(workerCode).not.toHaveProperty("globalOutbound"); + }); + + it("routes ambient egress through an HTTP gateway", async () => { + let loaderId: string | undefined; + let workerCode: Record | undefined; + const gateway = { fetch: async () => new Response() } as Fetcher; + const loader = { + get(name: string, getCode: () => Record) { + loaderId = name; + workerCode = getCode(); + return { + getEntrypoint: () => + fakeFetcher(() => ({ + id: "x", + events: framedStream([{ id: "x", seq: 1, name: "exit", value: 0 }]), + })), + }; + }, + }; + const backend = new WorkerShellBackend({ + loader, + workspace: { binding: "WorkspaceHost", id: "abc" }, + ctx: { exports: { WorkspaceServiceProxy: () => ({}) } }, + egress: { mode: "http-gateway", gateway, revision: "v1" }, + }); + + await backend.connect(); + + expect(loaderId).toBe("workspace-shell:abc:egress-http-gateway-v1"); + expect(workerCode).toMatchObject({ globalOutbound: gateway }); + }); + + it("passes egress policy to an external runtime source", async () => { + const runtime = fakeFetcher(() => ({ + id: "x", + events: framedStream([{ id: "x", seq: 1, name: "exit", value: 0 }]), + })); + let observed: unknown; + const backend = new WorkerShellBackend({ + source: { + type: "external-runtime", + async connect(options) { + observed = options.egress; + return runtime; + }, + }, + egress: { mode: "direct" }, + }); + + await backend.connect(); + + expect(observed).toEqual({ mode: "direct" }); + }); + it("disposes Loader entrypoint and worker handles exactly once", async () => { let entrypointDisposals = 0; let workerDisposals = 0; @@ -336,11 +457,7 @@ describe("WorkerShellBackend", () => { expect(workerDisposals).toBe(1); }); - it("resolves an async fetcher factory once per connect()", async () => { - // A factory that fetches code from KV before minting the - // Worker Loader stub will be async. The backend awaits it - // exactly once per connect(); subsequent shell.exec calls - // reuse the resolved Fetcher. + it("resolves an external runtime source once per connect()", async () => { const fetcher = fakeFetcher(() => ({ id: "x", events: framedStream([{ id: "x", seq: 1, name: "exit", value: 0 }]), @@ -352,9 +469,12 @@ describe("WorkerShellBackend", () => { }); await ws.ready(); const backend = new WorkerShellBackend({ - fetcher: async () => { - factoryCalls += 1; - return fetcher; + source: { + type: "external-runtime", + async connect() { + factoryCalls += 1; + return fetcher; + }, }, }); const handle = await backend.connect(); diff --git a/packages/computer/src/backends/worker-shell/worker-shell.ts b/packages/computer/src/backends/worker-shell/worker-shell.ts index 6890ce5f..478e8ddd 100644 --- a/packages/computer/src/backends/worker-shell/worker-shell.ts +++ b/packages/computer/src/backends/worker-shell/worker-shell.ts @@ -13,12 +13,6 @@ // and reaches its named ShellWorker entrypoint with // .getEntrypoint("ShellWorker"). // -// For deployments that need a different Fetcher source — a -// Workers service binding, a Workers-for-Platforms dispatch -// namespace, a stub from custom code — pass `fetcher` instead. -// The backend stays source-agnostic; the convenience options -// just fill in the Loader callback for the common case. -// // Because there's no second store, the BackendHandle declares // sync: "none". Workspace.push and Workspace.pull short-circuit; // reconcileWatermarks on connect is skipped. @@ -27,13 +21,14 @@ import type { ExecEvent, ShellRPC, SyncRPC, WorkspaceRPC } from "@cloudflare/com import type { BackendHandle, WorkspaceBackend } from "../../backend.js"; import type { WorkspaceServiceProxyProps } from "../../proxy.js"; +import { dynamicWorkerEgress, type WorkspaceEgressPolicy } from "../../runtime/egress.js"; import { SHELL_RUNTIME_MODULES } from "./runtime-modules.js"; import { assembleShellModules, type ShellModuleGroup } from "./shell-modules.js"; // The shape the loaded ShellWorker exposes. The host-side // implementation lives in ./entrypoint.ts; the backend consumes // it through the Fetcher the loader returns. -export interface WorkerShellFetcher { +export interface WorkerShellRuntime { exec(input: { command: string; cwd?: string; @@ -58,7 +53,7 @@ export interface WorkerShellFetcher { // Subset of cloudflare:workers' WorkerLoader the backend uses. // Declared structurally so the file doesn't import the workerd // types at module load. -interface WorkerLoaderLike { +export interface WorkerShellLoader { get( name: string, getCode: () => WorkerLoaderCode | Promise, @@ -81,34 +76,35 @@ interface WorkerLoaderCode { // present at runtime but not in the public type today; declaring // it structurally lets the backend use it without leaning on a // cast in every call site. -interface DurableObjectCtxWithExports { +interface WorkerShellContext { exports: { WorkspaceServiceProxy: (opts: { props: WorkspaceServiceProxyProps }) => unknown; }; } +export type WorkerShellSource = + | { + type: "loader"; + loader: WorkerShellLoader; + workspace: WorkspaceServiceProxyProps; + ctx: unknown; + } + | { + type: "external-runtime"; + connect(options: { + egress: WorkspaceEgressPolicy; + }): WorkerShellRuntime | Promise; + }; + export interface WorkerShellBackendOptions { - // The Worker Loader binding from env. Required when `fetcher` - // is omitted; the backend mints the Dynamic Worker through it. - loader?: WorkerLoaderLike; - - // Reference to the host DO that owns the Workspace. The - // backend uses {binding, id} to mint a WorkspaceServiceProxy - // loopback the shell reaches back through. Required when - // `fetcher` is omitted. + source?: WorkerShellSource; + + loader?: WorkerShellLoader; + workspace?: WorkspaceServiceProxyProps; - // DurableObjectState the backend lives inside. Used to reach - // ctx.exports.WorkspaceServiceProxy(...) when constructing the - // loopback. Required when `fetcher` is omitted. ctx?: unknown; - // The default loader id the backend hands to env.LOADER.get. - // Defaults to `workspace-shell:${workspace.id}` so the loader - // caches one isolate per workspace — a runaway Bash run in one - // workspace can't OOM the shell isolate of another. Override - // when you need a different cache key (multi-version rollouts, - // tenanted shells, etc.). loaderId?: string; // Compatibility date for the Dynamic Worker. Defaults to the @@ -119,17 +115,7 @@ export interface WorkerShellBackendOptions { // ["nodejs_compat"]. compatibilityFlags?: string[]; - // If set, takes precedence over loader / workspace / ctx and - // is used as the Fetcher source directly. Consulted once on - // connect(); the resolved value is held for the life of the - // handle. Async so a caller that fetches code from KV before - // minting the Worker Loader stub isn't forced into a - // synchronous API. - // - // Use this when the Fetcher comes from somewhere other than - // env.LOADER (a service binding, a dispatch namespace, a fake - // in tests). - fetcher?: () => unknown | Promise; + egress?: WorkspaceEgressPolicy; // Selector this backend is registered under in Workspace. // Defaults to "worker-shell"; override when the workspace hosts @@ -144,9 +130,7 @@ export interface WorkerShellBackendOptions { // backend folds them into the Loader modules table on top of // core. A group you never import is unreachable in your bundle // and the bundler drops it, so this is how you opt a command in - // without shipping the rest. Ignored on the `fetcher` path, - // where the caller assembles the modules table itself (use - // assembleShellModules there). + // without shipping the rest. commands?: readonly ShellModuleGroup[]; } @@ -157,32 +141,40 @@ export class WorkerShellBackend implements WorkspaceBackend { readonly type = "worker-shell"; readonly id: string; readonly #options: WorkerShellBackendOptions; + readonly #egress: WorkspaceEgressPolicy; + readonly #egressCacheKey: string; constructor(options: WorkerShellBackendOptions) { this.id = options.id ?? "worker-shell"; - if (options.fetcher === undefined) { + if (options.source === undefined) { if ( options.loader === undefined || options.workspace === undefined || options.ctx === undefined ) { throw new Error( - "WorkerShellBackend: pass either `fetcher` directly or all of " + - "`loader`, `workspace`, and `ctx` so the backend can " + - "mint the Dynamic Worker itself.", + "WorkerShellBackend requires `source` or all of `loader`, `workspace`, and `ctx`.", ); } + } else if ( + options.loader !== undefined || + options.workspace !== undefined || + options.ctx !== undefined + ) { + throw new Error("WorkerShellBackend cannot combine `source` with loader options."); } this.#options = options; + this.#egress = options.egress ?? { mode: "none" }; + this.#egressCacheKey = egressCacheKey(this.#egress); } async connect(): Promise { - const resolved = await this.#resolveFetcher(); - const fetcher = resolved.fetcher as WorkerShellFetcher; + const resolved = await this.#resolveRuntime(); + const runtime = resolved.runtime; const shell: ShellRPC = { async exec(input) { - const envelope = await fetcher.exec({ + const envelope = await runtime.exec({ command: input.source, cwd: input.cwd, id: input.id, @@ -193,11 +185,11 @@ export class WorkerShellBackend implements WorkspaceBackend { return { id: envelope.id, events: decodeFramedEvents(envelope.events) }; }, async getExec(input) { - const envelope = await fetcher.getExec(input); + const envelope = await runtime.getExec(input); return { id: envelope.id, events: decodeFramedEvents(envelope.events) }; }, async killExec(input) { - await fetcher.killExec(input); + await runtime.killExec(input); }, async disposeExec() { // The user Worker has no DB-backed log to dispose; the @@ -218,17 +210,21 @@ export class WorkerShellBackend implements WorkspaceBackend { }; } - async #resolveFetcher(): Promise<{ fetcher: unknown; dispose: () => void }> { - if (this.#options.fetcher !== undefined) { - return { fetcher: await this.#options.fetcher(), dispose: () => {} }; + async #resolveRuntime(): Promise<{ + runtime: WorkerShellRuntime; + dispose: () => void; + }> { + if (this.#options.source?.type === "external-runtime") { + return { + runtime: await this.#options.source.connect({ egress: this.#egress }), + dispose: () => {}, + }; } - // Convenience path: the backend builds the Loader callback - // itself. The constructor checks the required options are - // present, so the casts here are sound. - const loader = this.#options.loader as WorkerLoaderLike; - const workspace = this.#options.workspace as WorkspaceServiceProxyProps; - const ctx = this.#options.ctx as DurableObjectCtxWithExports; - const loaderId = this.#options.loaderId ?? `workspace-shell:${workspace.id}`; + const source = this.#options.source?.type === "loader" ? this.#options.source : undefined; + const loader = source?.loader ?? (this.#options.loader as WorkerShellLoader); + const workspace = source?.workspace ?? (this.#options.workspace as WorkspaceServiceProxyProps); + const ctx = (source?.ctx ?? this.#options.ctx) as WorkerShellContext; + const loaderId = `${this.#options.loaderId ?? `workspace-shell:${workspace.id}`}:${this.#egressCacheKey}`; const compatibilityDate = this.#options.compatibilityDate ?? DEFAULT_COMPAT_DATE; const compatibilityFlags = this.#options.compatibilityFlags ? [...DEFAULT_COMPAT_FLAGS, ...this.#options.compatibilityFlags] @@ -249,9 +245,7 @@ export class WorkerShellBackend implements WorkspaceBackend { // on the host side. HOST: ctx.exports.WorkspaceServiceProxy({ props: workspace }), }, - // The shell has no business reaching the public internet - // on its own. Filesystem RPCs go through env.HOST. - globalOutbound: null, + ...dynamicWorkerEgress(this.#egress), })); let entrypoint: unknown; try { @@ -262,7 +256,7 @@ export class WorkerShellBackend implements WorkspaceBackend { } let disposed = false; return { - fetcher: entrypoint, + runtime: entrypoint as WorkerShellRuntime, dispose: () => { if (disposed) return; disposed = true; @@ -362,6 +356,11 @@ function reshape(event: { return { id: event.id, seq: event.seq, name: "exit", code: event.value as number }; } +function egressCacheKey(policy: WorkspaceEgressPolicy): string { + if (policy.mode !== "http-gateway") return `egress-${policy.mode}`; + return `egress-http-gateway-${policy.revision ?? crypto.randomUUID()}`; +} + function disposeQuietly(value: { [Symbol.dispose]?: () => void }) { try { value[Symbol.dispose]?.(); diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index 2d521050..e99ebdbf 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -57,6 +57,7 @@ export { WorkspaceServiceProxy, type WorkspaceServiceProxyProps, } from "./proxy.js"; +export type { WorkspaceEgressPolicy } from "./runtime/egress.js"; export type { ModuleExecutionEnvelope, ModuleExecutionInput, diff --git a/packages/computer/src/proxy.ts b/packages/computer/src/proxy.ts index 861d9231..09e3ffc9 100644 --- a/packages/computer/src/proxy.ts +++ b/packages/computer/src/proxy.ts @@ -51,6 +51,7 @@ import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; import type { ArtifactsCLIInput, ArtifactsCLIResult } from "./artifacts/index.js"; +import { WORKSPACE_EGRESS_TOKEN_HEADER } from "./runtime/egress.js"; export interface WorkspaceProxyProps { // Name of a DurableObjectNamespace binding in env. The proxy @@ -59,12 +60,21 @@ export interface WorkspaceProxyProps { // Stringified DurableObjectId — typically `ctx.id.toString()` // from inside the owning DO's constructor. id: string; + egressToken?: string; } export class WorkspaceProxy extends WorkerEntrypoint { override async fetch(request: Request): Promise { const url = new URL(request.url); + if (this.ctx.props.egressToken !== undefined) { + const headers = new Headers(request.headers); + headers.set(WORKSPACE_EGRESS_TOKEN_HEADER, this.ctx.props.egressToken); + const stub = this.#hostStub(); + if (stub === undefined) return this.#missingBindingResponse(); + return stub.fetch(new Request(request, { headers })); + } + const callback = url.pathname.match(/^\/__workspace_connect\/([0-9a-f-]{36})\/(health|ws)$/); if (callback?.[2] === "health") { return new Response("ok\n", { @@ -84,21 +94,26 @@ export class WorkspaceProxy extends WorkerEntrypoint)[binding] as - | DurableObjectNamespace - | undefined; - if (!ns) { - return new Response(`WorkspaceProxy: env.${binding} is not a DurableObjectNamespace`, { - status: 500, - }); - } - const stub = ns.get(ns.idFromString(id)); + const stub = this.#hostStub(); + if (stub === undefined) return this.#missingBindingResponse(); return stub.fetch(request); } return new Response("not found", { status: 404 }); } + + #hostStub(): DurableObjectStub | undefined { + const { binding, id } = this.ctx.props; + const ns = (this.env as Record)[binding] as DurableObjectNamespace | undefined; + return ns?.get(ns.idFromString(id)); + } + + #missingBindingResponse(): Response { + return new Response( + `WorkspaceProxy: env.${this.ctx.props.binding} is not a DurableObjectNamespace`, + { status: 500 }, + ); + } } export class ArtifactsCLITarget extends RpcTarget { diff --git a/packages/computer/src/runtime/egress.ts b/packages/computer/src/runtime/egress.ts new file mode 100644 index 00000000..af88e5cd --- /dev/null +++ b/packages/computer/src/runtime/egress.ts @@ -0,0 +1,19 @@ +export const WORKSPACE_EGRESS_TOKEN_HEADER = "x-workspace-egress-token"; + +export type WorkspaceEgressPolicy = + | { mode: "none" } + | { mode: "direct" } + | { mode: "http-gateway"; gateway: Fetcher; revision?: string }; + +export function dynamicWorkerEgress(policy: WorkspaceEgressPolicy): { + globalOutbound?: Fetcher | null; +} { + switch (policy.mode) { + case "none": + return { globalOutbound: null }; + case "direct": + return {}; + case "http-gateway": + return { globalOutbound: policy.gateway }; + } +} diff --git a/packages/computer/tests/proxy-worker.ts b/packages/computer/tests/proxy-worker.ts index 5327ac46..44f3d914 100644 --- a/packages/computer/tests/proxy-worker.ts +++ b/packages/computer/tests/proxy-worker.ts @@ -31,6 +31,10 @@ export class TestStorageDO extends DurableObject { { status: 200 }, ); } + const egressToken = request.headers.get("x-workspace-egress-token"); + if (egressToken !== null) { + return Response.json({ url: request.url, egressToken }); + } return new Response("DO unknown path", { status: 404 }); } } @@ -39,8 +43,11 @@ export default class TestDriver extends WorkerEntrypoint { override async fetch(request: Request): Promise { const binding = request.headers.get("x-test-binding") ?? "COMPUTERD"; const id = request.headers.get("x-test-id") ?? ""; + const egressToken = request.headers.get("x-test-egress-token") ?? undefined; // biome-ignore lint/suspicious/noExplicitAny: ctx.exports isn't in @cloudflare/workers-types yet - const proxy = (this.ctx as any).exports.WorkspaceProxy({ props: { binding, id } }); + const proxy = (this.ctx as any).exports.WorkspaceProxy({ + props: { binding, id, egressToken }, + }); return proxy.fetch(request); } } diff --git a/packages/computer/tests/proxy.test.ts b/packages/computer/tests/proxy.test.ts index c7a1c214..036f6d8c 100644 --- a/packages/computer/tests/proxy.test.ts +++ b/packages/computer/tests/proxy.test.ts @@ -54,6 +54,20 @@ describe("WorkspaceProxy", () => { expect(await websocket.text()).toBe(`from-do:${token}`); }); + it("forwards arbitrary requests with an egress token", async () => { + const res = await SELF.fetch("https://api.example.test/v1/data", { + headers: { + "x-test-id": freshId(), + "x-test-egress-token": "secret-token", + }, + }); + + expect(await res.json()).toEqual({ + url: "https://api.example.test/v1/data", + egressToken: "secret-token", + }); + }); + it("/ws returns 500 when env[binding] is missing", async () => { const res = await SELF.fetch("http://proxy.test/ws", { headers: { "x-test-id": freshId(), "x-test-binding": "NOT_A_BINDING" },