proxy: an allow decision can observe the response body

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"]`).

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)
           },
         )
       }

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<unknown> {
   return (
     typeof value === 'object' &&
@@ -856,7 +927,7 @@ function isThenable(value: unknown): value is PromiseLike<unknown> {
 
 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' },
   )
 }

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 ' +

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)
     },
   )
 

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 () =>