fix(runtime/proxy): bound streaming success-body SSE reads at 16 MiB

wangmiao0668000666 · wangmiao0668000666 · commit 87d72e2e20ee · 2026-06-27T23:42:34.000+08:00
Apply createSseByteGuard to src/agents/runtime/proxy.ts (streamProxy) so the runtime proxy SSE parser cannot be exhausted by a hostile or malfunctioning proxy endpoint that streams an unbounded SSE body. The 16 MiB cap matches the non-streaming readProviderJsonResponse cap. What changed: - src/agents/runtime/proxy.ts: wrap response.body.getReader() in createSseByteGuard before the existing streamProxy SSE read loop. The abort handler and finally block use guard.cancel() instead of reader.cancel(). - Inline test added to proxy.test.ts: hostile 1 MiB pull stream through streamProxy, asserts errorMessage matches overflow pattern, cancel(reason) was an Error instance, pullCount bounded to 17-20. Reuses the createSseByteGuard helper from src/agents/streaming-byte-guard.ts (introduced in #96701 for the anthropic-transport-stream path; same helper, same 16 MiB cap, same overflow-error shape). This PR is the final per-surface rescue from the previously closed PR #96666, after PR #3c (#96762). No SDK surface change. No repro script committed (proof in this body).
diff --git a/src/agents/runtime/proxy.test.ts b/src/agents/runtime/proxy.test.ts
@@ -189,4 +189,47 @@ describe("streamProxy", () => {
       errorMessage: "Proxy stream ended before terminal event",
     });
   });
+
+  it("bounds streamed proxy success bodies without content-length", async () => {
+    // 1 MiB chunks; cap is 16 MiB so the bounded reader cancels well before
+    // draining the full 32 MiB advertised body.
+    const CHUNK = 1024 * 1024;
+    const TOTAL = 32;
+    let pullCount = 0;
+    let cancelReason: unknown;
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(
+        async () =>
+          new Response(
+            new ReadableStream({
+              pull(controller) {
+                pullCount += 1;
+                if (pullCount > TOTAL) {
+                  controller.close();
+                  return;
+                }
+                controller.enqueue(new Uint8Array(CHUNK));
+              },
+              cancel(reason) {
+                cancelReason = reason;
+              },
+            }),
+            { status: 200 },
+          ),
+      ),
+    );
+
+    const result = await streamProxy(model, context, {
+      authToken: "token",
+      proxyUrl: "https://proxy.example",
+    }).result();
+
+    expect(result.stopReason).toBe("error");
+    expect(result.errorMessage).toMatch(/Proxy stream body exceeded \d+ bytes/);
+    expect(cancelReason).toBeInstanceOf(Error);
+    // 16 MiB + a couple of overshoot pulls, well under 32.
+    expect(pullCount).toBeGreaterThanOrEqual(17);
+    expect(pullCount).toBeLessThanOrEqual(20);
+  });
 });
diff --git a/src/agents/runtime/proxy.ts b/src/agents/runtime/proxy.ts
@@ -3,6 +3,7 @@
  * The server manages auth and proxies requests to LLM providers.
  */
 
+import { createSseByteGuard } from "../../agents/streaming-byte-guard.js";
 // Internal import for JSON parsing utility
 import type {
   AssistantMessage,
@@ -124,6 +125,8 @@ function sanitizeProxyModel(model: Model): Model {
   return safeModel as Model;
 }
 
+const PROXY_STREAM_BODY_MAX_BYTES = 16 * 1024 * 1024;
+
 export function streamProxy(
   model: Model,
   context: Context,
@@ -152,9 +155,12 @@ export function streamProxy(
     };
 
     let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
+    let guard: ReturnType<typeof createSseByteGuard> | undefined;
 
     const abortHandler = () => {
-      if (reader) {
+      if (guard) {
+        guard.cancel("Request aborted by user").catch(() => {});
+      } else if (reader) {
         reader.cancel("Request aborted by user").catch(() => {});
       }
     };
@@ -192,6 +198,11 @@ export function streamProxy(
       }
 
       reader = response.body!.getReader();
+      guard = createSseByteGuard(reader, {
+        maxBytes: PROXY_STREAM_BODY_MAX_BYTES,
+        onOverflow: ({ size, maxBytes }) =>
+          new Error(`Proxy stream body exceeded ${maxBytes} bytes (received ${size})`),
+      });
       const decoder = new TextDecoder();
       let buffer = "";
       let terminalEventSeen = false;
@@ -214,7 +225,7 @@ export function streamProxy(
       };
 
       while (true) {
-        const { done, value } = await reader.read();
+        const { done, value } = await guard.read();
         if (done) {
           break;
         }
@@ -256,6 +267,9 @@ export function streamProxy(
       });
       stream.end();
     } finally {
+      try {
+        await guard?.cancel();
+      } catch {}
       try {
         reader?.releaseLock();
       } catch {}
