From ea8bf4d95a065ad5483aece4a3bc3ccc98e0e324 Mon Sep 17 00:00:00 2001 From: Priya Raman Date: Wed, 26 Aug 2026 23:33:01 +0000 Subject: [PATCH] feat(adapter-utils): bound the sandbox side of the http2 bridge clients Each sandbox HTTP/2 gateway client now waits for the host's first remoteSettings event, with a time bound, before it opens its first stream. Each client also counts response bytes and fails closed past the configured body size limit. This defends the sandbox process; it does not change any host-side bound or default. Co-authored-by: Paperclip --- .../src/sandbox-callback-bridge.test.ts | 321 +++++++++++++++++- .../src/sandbox-callback-bridge.ts | 222 ++++++++---- 2 files changed, 481 insertions(+), 62 deletions(-) diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts index 6fdb44805bc..99a7c5d8e8f 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts @@ -3,7 +3,9 @@ import { createServer } from "node:http"; import { mkdir, mkdtemp, readFile, readdir, rm, utimes, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { Duplex, duplexPair } from "node:stream"; import { promisify } from "node:util"; +import http2 from "node:http2"; import { afterEach, describe, expect, it, vi } from "vitest"; import { getActiveStepContext, measureStartupStep } from "./acpx-engine/startup-timing.js"; @@ -14,6 +16,7 @@ import { createFileSystemSandboxCallbackBridgeQueueClient, createSandboxCallbackBridgeAsset, createSandboxCallbackBridgeToken, + createSandboxHttp2BridgeGateway, getSandboxCallbackBridgeServerSource, sandboxCallbackBridgeDirectories, syncRemoteTextFileWithHashSkip, @@ -3100,8 +3103,17 @@ describe("sandbox callback bridge", () => { async function startHttp2GatewayForTest(options: { bridgeToken: string; maxBodyBytes?: number; + responseTimeoutMs?: number; forwardRequest: (request: Http2BridgeForwardRequest) => Promise; - }): Promise<{ baseUrl: string; stop: () => Promise }> { + onSession?: (session: http2.ServerHttp2Session) => void; + /** + * When true, hold every byte the host writes toward the sandbox process + * until a test calls the returned `releaseHostBytes`. This lets a test + * prove the sandbox gateway waits for the host's settings frame before + * it opens its first stream. + */ + gateHostBytes?: boolean; + }): Promise<{ baseUrl: string; stop: () => Promise; releaseHostBytes: () => void }> { const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-http2-test-")); cleanupDirs.push(rootDir); const entrypoint = path.join(rootDir, "paperclip-bridge-server.mjs"); @@ -3130,6 +3142,9 @@ describe("sandbox callback bridge", () => { ...(options.maxBodyBytes != null ? { PAPERCLIP_BRIDGE_MAX_BODY_BYTES: String(options.maxBodyBytes) } : {}), + ...(options.responseTimeoutMs != null + ? { PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS: String(options.responseTimeoutMs) } + : {}), }, stdio: ["pipe", "pipe", "pipe"], }); @@ -3182,9 +3197,24 @@ describe("sandbox callback bridge", () => { } }); + let hostBytesOpen = !options.gateHostBytes; + const bufferedHostBytes: Buffer[] = []; + const releaseHostBytes = () => { + if (hostBytesOpen) return; + hostBytesOpen = true; + for (const chunk of bufferedHostBytes.splice(0)) { + child.stdin.write(chunk); + } + }; + const channel: CommandManagedDuplexChannel = { write: (data) => { - child.stdin.write(Buffer.from(data)); + const chunk = Buffer.from(data); + if (hostBytesOpen) { + child.stdin.write(chunk); + } else { + bufferedHostBytes.push(chunk); + } }, onData: (listener) => { dataListeners.push(listener); @@ -3205,6 +3235,7 @@ describe("sandbox callback bridge", () => { const handle = createHttp2BridgeServer({ bridgeToken: options.bridgeToken, forwardRequest: options.forwardRequest, + onSession: options.onSession, }); const boundDuplex = handle.bindChannel(channel); dispatchStarted = true; @@ -3224,7 +3255,7 @@ describe("sandbox callback bridge", () => { }; cleanupFns.push(stop); - return { baseUrl: `http://127.0.0.1:${assignedPort}`, stop }; + return { baseUrl: `http://127.0.0.1:${assignedPort}`, stop, releaseHostBytes }; } it("forwards the exact request body bytes to the HTTP/2 host handler, including a non-ASCII character", async () => { @@ -3434,4 +3465,288 @@ describe("sandbox callback bridge", () => { expect(typeof seenRequests[0]?.body).toBe("string"); expect(seenRequests[0]?.body).toBe(requestBodyText); }); + + /** + * Wrap one side of a paired in-memory `Duplex` as a minimal + * `CommandManagedDuplexChannel`, so the real host `createHttp2BridgeServer` + * can bind to it directly, with no spawned process in between. + */ + function channelFromDuplex(duplex: Duplex): CommandManagedDuplexChannel { + const dataListeners: Array<(chunk: Uint8Array) => void> = []; + duplex.on("data", (chunk: Buffer) => { + for (const listener of dataListeners) listener(chunk); + }); + return { + write: (data) => { + duplex.write(Buffer.from(data)); + }, + onData: (listener) => { + dataListeners.push(listener); + }, + onExit: (listener) => { + duplex.once("end", () => listener({ exitCode: null, transportClosed: true })); + }, + stop: () => { + duplex.destroy(); + }, + close: async () => { + duplex.end(); + }, + }; + } + + /** + * Hold back every chunk `source` emits until a test calls `release()`. A + * test uses this to delay the host's settings frame from reaching the + * exported sandbox HTTP/2 client gateway, so it can prove the gateway + * waits for that frame before it opens its first stream. + */ + function createGatedDuplex(source: Duplex): { gated: Duplex; release: () => void } { + let open = false; + const buffered: Buffer[] = []; + const gated = new Duplex({ + read() {}, + write(chunk, _encoding, callback) { + source.write(chunk); + callback(); + }, + final(callback) { + source.end(); + callback(); + }, + }); + source.on("data", (chunk: Buffer) => { + if (open) { + gated.push(chunk); + } else { + buffered.push(chunk); + } + }); + source.on("end", () => { + if (open) gated.push(null); + }); + return { + gated, + release: () => { + open = true; + for (const chunk of buffered.splice(0)) gated.push(chunk); + }, + }; + } + + it("the gateway client opens its first stream only after the host settings frame arrives", async () => { + const bridgeToken = createSandboxCallbackBridgeToken(); + const [serverSide, clientSide] = duplexPair(); + let sawStream = false; + const handle = createHttp2BridgeServer({ + bridgeToken, + forwardRequest: async () => ({ status: 200, headers: {}, body: "" }), + onSession: (session) => { + session.on("stream", () => { + sawStream = true; + }); + }, + }); + handle.bindChannel(channelFromDuplex(serverSide)); + + const gate = createGatedDuplex(clientSide); + const gateway = createSandboxHttp2BridgeGateway({ + bridgeToken, + createConnection: () => gate.gated, + }); + cleanupFns.push(async () => { + await gateway.close(); + serverSide.destroy(); + clientSide.destroy(); + }); + + const forwardPromise = gateway.forwardRequest({ + method: "GET", + path: "/api/agents/me", + query: "", + headers: {}, + body: Buffer.alloc(0), + receivedToken: bridgeToken, + }); + + // Wait before the settings frame is let through, so an early dispatch + // has time to open a stream if the gateway does not wait for it. + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(sawStream).toBe(false); + + gate.release(); + const response = await forwardPromise; + expect(response.status).toBe(200); + expect(sawStream).toBe(true); + }, 15_000); + + it("a host that sends no settings frame fails the first request at the bound", async () => { + const bridgeToken = createSandboxCallbackBridgeToken(); + const blackHole = new Duplex({ + read() {}, + write(_chunk, _encoding, callback) { + callback(); + }, + }); + const gateway = createSandboxHttp2BridgeGateway({ + bridgeToken, + createConnection: () => blackHole, + responseTimeoutMs: 50, + }); + // `gateway.close()` sends a graceful GOAWAY and waits for the transport + // to end, which a black-hole duplex never does. Destroy the transport + // directly instead, so teardown does not hang. + cleanupFns.push(async () => { + blackHole.destroy(); + }); + + await expect( + gateway.forwardRequest({ + method: "GET", + path: "/api/agents/me", + query: "", + headers: {}, + body: Buffer.alloc(0), + receivedToken: bridgeToken, + }), + ).rejects.toThrow("Timed out waiting for the HTTP/2 bridge host settings handshake."); + }); + + it("a failed settings handshake raises no unhandled rejection when no request waits", async () => { + const bridgeToken = createSandboxCallbackBridgeToken(); + const blackHole = new Duplex({ + read() {}, + write(_chunk, _encoding, callback) { + callback(); + }, + }); + const unhandled: unknown[] = []; + const onUnhandledRejection = (reason: unknown) => { + unhandled.push(reason); + }; + process.on("unhandledRejection", onUnhandledRejection); + try { + const gateway = createSandboxHttp2BridgeGateway({ + bridgeToken, + createConnection: () => blackHole, + responseTimeoutMs: 20, + }); + // `gateway.close()` sends a graceful GOAWAY and waits for the + // transport to end, which a black-hole duplex never does. Destroy the + // transport directly instead, so teardown does not hang. + cleanupFns.push(async () => { + blackHole.destroy(); + }); + + // No request is in flight while the handshake bound passes, so the + // stored handshake promise settles with nobody awaiting it yet. + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(unhandled).toHaveLength(0); + + // A request made after the bound already passed must still fail + // closed with the fixed handshake message. This proves the rejection + // was stored, not swallowed. + await expect( + gateway.forwardRequest({ + method: "GET", + path: "/api/agents/me", + query: "", + headers: {}, + body: Buffer.alloc(0), + receivedToken: bridgeToken, + }), + ).rejects.toThrow("Timed out waiting for the HTTP/2 bridge host settings handshake."); + } finally { + process.off("unhandledRejection", onUnhandledRejection); + } + }, 10_000); + + it("the gateway client fails closed on a response body past the configured size limit", async () => { + const bridgeToken = createSandboxCallbackBridgeToken(); + const maxBodyBytes = 16; + const [serverSide, clientSide] = duplexPair(); + const handle = createHttp2BridgeServer({ + bridgeToken, + forwardRequest: async () => ({ + status: 200, + headers: { "content-type": "text/plain" }, + body: "x".repeat(maxBodyBytes + 1), + }), + }); + handle.bindChannel(channelFromDuplex(serverSide)); + + const gateway = createSandboxHttp2BridgeGateway({ + bridgeToken, + createConnection: () => clientSide, + maxBodyBytes, + }); + cleanupFns.push(async () => { + await gateway.close(); + serverSide.destroy(); + clientSide.destroy(); + }); + + await expect( + gateway.forwardRequest({ + method: "GET", + path: "/api/agents/me", + query: "", + headers: {}, + body: Buffer.alloc(0), + receivedToken: bridgeToken, + }), + ).rejects.toThrow("Bridge response body exceeded the configured size limit."); + }); + + it("the generated gateway asset opens its first stream only after the host settings frame arrives", async () => { + const bridgeToken = createSandboxCallbackBridgeToken(); + let sawStream = false; + const gateway = await startHttp2GatewayForTest({ + bridgeToken, + gateHostBytes: true, + onSession: (session) => { + session.on("stream", () => { + sawStream = true; + }); + }, + forwardRequest: async () => ({ status: 200, headers: {}, body: "" }), + }); + + const responsePromise = fetch(`${gateway.baseUrl}/api/agents/me`, { + method: "GET", + headers: { authorization: `Bearer ${bridgeToken}` }, + }); + + await new Promise((resolve) => setTimeout(resolve, 200)); + expect(sawStream).toBe(false); + + gateway.releaseHostBytes(); + const response = await responsePromise; + expect(response.status).toBe(200); + expect(sawStream).toBe(true); + }, 15_000); + + it("the generated gateway asset fails closed on a response body past the configured size limit", async () => { + const bridgeToken = createSandboxCallbackBridgeToken(); + const maxBodyBytes = 32; + const oversizedResponseBody = "x".repeat(maxBodyBytes + 1); + const gateway = await startHttp2GatewayForTest({ + bridgeToken, + maxBodyBytes, + forwardRequest: async () => ({ + status: 200, + headers: { "content-type": "text/plain" }, + body: oversizedResponseBody, + }), + }); + + const response = await fetch(`${gateway.baseUrl}/api/agents/me`, { + method: "GET", + headers: { authorization: `Bearer ${bridgeToken}` }, + }); + expect(response.status).toBe(502); + await expect(response.json()).resolves.toMatchObject({ + error: "Bridge response body exceeded the configured size limit.", + }); + }, 15_000); }); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index 2ca6a8dec6a..e344d0be4ea 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -1890,6 +1890,19 @@ export interface CreateSandboxHttp2BridgeGatewayOptions { * `http2-bridge-server.ts` to know which requests need a retry elsewhere. */ onGoaway?: (record: { lastStreamId: number; errorCode: number }) => void; + /** + * The time bound, in milliseconds, on the wait for the host's first + * `remoteSettings` event. The gateway opens no stream before that event + * arrives or before this bound passes, whichever comes first. The default + * is {@link DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS}. + */ + responseTimeoutMs?: number; + /** + * The response body byte limit. The gateway counts the bytes it collects + * from each response and fails closed, destroying the stream, past this + * limit. The default is {@link DEFAULT_BRIDGE_MAX_BODY_BYTES}. + */ + maxBodyBytes?: number; } function forwardOneHttp2Request( @@ -1902,6 +1915,7 @@ function forwardOneHttp2Request( headers: Record; body: Buffer; }, + maxBodyBytes: number, ): Promise { return new Promise((resolve, reject) => { const query = request.query.trim(); @@ -1920,7 +1934,8 @@ function forwardOneHttp2Request( reject(error instanceof Error ? error : new Error(String(error))); return; } - const chunks: Buffer[] = []; + let chunks: Buffer[] = []; + let receivedBytes = 0; let responseHeaders: Record = {}; let status = 502; let settled = false; @@ -1938,7 +1953,17 @@ function forwardOneHttp2Request( responseHeaders[key] = Array.isArray(value) ? value.join(", ") : String(value); } }); - stream.on("data", (chunk: Buffer) => chunks.push(chunk)); + stream.on("data", (chunk: Buffer) => { + if (settled) return; + receivedBytes += chunk.length; + if (receivedBytes > maxBodyBytes) { + chunks = []; + stream.destroy(); + settle(() => reject(new Error("Bridge response body exceeded the configured size limit."))); + return; + } + chunks.push(chunk); + }); stream.once("end", () => settle(() => resolve({ status, headers: responseHeaders, body: Buffer.concat(chunks) }))); stream.once("error", (error) => settle(() => reject(error instanceof Error ? error : new Error(String(error)))), @@ -1952,6 +1977,35 @@ function forwardOneHttp2Request( }); } +/** + * Create the settings-handshake wait for one HTTP/2 client session. The + * client must wait for the host's first `remoteSettings` event before it + * opens its first stream (see the module-level notes on `forwardRequest`). + * A timer bounds the wait so a silent host never hangs a request forever. + * The timer is unref'd, so it never holds the process open, and the + * returned promise carries a no-operation rejection handler, so a rejection + * with no request yet in flight never raises an unhandled rejection. + */ +function createRemoteSettingsReadyPromise(session: http2.ClientHttp2Session, timeoutMs: number): Promise { + let settled = false; + const ready = new Promise((resolve, reject) => { + const timer = setTimeout(() => { + if (settled) return; + settled = true; + reject(new Error("Timed out waiting for the HTTP/2 bridge host settings handshake.")); + }, timeoutMs); + timer.unref(); + session.once("remoteSettings", () => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve(); + }); + }); + ready.catch(() => undefined); + return ready; +} + /** * Create the sandbox HTTP/2 client gateway. It opens one HTTP/2 client * session on the transport `createConnection` returns, and forwards each @@ -1965,6 +2019,8 @@ export function createSandboxHttp2BridgeGateway( ): SandboxHttp2BridgeGateway { const authority = options.authority?.trim() || SANDBOX_HTTP2_GATEWAY_DEFAULT_AUTHORITY; const headerAllowlist = options.headerAllowlist ?? DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST; + const responseTimeoutMs = options.responseTimeoutMs ?? DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS; + const maxBodyBytes = options.maxBodyBytes ?? DEFAULT_BRIDGE_MAX_BODY_BYTES; const session = http2.connect(`http://${authority}`, { createConnection: options.createConnection, }); @@ -1975,6 +2031,10 @@ export function createSandboxHttp2BridgeGateway( session.on("goaway", (errorCode: number, lastStreamId: number) => { options.onGoaway?.({ lastStreamId, errorCode }); }); + // One handshake promise for the whole session, created here before any + // request. Every `forwardRequest` call awaits this same stored promise, so + // exactly one timer is armed per session, not one per request. + const remoteSettingsReady = createRemoteSettingsReadyPromise(session, responseTimeoutMs); return { forwardRequest(request: SandboxHttp2BridgeGatewayRequest): Promise { @@ -1984,14 +2044,20 @@ export function createSandboxHttp2BridgeGateway( if (!compareBridgeTokensConstantTime(options.bridgeToken, request.receivedToken)) { return Promise.reject(new Error("Invalid bridge token.")); } - return forwardOneHttp2Request(session, { - bridgeToken: options.bridgeToken, - method: request.method, - path: request.path, - query: request.query, - headers: sanitizeSandboxCallbackBridgeHeaders(request.headers, headerAllowlist), - body: request.body, - }); + return remoteSettingsReady.then(() => + forwardOneHttp2Request( + session, + { + bridgeToken: options.bridgeToken, + method: request.method, + path: request.path, + query: request.query, + headers: sanitizeSandboxCallbackBridgeHeaders(request.headers, headerAllowlist), + body: request.body, + }, + maxBodyBytes, + ), + ); }, close(): Promise { return new Promise((resolve) => { @@ -2387,6 +2453,13 @@ function runHttp2Gateway() { const authority = "bridge.internal"; let session = null; let unavailable = false; + // One handshake promise for the whole session, created where the session + // is created, before any request. Every forwarded request awaits this + // same stored promise, so exactly one timer is armed per session, not one + // per request. The catch below stops an unobserved rejection (a failed + // handshake with no request yet in flight) from raising an unhandled + // rejection warning. + let remoteSettingsReady = null; function openSession() { if (session) return session; @@ -2403,61 +2476,92 @@ function runHttp2Gateway() { session.on("goaway", (errorCode, lastStreamId) => { diag("host sent GOAWAY (errorCode=" + errorCode + ", lastStreamId=" + lastStreamId + ")"); }); + const activeSession = session; + let handshakeSettled = false; + remoteSettingsReady = new Promise((resolve, reject) => { + const timer = setTimeout(function () { + if (handshakeSettled) return; + handshakeSettled = true; + reject(new Error("Timed out waiting for the HTTP/2 bridge host settings handshake.")); + }, responseTimeoutMs); + timer.unref(); + activeSession.once("remoteSettings", function () { + if (handshakeSettled) return; + handshakeSettled = true; + clearTimeout(timer); + resolve(); + }); + }); + remoteSettingsReady.catch(function () {}); return session; } function forwardOverHttp2(request) { - return new Promise((resolve, reject) => { - const activeSession = openSession(); - const query = request.query || ""; - const pathWithQuery = - query.length === 0 ? request.path : request.path + (query.charAt(0) === "?" ? query : "?" + query); - const outboundHeaders = Object.assign( - { - ":method": request.method, - ":path": pathWithQuery, - authorization: "Bearer " + bridgeToken, - }, - request.headers, - ); - let stream; - try { - stream = activeSession.request(outboundHeaders, { endStream: request.body.length === 0 }); - } catch (error) { - reject(error instanceof Error ? error : new Error(String(error))); - return; - } - const chunks = []; - let status = 502; - let responseHeaders = {}; - let settled = false; - const settle = (run) => { - if (settled) return; - settled = true; - run(); - }; - stream.on("response", (headers) => { - const raw = headers[":status"]; - status = typeof raw === "number" ? raw : Number(raw) || 502; - responseHeaders = {}; - for (const [key, value] of Object.entries(headers)) { - if (key.charAt(0) === ":" || value == null) continue; - responseHeaders[key] = Array.isArray(value) ? value.join(", ") : String(value); + const activeSession = openSession(); + const handshakeReady = remoteSettingsReady; + return handshakeReady.then(function () { + return new Promise((resolve, reject) => { + const query = request.query || ""; + const pathWithQuery = + query.length === 0 ? request.path : request.path + (query.charAt(0) === "?" ? query : "?" + query); + const outboundHeaders = Object.assign( + { + ":method": request.method, + ":path": pathWithQuery, + authorization: "Bearer " + bridgeToken, + }, + request.headers, + ); + let stream; + try { + stream = activeSession.request(outboundHeaders, { endStream: request.body.length === 0 }); + } catch (error) { + reject(error instanceof Error ? error : new Error(String(error))); + return; + } + let chunks = []; + let receivedBytes = 0; + let status = 502; + let responseHeaders = {}; + let settled = false; + const settle = (run) => { + if (settled) return; + settled = true; + run(); + }; + stream.on("response", (headers) => { + const raw = headers[":status"]; + status = typeof raw === "number" ? raw : Number(raw) || 502; + responseHeaders = {}; + for (const [key, value] of Object.entries(headers)) { + if (key.charAt(0) === ":" || value == null) continue; + responseHeaders[key] = Array.isArray(value) ? value.join(", ") : String(value); + } + }); + stream.on("data", (chunk) => { + if (settled) return; + receivedBytes += chunk.length; + if (receivedBytes > maxBodyBytes) { + chunks = []; + stream.destroy(); + settle(() => reject(new Error("Bridge response body exceeded the configured size limit."))); + return; + } + chunks.push(chunk); + }); + stream.once("end", () => + settle(() => resolve({ status: status, headers: responseHeaders, body: Buffer.concat(chunks) })), + ); + stream.once("error", (error) => + settle(() => reject(error instanceof Error ? error : new Error(String(error)))), + ); + stream.once("aborted", () => settle(() => reject(new Error("Bridge HTTP/2 stream aborted.")))); + if (request.body.length > 0) { + stream.end(request.body); + } else if (!stream.writableEnded) { + stream.end(); } }); - stream.on("data", (chunk) => chunks.push(chunk)); - stream.once("end", () => - settle(() => resolve({ status: status, headers: responseHeaders, body: Buffer.concat(chunks) })), - ); - stream.once("error", (error) => - settle(() => reject(error instanceof Error ? error : new Error(String(error)))), - ); - stream.once("aborted", () => settle(() => reject(new Error("Bridge HTTP/2 stream aborted.")))); - if (request.body.length > 0) { - stream.end(request.body); - } else if (!stream.writableEnded) { - stream.end(); - } }); }