From 8c8692df6e4ec19b9401e861af3485f790f12cfd Mon Sep 17 00:00:00 2001 From: Thomas <9669989+thomaslwang@users.noreply.github.com> Date: Sat, 31 Oct 2026 07:26:51 -0700 Subject: [PATCH] proxy: an allow decision can observe the response body onResponse reports a response's status and headers, and a request's body is already readable through filterRequest. The response body had no such hook, so a consumer could not see what came back: for an agent's model traffic, that is where the tool calls are. onResponseChunk gets a copy of each chunk as the upstream sent it, and onResponseEnd is called once and says whether the whole body arrived. Both are observers, like onResponse: they cannot change or hold the response, and their failures are logged and ignored. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- src/sandbox/http-proxy.ts | 13 +- src/sandbox/request-filter.ts | 73 +++++++++- src/sandbox/sandbox-config.ts | 3 +- src/sandbox/tls-terminate-proxy.ts | 5 +- test/sandbox/request-rewriting.test.ts | 181 +++++++++++++++++++++++++ 6 files changed, 270 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 61714c9a1..a30394f75 100644 --- a/README.md +++ b/README.md @@ -493,7 +493,7 @@ These two are read at each CONNECT, so `updateConfig` changes them for the tunne } ``` -**Request rewriting** (with `network.filterRequest`, set by library consumers): an allow decision may also carry header edits (`removeHeaders`, `setHeaders`) and an `onResponse` observer, and a deny decision may choose its status (403 by default). These options are read when the proxy starts; `updateConfig` does not change them. +**Request rewriting** (with `network.filterRequest`, set by library consumers): an allow decision may also carry header edits (`removeHeaders`, `setHeaders`) and `onResponse`, `onResponseChunk` and `onResponseEnd` observers, and a deny decision may choose its status (403 by default). `onResponseChunk` gets a copy of each body chunk as the upstream sent it (a compressed body stays compressed), and `onResponseEnd` is called once and says whether the whole body arrived. Neither can change or hold the response, and what a caller keeps is its own to bound. These options are read when the proxy starts; `updateConfig` does not change them. - `network.allowPlaintextHeaderSet` - Let an allow decision set headers on plain-HTTP requests. Off by default: a header set there would travel in cleartext, so while it is off an allow decision that carries `setHeaders` is refused with 403, and nothing is forwarded, for every request the proxy receives in cleartext (an absolute `https://` URI included). An allow without `setHeaders` is unaffected and removals always apply. The callback's second argument reports `scheme: 'http'` for these requests. - `network.stripResponseHeaders` - Response headers never passed back to the sandboxed client, matched case-insensitively with `-`, `_` and `.` folded (e.g. `["set-cookie"]`). diff --git a/src/sandbox/http-proxy.ts b/src/sandbox/http-proxy.ts index d450b1065..0ba1a9cc3 100644 --- a/src/sandbox/http-proxy.ts +++ b/src/sandbox/http-proxy.ts @@ -24,6 +24,7 @@ import { type DenyMark, applyHeaderEdits, notifyResponse, + observeResponseBody, removeHeadersFolded, hostHeaderValues, hostMismatch, @@ -1286,7 +1287,9 @@ export function createHttpProxyServer(options: HttpProxyServerOptions): Server { if (options.stripResponseHeaders) { removeHeadersFolded(outHeaders, options.stripResponseHeaders) } - if (relayResponseHead(res, proxyRes, outHeaders)) proxyRes.pipe(res) + const relayed = relayResponseHead(res, proxyRes, outHeaders) + observeResponseBody(decision, proxyRes, relayed) + if (relayed) proxyRes.pipe(res) }, ) } else if (parentUrl) { @@ -1321,7 +1324,9 @@ export function createHttpProxyServer(options: HttpProxyServerOptions): Server { if (options.stripResponseHeaders) { removeHeadersFolded(outHeaders, options.stripResponseHeaders) } - if (relayResponseHead(res, proxyRes, outHeaders)) proxyRes.pipe(res) + const relayed = relayResponseHead(res, proxyRes, outHeaders) + observeResponseBody(decision, proxyRes, relayed) + if (relayed) proxyRes.pipe(res) }, ) } else { @@ -1366,7 +1371,9 @@ export function createHttpProxyServer(options: HttpProxyServerOptions): Server { if (options.stripResponseHeaders) { removeHeadersFolded(outHeaders, options.stripResponseHeaders) } - if (relayResponseHead(res, proxyRes, outHeaders)) proxyRes.pipe(res) + const relayed = relayResponseHead(res, proxyRes, outHeaders) + observeResponseBody(decision, proxyRes, relayed) + if (relayed) proxyRes.pipe(res) }, ) } diff --git a/src/sandbox/request-filter.ts b/src/sandbox/request-filter.ts index baa0e4183..0610cfbec 100644 --- a/src/sandbox/request-filter.ts +++ b/src/sandbox/request-filter.ts @@ -70,6 +70,24 @@ export type RequestDecision = { * response is not held for it. */ onResponse?: (status: number, headers: IncomingHttpHeaders) => void + /** + * Allow only: called with each chunk of the upstream response body as it + * is relayed, in order, with the bytes the upstream sent: a compressed + * body stays compressed, as the `content-encoding` given to `onResponse` + * says. It gets a copy, so it observes and cannot change the response. The + * response is not held for it, and nothing bounds what it keeps: a caller + * that buffers sets its own limit. A throw or a rejection is logged and + * ignored, and later chunks still arrive. + */ + onResponseChunk?: (chunk: Buffer) => void + /** + * Allow only: called once when the upstream response is over. `complete` + * is true when the whole body arrived, and false when it did not: the + * upstream or the client went away part-way, or the proxy replaced an + * unrelayable response with its own 502. A throw or a rejection is logged + * and ignored. + */ + onResponseEnd?: (complete: boolean) => void } /** @@ -454,6 +472,15 @@ function malformedDecisionReason(decision: unknown): string | undefined { if (d.onResponse !== undefined && typeof d.onResponse !== 'function') { return 'onResponse is not a function' } + if ( + d.onResponseChunk !== undefined && + typeof d.onResponseChunk !== 'function' + ) { + return 'onResponseChunk is not a function' + } + if (d.onResponseEnd !== undefined && typeof d.onResponseEnd !== 'function') { + return 'onResponseEnd is not a function' + } const removals = d.removeHeaders if (removals !== undefined) { if (!Array.isArray(removals)) return 'removeHeaders is not an array' @@ -846,6 +873,50 @@ export function notifyResponse( } } +/** + * Report an upstream response's body to an allow decision's observers, + * containing their failures. `relayed` is what relayResponseHead answered: a + * response it replaced has no body to report. Call it before the body is + * piped, and in the same tick: a data listener starts the stream flowing, and + * pipe() then paces both. + */ +export function observeResponseBody( + decision: RequestDecision | undefined, + upstream: IncomingMessage, + relayed: boolean, +): void { + const onChunk = decision?.onResponseChunk + const onEnd = decision?.onResponseEnd + if (!onChunk && !onEnd) return + const contained = (call: () => unknown): void => { + try { + const returned = call() + if (isThenable(returned)) returned.then(undefined, logResponseFailure) + } catch (err) { + logResponseFailure(err) + } + } + let ended = false + const end = (complete: boolean): void => { + if (ended) return + ended = true + if (onEnd) contained(() => onEnd(complete)) + } + if (!relayed) return end(false) + if (onChunk) { + upstream.on('data', (chunk: Buffer) => { + if (ended) return + const copy = Buffer.from(chunk) + contained(() => onChunk(copy)) + }) + } + // 'end' alone does not say the body was whole, and a cut body does not reach + // 'close' on every runtime. + upstream.once('end', () => end(upstream.complete)) + upstream.once('error', () => end(false)) + upstream.once('close', () => end(false)) +} + function isThenable(value: unknown): value is PromiseLike { return ( typeof value === 'object' && @@ -856,7 +927,7 @@ function isThenable(value: unknown): value is PromiseLike { function logResponseFailure(err: unknown): void { logForDebugging( - `[request-filter] onResponse failed: ${(err as Error)?.message ?? String(err)}`, + `[request-filter] response observer failed: ${(err as Error)?.message ?? String(err)}`, { level: 'error' }, ) } diff --git a/src/sandbox/sandbox-config.ts b/src/sandbox/sandbox-config.ts index 3c4f96974..61c4521b7 100644 --- a/src/sandbox/sandbox-config.ts +++ b/src/sandbox/sandbox-config.ts @@ -787,7 +787,8 @@ export const NetworkConfigSchema = z.object({ 'Per-request filter callback. Receives the parsed HTTP request ' + '(web-standard Request) and request details, and returns {action, ' + 'reason?, status?} or, for an allow, header edits (removeHeaders, ' + - 'setHeaders) and an onResponse observer. Denied requests get the ' + + 'setHeaders) and onResponse, onResponseChunk and onResponseEnd ' + + 'observers. Denied requests get the ' + 'status (403 by default) with the reason. If the callback throws, the ' + 'request is denied. Applies to plain HTTP through the proxy and, ' + 'when tlsTerminate is configured, to terminated HTTPS. SRT does not ' + diff --git a/src/sandbox/tls-terminate-proxy.ts b/src/sandbox/tls-terminate-proxy.ts index 3cdebc5c5..2d5a6a240 100644 --- a/src/sandbox/tls-terminate-proxy.ts +++ b/src/sandbox/tls-terminate-proxy.ts @@ -44,6 +44,7 @@ import { type DenyMark, applyHeaderEdits, notifyResponse, + observeResponseBody, removeHeadersFolded, hostHeaderValues, hostMismatch, @@ -1339,7 +1340,9 @@ async function forwardUpstream( if (target.stripResponseHeaders) { removeHeadersFolded(outHeaders, target.stripResponseHeaders) } - if (relayResponseHead(res, upRes, outHeaders)) relayResponse(upRes, res) + const relayed = relayResponseHead(res, upRes, outHeaders) + observeResponseBody(decision, upRes, relayed) + if (relayed) relayResponse(upRes, res) }, ) diff --git a/test/sandbox/request-rewriting.test.ts b/test/sandbox/request-rewriting.test.ts index f890ef73a..f17c0ebb2 100644 --- a/test/sandbox/request-rewriting.test.ts +++ b/test/sandbox/request-rewriting.test.ts @@ -951,6 +951,187 @@ describe('an allow decision can observe the upstream response', () => { ) }, 15_000) + test('the body observers see every byte and the end, on both paths', async () => { + const chunks: Buffer[] = [] + const ends: boolean[] = [] + await withProxy( + async () => ({ + action: 'allow', + onResponseChunk: chunk => { + chunks.push(chunk) + }, + onResponseEnd: complete => { + ends.push(complete) + }, + }), + {}, + async port => { + expect((await viaTls(port, '/observed')).body).toBe('upstream ok') + expect((await viaPlain(port, '/observed')).body).toBe('upstream ok') + }, + ) + expect(Buffer.concat(chunks).toString()).toBe('upstream okupstream ok') + expect(ends).toEqual([true, true]) + }, 15_000) + + test('onResponseEnd alone does not take the body from the client', async () => { + const ends: boolean[] = [] + await withProxy( + async () => ({ + action: 'allow', + onResponseEnd: complete => { + ends.push(complete) + }, + }), + {}, + async port => { + expect((await viaTls(port, '/observed')).body).toBe('upstream ok') + expect((await viaPlain(port, '/observed')).body).toBe('upstream ok') + }, + ) + expect(ends).toEqual([true, true]) + }, 15_000) + + test('a body observer gets a copy: changing it does not change what the client receives', async () => { + await withProxy( + async () => ({ + action: 'allow', + onResponseChunk: chunk => { + chunk.fill(0x58) + }, + }), + {}, + async port => { + expect((await viaTls(port, '/observed')).body).toBe('upstream ok') + expect((await viaPlain(port, '/observed')).body).toBe('upstream ok') + }, + ) + }, 15_000) + + test('body observers that throw or reject do not disturb the response', async () => { + const unhandled: unknown[] = [] + const onUnhandled = (reason: unknown): void => { + unhandled.push(reason) + } + process.on('unhandledRejection', onUnhandled) + try { + let calls = 0 + await withProxy( + async () => ({ + action: 'allow', + onResponseChunk: () => { + calls++ + throw new Error('chunk observer failed') + }, + onResponseEnd: async () => { + calls++ + throw new Error('end observer failed') + }, + }), + {}, + async port => { + expect((await viaTls(port, '/observed')).body).toBe('upstream ok') + expect((await viaPlain(port, '/observed')).body).toBe('upstream ok') + }, + ) + await new Promise(r => setTimeout(r, 50)) + expect(calls).toBeGreaterThanOrEqual(4) + expect(unhandled).toEqual([]) + } finally { + process.off('unhandledRejection', onUnhandled) + } + }, 15_000) + + test('a body cut off part-way ends incomplete, on both paths', async () => { + const cut = (req: IncomingMessage, res: ServerResponse): void => { + req.resume() + res.writeHead(200, { 'content-length': '100' }) + res.write('partial') + // Long enough for the head and the first bytes to reach the proxy. + setTimeout(() => res.destroy(), 100) + } + httpsUpstream.removeListener('request', upstreamHandler) + httpUpstream.removeListener('request', upstreamHandler) + httpsUpstream.on('request', cut) + httpUpstream.on('request', cut) + try { + const chunks: Buffer[] = [] + const ends: boolean[] = [] + await withProxy( + async () => ({ + action: 'allow', + onResponseChunk: chunk => { + chunks.push(chunk) + }, + onResponseEnd: complete => { + ends.push(complete) + }, + }), + {}, + async port => { + await viaTls(port, '/observed').catch(() => undefined) + await viaPlain(port, '/observed').catch(() => undefined) + }, + ) + await new Promise(r => setTimeout(r, 50)) + expect(ends).toEqual([false, false]) + expect(Buffer.concat(chunks).toString()).toBe('partialpartial') + } finally { + httpsUpstream.removeListener('request', cut) + httpUpstream.removeListener('request', cut) + httpsUpstream.on('request', upstreamHandler) + httpUpstream.on('request', upstreamHandler) + } + }, 15_000) + + test('a response the proxy cannot relay ends incomplete, with no bytes', async () => { + const bad = (req: IncomingMessage, res: ServerResponse): void => { + req.resume() + // A status outside 100-999 cannot be written: the client gets a 502. + res.socket?.end('HTTP/1.1 099 x\r\nContent-Length: 5\r\n\r\nhello') + } + httpUpstream.removeListener('request', upstreamHandler) + httpUpstream.on('request', bad) + try { + const chunks: Buffer[] = [] + const ends: boolean[] = [] + await withProxy( + async () => ({ + action: 'allow', + onResponseChunk: chunk => { + chunks.push(chunk) + }, + onResponseEnd: complete => { + ends.push(complete) + }, + }), + {}, + async port => { + expect((await viaPlain(port, '/observed')).status).toBe(502) + }, + ) + await new Promise(r => setTimeout(r, 50)) + expect(ends).toEqual([false]) + expect(chunks).toEqual([]) + } finally { + httpUpstream.removeListener('request', bad) + httpUpstream.on('request', upstreamHandler) + } + }, 15_000) + + test('a body observer that is not a function denies', async () => { + for (const field of ['onResponseChunk', 'onResponseEnd']) { + await withProxy( + async () => + ({ action: 'allow', [field]: 'nope' }) as unknown as RequestDecision, + {}, + async port => { + expect((await viaPlain(port, '/observed')).status).toBe(403) + }, + ) + } + }, 15_000) + test('an onResponse that is not a function denies', async () => { await withProxy( async () =>