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

wangmiao0668000666 · wangmiao0668000666 · commit 7c3dea35e1b2 · 2026-06-27T23:48:34.000+08:00
Apply createSseByteGuard to src/llm/providers/anthropic.ts (legacy provider iterateSseMessages) so the Anthropic Messages provider SSE 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. What changed: - src/llm/providers/anthropic.ts: wrap body.getReader() in createSseByteGuard before the existing iterateSseMessages loop. The function is also re-exported as iterateSseMessagesForTest for direct test access. - Inline test added to anthropic.test.ts: hostile 1 MiB pull stream, asserts canonical overflow message, 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 second of the planned per-surface rescue series for the previously closed PR #96632, after #96701. 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 legacy-provider success-body gap.
diff --git a/src/llm/providers/anthropic.test.ts b/src/llm/providers/anthropic.test.ts
@@ -21,7 +21,7 @@ vi.mock("@anthropic-ai/sdk", () => ({
   },
 }));
 
-import { streamAnthropic, streamSimpleAnthropic } from "./anthropic.js";
+import { streamAnthropic, streamSimpleAnthropic, iterateSseMessagesForTest } from "./anthropic.js";
 
 function createSseResponse(events: Record<string, unknown>[] = []): Response {
   const body = events
@@ -1306,4 +1306,41 @@ describe("Anthropic provider", () => {
       },
     ]);
   });
+
+  it("bounds streamed Anthropic provider 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;
+    const overflowing = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        pullCount += 1;
+        if (pullCount > TOTAL) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(new Uint8Array(CHUNK));
+      },
+      cancel(reason) {
+        cancelReason = reason;
+      },
+    });
+    let caught: Error | null = null;
+    try {
+      for await (const event of iterateSseMessagesForTest(overflowing)) {
+        expect(event).toBeDefined();
+      }
+    } catch (err) {
+      caught = err as Error;
+    }
+    expect(caught?.message).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/llm/providers/anthropic.ts b/src/llm/providers/anthropic.ts
@@ -32,6 +32,7 @@ import {
   usesClaudeFable5MessagesContract,
 } from "../../shared/anthropic-model-contract.js";
 import { applyAnthropicRefusal } from "../../shared/anthropic-refusal.js";
+import { createSseByteGuard } from "../../agents/streaming-byte-guard.js";
 import { createDeferredEventBuffer } from "../../shared/deferred-event-buffer.js";
 import { notifyLlmRequestActivity } from "../../shared/llm-request-activity.js";
 import { getEnvApiKey } from "../env-api-keys.js";
@@ -72,6 +73,7 @@ import { adjustMaxTokensForThinking, buildBaseOptions } from "./simple-options.j
 import { transformMessages } from "./transform-messages.js";
 
 const ANTHROPIC_CACHE_CONTROL_LIMIT = 4;
+const ANTHROPIC_MESSAGES_SUCCESS_BODY_MAX_BYTES = 16 * 1024 * 1024;
 
 function getCacheControl(
   model: Model<"anthropic-messages">,
@@ -347,6 +349,17 @@ async function* iterateSseMessages(
   signal?: AbortSignal,
 ): AsyncGenerator<ServerSentEvent> {
   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();
   const state: SseDecoderState = { event: null, data: [], raw: [] };
   let buffer = "";
@@ -357,7 +370,7 @@ async function* iterateSseMessages(
         throw new Error("Request was aborted");
       }
 
-      const { value, done } = await reader.read();
+      const { value, done } = await guard.read();
       if (done) {
         break;
       }
@@ -1495,3 +1508,10 @@ function mapStopReason(reason: string): StopReason {
       throw new Error(`Unhandled stop reason: ${reason}`);
   }
 }
+
+/**
+ * Internal — exported under a separate name so production callers go through
+ * `streamAnthropic`. Tests can drive it directly to exercise the bounded-reader
+ * behaviour without the surrounding SDK client wrapper.
+ */
+export const iterateSseMessagesForTest = iterateSseMessages;
