fix(anthropic): bound streaming 200 success-body SSE reads at 16 MiB

wangmiao0668000666 · wangmiao0668000666 · commit 7a45023b33ae · 2026-06-27T23:48:57.000+08:00
Apply createSseByteGuard to src/agents/anthropic-transport-stream.ts so the Anthropic Messages transport parser cannot be exhausted by a hostile or malfunctioning Anthropic-compatible endpoint that streams an unbounded SSE body. The 16 MiB cap matches the non-streaming readProviderJsonResponse cap already used for non-streaming Anthropic success bodies. What changed: - New helper src/agents/streaming-byte-guard.ts (78 lines, internal core): createSseByteGuard wraps a ReadableStreamDefaultReader<Uint8Array>, tracks accumulated bytes across the existing chunk loop, cancels the underlying reader on overflow, and throws a canonical overflow error. Internal for now; if extensions need it, promote to a plugin-SDK subpath in a separate PR with full SDK metadata sync. - src/agents/anthropic-transport-stream.ts: wrap body.getReader() in createSseByteGuard before yielding to the existing parseAnthropicSseBody loop. The readAnthropicSseChunk type was widened to accept any { read, cancel } so the same wrapper drives the guarded reader. - Inline test added to anthropic-transport-stream.test.ts: hostile 1 MiB pull stream, asserts canonical overflow message in result.errorMessage, cancel(reason) was an Error instance, pullCount bounded to 17-20. No SDK surface change. No repro script committed (proof in this body). The companion error-body path is already bounded (readAnthropicMessages ErrorBodySnippet, 8 KiB + 10s idle timeout, #95108); this PR closes the symmetric success-body gap.
diff --git a/src/agents/anthropic-transport-stream.test.ts b/src/agents/anthropic-transport-stream.test.ts
@@ -2903,4 +2903,47 @@ describe("anthropic transport stream", () => {
     // stream (a timing artefact of synchronous mock SSE delivery).
     expect(eventTypes).not.toContain("start");
   });
+
+  it("bounds streamed Anthropic success bodies without content-length", async () => {
+    // 1 MiB chunks; cap is 16 MiB so the bounded reader should cancel well
+    // before draining the full 32 MiB advertised body.
+    const CHUNK = 1024 * 1024;
+    const TOTAL = 32;
+    let pullCount = 0;
+    let cancelReason: unknown;
+    const overflowing = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        pullCount += 1;
+        if (pullCount > TOTAL) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(new Uint8Array(CHUNK));
+      },
+      cancel(reason) {
+        cancelReason = reason;
+      },
+    });
+    guardedFetchMock.mockResolvedValueOnce(
+      new Response(overflowing, {
+        status: 200,
+        headers: { "content-type": "text/event-stream" },
+      }),
+    );
+
+    const result = await runTransportStream(
+      makeAnthropicTransportModel(),
+      { messages: [{ role: "user", content: "hi" }] } as AnthropicStreamContext,
+      { apiKey: "sk-ant-api" } as AnthropicStreamOptions,
+    );
+
+    expect(result.stopReason).toBe("error");
+    expect(result.errorMessage).toMatch(
+      /Anthropic Messages success body exceeded 16777216 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/anthropic-transport-stream.ts b/src/agents/anthropic-transport-stream.ts
@@ -50,6 +50,7 @@ import {
 } from "./anthropic-tool-projection.js";
 import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./copilot-dynamic-headers.js";
 import { parseJsonObjectPreservingUnsafeIntegers } from "./json-unsafe-integers.js";
+import { createSseByteGuard } from "./streaming-byte-guard.js";
 import { resolveProviderEndpoint } from "./provider-attribution.js";
 import { buildGuardedModelFetch } from "./provider-transport-fetch.js";
 import type { StreamFn } from "./runtime/index.js";
@@ -69,6 +70,7 @@ const CLAUDE_CODE_VERSION = "2.1.75";
 const ANTHROPIC_MESSAGES_ERROR_BODY_MAX_BYTES = 8 * 1024;
 const ANTHROPIC_MESSAGES_ERROR_BODY_MAX_CHARS = 400;
 const ANTHROPIC_MESSAGES_ERROR_BODY_READ_IDLE_TIMEOUT_MS = 10_000;
+const ANTHROPIC_MESSAGES_SUCCESS_BODY_MAX_BYTES = 16 * 1024 * 1024;
 const CLAUDE_CODE_TOOLS = [
   "Read",
   "Write",
@@ -613,7 +615,7 @@ function createAbortError(signal: AbortSignal): Error {
 }
 
 function readAnthropicSseChunk(
-  reader: ReadableStreamDefaultReader<Uint8Array>,
+  reader: { read: () => Promise<ReadableStreamReadResult<Uint8Array>>; cancel: (reason?: unknown) => Promise<void> },
   signal?: AbortSignal,
 ): Promise<ReadableStreamReadResult<Uint8Array>> {
   if (!signal) {
@@ -675,12 +677,23 @@ async function* parseAnthropicSseBody(
   signal?: AbortSignal,
 ): AsyncIterable<Record<string, unknown>> {
   const reader = body.getReader();
+  // Cap the streaming 200 success-body read at 16 MiB, mirroring the
+  // non-streaming `readProviderJsonResponse` cap so a hostile or
+  // malfunctioning Anthropic-compatible endpoint cannot exhaust memory by
+  // streaming an unbounded SSE body.
+  const guard = createSseByteGuard(reader, {
+    maxBytes: ANTHROPIC_MESSAGES_SUCCESS_BODY_MAX_BYTES,
+    onOverflow: ({ size, maxBytes }) =>
+      new Error(
+        `Anthropic Messages success body exceeded ${maxBytes} bytes (received ${size})`,
+      ),
+  });
   const decoder = new TextDecoder();
   let buffer = "";
   let completed = false;
   try {
     while (true) {
-      const { done, value } = await readAnthropicSseChunk(reader, signal);
+      const { done, value } = await readAnthropicSseChunk(guard, signal);
       if (done) {
         completed = true;
         break;
