diff --git a/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts index 17194fdd549..6d9b0aa6555 100644 --- a/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts +++ b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts @@ -18,6 +18,7 @@ import { DUPLEX_BODY_CHUNK_RAW_BYTES, DUPLEX_FRAME_VERSION, encodeDuplexFrame, + type DuplexBodyChunkFrame, type DuplexFrame, type DuplexReadyFrame, type DuplexRequestFrame, @@ -859,4 +860,122 @@ describe("duplex bridge local end-to-end harness", () => { const later = await post(); expect(later.status).toBe(200); }, 20000); + + it("forwards a cap-sized malformed-UTF-8 body byte-for-byte and emits no expanded frame", async () => { + // Regression test for the admission-bypass block. The gateway used to decode + // the raw body to a UTF-8 string, then re-encode it. Each invalid byte + // became the three-byte U+FFFD replacement, so a cap-sized malformed body + // expanded to about three times the bytes that admission charged. The gateway + // now preserves the raw bytes, so the body rides base64 frames unchanged and + // no frame carries more than the charged bytes. + const maxBodyBytes = 700 * 1024; + const harness = await createHarness({ maxBodyBytes }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + harness.api.setResponder((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + }); + + // 0xFF is never a valid UTF-8 byte. A cap-sized body of 0xFF bytes is the + // worst case for the old expansion path. + const malformed = Buffer.alloc(maxBodyBytes, 0xff); + const response = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: new Uint8Array(malformed), + }); + expect(response.status).toBe(200); + + // The observed frames are the exact bytes the gateway sent to the host. The + // envelope reports the raw byte count, not an expanded count. + const requestFrame = harness.observedFrames.find( + (frame): frame is DuplexRequestFrame => frame.type === "request", + ); + expect(requestFrame?.bodyByteCount).toBe(maxBodyBytes); + + // The body_chunk frames decode to the exact source bytes. No frame carried an + // expanded body, so no body larger than the charged bytes reached the host. + const bodyChunks = harness.observedFrames.filter( + (frame): frame is DuplexBodyChunkFrame => frame.type === "body_chunk", + ); + const reassembled = Buffer.concat( + bodyChunks.map((frame) => Buffer.from(frame.data, "base64")), + ); + expect(reassembled.length).toBe(maxBodyBytes); + expect(reassembled.equals(malformed)).toBe(true); + + // The route stays usable after the malformed body. + const later = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: JSON.stringify({ ok: true }), + }); + expect(later.status).toBe(200); + }, 30000); + + it("caps concurrent body reads at the retained-byte model, then releases the reserve", async () => { + // Production-size accounting. Admission charges maxBodyBytes per read against + // the inflight bound, so the bound admits floor(bound / maxBodyBytes) reads. + // With the production constants that is floor(64 MiB / (10 MiB + 1)) = 6. The + // reservation is the worst-case body per read, so a small body still charges + // the full maxBodyBytes. So the test drives the model with small bodies and + // asserts the admitted count matches the model. + const maxBodyBytes = 10 * 1024 * 1024 + 1; + const maxInflightBodyBytes = 64 * 1024 * 1024; + const admitCount = Math.floor(maxInflightBodyBytes / maxBodyBytes); + expect(admitCount).toBe(6); + const harness = await createHarness({ + maxBodyBytes, + maxInflightBodyBytes, + // A short response deadline releases each hung read fast. + responseTimeoutMs: 800, + }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + // Each admitted read hangs on the forward until it times out, so all admitted + // reads hold their reserve at once and the next read cannot reserve. + harness.setForwardMode("hang"); + const body = "a".repeat(1024); + const post = () => + fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body, + }); + const responses = await Promise.all( + Array.from({ length: admitCount + 1 }, () => post()), + ); + const statuses = responses.map((response) => response.status); + // The model admits exactly admitCount reads. Each admitted read times out + // with a 502. The one read past the model is shed with a 503 before it + // buffers a body. + const shedCount = statuses.filter((status) => status === 503).length; + const admittedTimedOut = statuses.filter((status) => status === 502).length; + expect(shedCount).toBe(1); + expect(admittedTimedOut).toBe(admitCount); + const shed = responses.find((response) => response.status === 503); + await expect(shed?.json()).resolves.toEqual({ error: "bridge_busy" }); + + // The admitted reads released their reserve on timeout. A later request is + // admitted again and reaches the API. + harness.setForwardMode("proxy"); + harness.api.setResponder((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + }); + const later = await post(); + expect(later.status).toBe(200); + }, 30000); }); diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 26cea1c56b7..aecce32ec1a 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -2557,6 +2557,78 @@ describe("sandbox adapter execution targets", () => { } }); + it("rejects an invalid-UTF-8 request body with a clean 400 and never forwards it", async () => { + // The file gateway writes the request body as a JSON string that the host + // reader consumes. A valid JSON body is always valid UTF-8. An invalid body + // would expand during a lossy UTF-8 round trip, so the gateway would retain + // and forward more bytes than admission charged. The gateway rejects an + // invalid-UTF-8 body with a clean 400 before it writes the body, so no + // expanded body reaches the host and the host runs no mutation. + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-invalid-utf8-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + const runtimeRootDir = path.join(remoteCwd, ".paperclip-runtime", "codex"); + await mkdir(runtimeRootDir, { recursive: true }); + + const requests: Array<{ method: string; url: string }> = []; + const apiServer = createServer((req, res) => { + requests.push({ method: req.method ?? "GET", url: req.url ?? "/" }); + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + }); + await new Promise((resolve, reject) => { + apiServer.once("error", reject); + apiServer.listen(0, "127.0.0.1", () => resolve()); + }); + const address = apiServer.address(); + if (!address || typeof address === "string") { + throw new Error("Expected the bridge test API server to listen on a TCP port."); + } + + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "e2b", + environmentId: "env-1", + leaseId: "lease-1", + remoteCwd, + runner: createLocalSandboxRunner(), + timeoutMs: 30_000, + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-bridge-invalid-utf8", + target, + runtimeRootDir, + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: `http://127.0.0.1:${address.port}`, + maxBodyBytes: 4096, + }); + try { + // 0xFF is never a valid UTF-8 byte, so this body is not valid UTF-8. + const malformed = new Uint8Array([0x7b, 0xff, 0xff, 0x7d]); + const response = await fetch(`${bridge!.env.PAPERCLIP_API_URL}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${bridge!.env.PAPERCLIP_API_KEY}`, + "content-type": "application/json", + }, + body: malformed, + }); + // The gateway answers a clean, non-retryable 400 and leaks no internal + // message. + expect(response.status).toBe(400); + await expect(response.json()).resolves.toEqual({ error: "invalid_utf8_body" }); + // The gateway rejected the body before it wrote the request, so the host + // reader forwarded nothing and ran no mutation. + expect(requests).toEqual([]); + } finally { + await bridge?.stop(); + await new Promise((resolve) => apiServer.close(() => resolve())); + } + }); + it("fails an oversized host response with a non-retryable 409 so a committed mutation never repeats", async () => { // The host receives the request and commits the mutation, then sends a // response body over the size limit. The forward reads the body after the diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index 46ddab29e3b..09930cad077 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -2146,7 +2146,38 @@ async function readBody(req) { throw new BridgeBodyTooLargeError(); } } - return Buffer.concat(chunks).toString("utf8"); + // Return the raw request bytes. The duplex gateway base64-encodes these bytes + // unchanged, so a malformed body cannot expand during transport and cannot + // exceed the bytes that admission charged. The file gateway decodes a valid + // UTF-8 body with decodeUtf8Body and rejects an invalid one. + return Buffer.concat(chunks); +} + +// Strict UTF-8 decoder for the file gateway. It throws on any invalid byte, so +// the gateway rejects a malformed body instead of expanding it. A single decode +// call does not stream, so the gateway reuses one decoder across requests. +const bridgeUtf8Decoder = new TextDecoder("utf8", { fatal: true }); + +// Typed marker for a request body that is not valid UTF-8. Only the file gateway +// uses it. That gateway writes the body as a JSON string that the host reader +// consumes. A valid JSON body is always valid UTF-8 and round-trips +// byte-for-byte. An invalid body would expand, because each invalid byte becomes +// the 3-byte U+FFFD replacement. The gateway must not retain or forward more +// bytes than admission charged, so it answers a clean 400 instead. +class BridgeInvalidBodyError extends Error { + constructor() { + super("invalid_utf8_body"); + this.code = "invalid_utf8_body"; + } +} + +// Decode a request body Buffer to a UTF-8 string, or throw BridgeInvalidBodyError. +function decodeUtf8Body(buffer) { + try { + return bridgeUtf8Decoder.decode(buffer); + } catch (error) { + throw new BridgeInvalidBodyError(); + } } function tokensMatch(received) { @@ -2215,7 +2246,7 @@ async function runFileGateway() { return; } try { - const requestBody = await readBody(req); + const requestBody = decodeUtf8Body(await readBody(req)); const payload = { id: requestId, method: req.method || "GET", @@ -2258,6 +2289,10 @@ async function runFileGateway() { writeJsonResponse(res, 413, { error: "request_too_large" }); return; } + if (error && error.code === "invalid_utf8_body") { + writeJsonResponse(res, 400, { error: "invalid_utf8_body" }); + return; + } writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) }); } }); @@ -2293,26 +2328,26 @@ async function runFileGateway() { }); } -// Split one whole body buffer into body_chunk frames. The gateway buffers the -// whole source body once, then splits it here. Each frame carries one fixed raw -// slice as base64 text, except the final frame, which carries the remaining -// bytes. A zero-length body yields no frame. Send-side true streaming is a -// separate goal; this whole-body split is acceptable for this gateway. -function splitDuplexBodyIntoChunks(id, body) { - const frames = []; +// Yield the body_chunk frames for one request body, one frame at a time. The +// gateway encodes and writes each frame, then drops it before it builds the next +// frame. So the gateway holds the raw body plus at most one base64 frame, not +// the raw body plus every base64 frame. Each frame carries one fixed raw slice +// as base64 text, except the final frame, which carries the remaining bytes. A +// zero-length body yields no frame. See the retained-peak note at the duplex +// send site. +function* duplexBodyChunkFrames(id, body) { let seq = 0; for (let offset = 0; offset < body.length; offset += DUPLEX_BODY_CHUNK_RAW_BYTES) { const slice = body.subarray(offset, offset + DUPLEX_BODY_CHUNK_RAW_BYTES); - frames.push({ + yield { version: DUPLEX_FRAME_VERSION, type: "body_chunk", id: id, seq: seq, data: slice.toString("base64"), - }); + }; seq += 1; } - return frames; } function runDuplexGateway() { @@ -2524,7 +2559,7 @@ function runDuplexGateway() { return; } try { - const requestBodyBuffer = Buffer.from(await readBody(req), "utf8"); + const requestBodyBuffer = await readBody(req); if (unavailable) { writeJsonResponse(res, 503, { error: "bridge_unavailable" }); return; @@ -2547,26 +2582,35 @@ function runDuplexGateway() { writeJsonResponse(res, 413, { error: "request_too_large" }); return; } - // Pre-encode every body_chunk frame and enforce the frame size bound on - // each. A fixed raw slice never exceeds the bound, so this guard is - // defensive. On a rejection, fail this one local request with a clean 413. - // The frames never leave the gateway, so no other in-flight request is - // affected and the channel stays open. - const chunkFrames = splitDuplexBodyIntoChunks(requestId, requestBodyBuffer); - const encodedChunks = []; + // Validate every body_chunk frame against the frame bound before the + // gateway emits the envelope. A fixed 256 KiB raw slice never exceeds the + // 1 MB frame bound, so this guard is defensive. On a rejection the gateway + // fails this one local request with a clean 413 and emits nothing, so no + // partial body reaches the host and the channel stays open. let chunkTooLarge = false; - for (const chunk of chunkFrames) { - const encodedChunk = encodeDuplexFrameChecked(chunk); - if (!encodedChunk.ok) { + for (const frame of duplexBodyChunkFrames(requestId, requestBodyBuffer)) { + if (!encodeDuplexFrameChecked(frame).ok) { chunkTooLarge = true; break; } - encodedChunks.push(encodedChunk.line); } if (chunkTooLarge) { writeJsonResponse(res, 413, { error: "request_too_large" }); return; } + // Retained-peak note. The gateway preserves the raw request bytes end to + // end and streams the body as body_chunk frames, one frame at a time. It + // does not hold the raw body plus the full base64 frame array plus the + // full encoded-line array (about 3.7x the raw body). The retained peak per + // read is now the raw body plus one encoded frame: + // raw body <= maxBodyBytes (about 10 MiB) + // one body_chunk <= DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB) + // so the peak is about maxBodyBytes + 1 MB per read. Admission charges + // maxBodyBytes per read against the 64 MiB inflight bound. The bound admits + // at most floor(64 MiB / maxBodyBytes) = 6 reads, so the true retained peak + // stays near 6 * (10 MiB + 1 MB) ~= 66 MiB. That is a fixed small margin + // over the admission threshold, not a multiple of the body. A malformed + // body cannot expand, because its raw bytes ride base64 frames unchanged. const response = await new Promise((resolve) => { const timer = setTimeout(() => { responseAssembly.delete(requestId); @@ -2580,7 +2624,11 @@ function runDuplexGateway() { }, responseTimeoutMs); pending.set(requestId, { resolve: resolve, timer: timer }); process.stdout.write(encodedRequest.line); - for (const line of encodedChunks) process.stdout.write(line); + for (const frame of duplexBodyChunkFrames(requestId, requestBodyBuffer)) { + const encoded = encodeDuplexFrameChecked(frame); + // Pass 1 proved every frame fits, so encoded.ok is always true here. + process.stdout.write(encoded.line); + } }); res.statusCode = typeof response.status === "number" ? response.status : 200; for (const [key, value] of Object.entries(response.headers || {})) {