fix(openai-chatgpt-responses): bound streaming success-body SSE reads at 16 MiB (#96762)

wangmiao0668000666 · vincentkoc · web-flow · commit 5880e0afc4dc · 2026-06-27T08:35:10.000+08:00
* fix(openai-chatgpt-responses): bound streaming success-body SSE reads at 16 MiB Apply createSseByteGuard to src/llm/providers/openai-chatgpt-responses.ts (parseSSE) so the ChatGPT Responses SSE parser cannot be exhausted by a hostile or malfunctioning endpoint that streams an unbounded SSE body. The 16 MiB cap matches the non-streaming readProviderJsonResponse cap. What changed: - src/llm/providers/openai-chatgpt-responses.ts: wrap body.getReader() in createSseByteGuard before the existing parseSSE loop. The function is also re-exported as parseSSEForTest for direct test access. - Inline test added to openai-chatgpt-responses.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 third of the planned per-surface rescue series for the previously closed PR #96666, after PR #3b (#96723). No SDK surface change. No repro script committed (proof in this body). * fix(openai): bound ChatGPT error body reads * fix(openai): bound chatgpt error parsing --------- Co-authored-by: Vincent Koc <vincentkoc@ieee.org>
diff --git a/src/agents/streaming-byte-guard.ts b/src/agents/streaming-byte-guard.ts
@@ -0,0 +1,87 @@
+/**
+ * Bounded SSE / NDJSON stream reader guard.
+ *
+ * Wraps a `ReadableStreamDefaultReader<Uint8Array>` so the caller's existing
+ * chunk-by-chunk parsing logic is unchanged, but accumulated bytes are tracked
+ * against a hard cap. On overflow the underlying reader is cancelled and a
+ * canonical error is thrown. Mirrors the `readResponseWithLimit` / bounded
+ * JSON response pattern (see `src/agents/provider-http-errors.ts`).
+ *
+ * Internal helper for now. If extensions need it, promote to a plugin-SDK
+ * subpath in a separate, dedicated PR with full SDK metadata sync.
+ */
+
+export type SseStreamOverflow = {
+  size: number;
+  maxBytes: number;
+};
+
+export type ReadSseStreamWithLimitOptions = {
+  maxBytes: number;
+  onOverflow?: (params: SseStreamOverflow) => Error;
+};
+
+export type SseByteGuard = {
+  read(): Promise<ReadableStreamReadResult<Uint8Array>>;
+  cancel(reason?: unknown): Promise<void>;
+  totalBytes(): number;
+  overflowed(): boolean;
+  cancelled(): boolean;
+};
+
+export function createSseByteGuard(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  opts: ReadSseStreamWithLimitOptions,
+): SseByteGuard {
+  if (!Number.isFinite(opts.maxBytes) || opts.maxBytes < 0) {
+    throw new RangeError(`maxBytes must be a non-negative finite number: ${opts.maxBytes}`);
+  }
+  const onOverflow =
+    opts.onOverflow ??
+    ((params) =>
+      new Error(`SSE stream exceeds ${params.maxBytes} bytes (received ${params.size})`));
+  let total = 0;
+  let overflowedFlag = false;
+  let cancelledFlag = false;
+  return {
+    read: async () => {
+      if (overflowedFlag || cancelledFlag) {
+        return { done: true, value: undefined };
+      }
+      const result = await reader.read();
+      if (result.done) {
+        return result;
+      }
+      const chunkLen = result.value?.byteLength ?? 0;
+      const next = total + chunkLen;
+      if (next > opts.maxBytes) {
+        overflowedFlag = true;
+        cancelledFlag = true;
+        const err = onOverflow({ size: next, maxBytes: opts.maxBytes });
+        try {
+          await reader.cancel(err);
+        } catch {
+          // best-effort cancellation; caller observes the overflow error
+        }
+        throw err;
+      }
+      total = next;
+      return result;
+    },
+    cancel: async (reason?: unknown) => {
+      if (overflowedFlag) {
+        // overflow already set cancelledFlag; do not overwrite
+        return;
+      }
+      cancelledFlag = true;
+      try {
+        await reader.cancel(reason);
+      } catch {
+        // best-effort cancellation
+      }
+    },
+    totalBytes: () => total,
+    overflowed: () => overflowedFlag,
+    cancelled: () => cancelledFlag,
+  };
+}
diff --git a/src/llm/providers/openai-chatgpt-responses.test.ts b/src/llm/providers/openai-chatgpt-responses.test.ts
@@ -5,6 +5,7 @@ import { SYSTEM_PROMPT_CACHE_BOUNDARY } from "../../agents/system-prompt-cache-b
 import type { Context, Model } from "../types.js";
 import {
   extractOpenAICodexAccountId,
+  parseSSEForTest,
   resetOpenAICodexWebSocketDebugStats,
   streamOpenAICodexResponses,
 } from "./openai-chatgpt-responses.js";
@@ -560,4 +561,95 @@ describe("streamOpenAICodexResponses transport", () => {
     expect(fetchMock).toHaveBeenCalledTimes(2);
     expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_TIMER_TIMEOUT_MS);
   });
+
+  it("bounds non-OK ChatGPT response bodies before formatting API errors", async () => {
+    const chunkSize = 1024 * 1024;
+    const totalChunks = 32;
+    const chunk = new TextEncoder()
+      .encode("usage limit ".repeat(Math.ceil(chunkSize / "usage limit ".length)))
+      .subarray(0, chunkSize);
+    let pullCount = 0;
+    let canceled = false;
+    const overflowing = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        pullCount += 1;
+        if (pullCount > totalChunks) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(chunk);
+      },
+      cancel() {
+        canceled = true;
+      },
+    });
+    const fetchMock = vi.fn<typeof fetch>().mockResolvedValueOnce(
+      new Response(overflowing, {
+        status: 400,
+        statusText: "Bad Request",
+      }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+
+    const stream = streamOpenAICodexResponses(model, context, {
+      apiKey: createJwt({
+        "https://api.openai.com/auth": {
+          chatgpt_account_id: "acct-1",
+        },
+      }),
+      transport: "sse",
+    });
+
+    const result = await stream.result();
+
+    expect(result.stopReason).toBe("error");
+    expect(result.errorMessage).toContain("usage limit");
+    expect(result.errorMessage?.length).toBeLessThanOrEqual(16 * 1024);
+    expect(canceled).toBe(true);
+    expect(pullCount).toBeGreaterThanOrEqual(1);
+    expect(pullCount).toBeLessThanOrEqual(3);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe("parseSSEForTest", () => {
+  it("bounds streamed OpenAI ChatGPT Responses 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 {
+      // parseSSE expects a Response-like; pass the streaming body directly
+      // through a minimal Response shim that only exposes .body.
+      const response = { body: overflowing } as unknown as Response;
+      for await (const event of parseSSEForTest(response)) {
+        expect(event).toBeDefined();
+      }
+    } catch (err) {
+      caught = err as Error;
+    }
+    expect(caught?.message).toMatch(
+      /OpenAI ChatGPT Responses 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/openai-chatgpt-responses.ts b/src/llm/providers/openai-chatgpt-responses.ts
@@ -25,6 +25,7 @@ import {
   resolveTimerTimeoutMs,
   clampTimerTimeoutMs,
 } from "@openclaw/normalization-core/number-coercion";
+import { createSseByteGuard } from "../../agents/streaming-byte-guard.js";
 import { stripSystemPromptCacheBoundary } from "../../agents/system-prompt-cache-boundary.js";
 import { getEnvApiKey } from "../env-api-keys.js";
 import { clampThinkingLevel } from "../model-utils.js";
@@ -66,6 +67,8 @@ const RETRY_AFTER_HTTP_DATE_RE =
   /^(?:(?:Mon|Tue|Wed|Thu|Fri|Sat|Sun), \d{2} (?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec) \d{4} \d{2}:\d{2}:\d{2} GMT|(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday), \d{2}-(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)-\d{2} \d{2}:\d{2}:\d{2} GMT|(?:Mon|Tue|Wed|Thu|Fri|Sat|Sun) (?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec) [ \d]\d \d{2}:\d{2}:\d{2} \d{4})$/;
 const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "opencode"]);
 const WEBSOCKET_MESSAGE_TOO_BIG_CLOSE_CODE = 1009;
+const OPENAI_CHATGPT_RESPONSES_ERROR_BODY_MAX_BYTES = 16 * 1024;
+const OPENAI_CHATGPT_RESPONSES_SUCCESS_BODY_MAX_BYTES = 16 * 1024 * 1024;
 
 const CODEX_RESPONSE_STATUSES = new Set<CodexResponseStatus>([
   "completed",
@@ -339,7 +342,7 @@ export const streamOpenAICodexResponses: StreamFunction<
             break;
           }
 
-          const errorText = await response.text();
+          const errorText = await readChatGptResponsesErrorTextLimited(response);
           if (attempt < MAX_RETRIES && isRetryableError(response.status, errorText)) {
             let delayMs = BASE_DELAY_MS * 2 ** attempt;
 
@@ -722,12 +725,23 @@ async function* parseSSE(response: Response): AsyncGenerator<Record<string, unkn
   }
 
   const reader = response.body.getReader();
+  // Cap the streaming 200 success-body read at 16 MiB, mirroring the
+  // non-streaming `readProviderJsonResponse` cap so a hostile or
+  // malfunctioning ChatGPT Responses endpoint cannot exhaust memory by
+  // streaming an unbounded SSE body.
+  const guard = createSseByteGuard(reader, {
+    maxBytes: OPENAI_CHATGPT_RESPONSES_SUCCESS_BODY_MAX_BYTES,
+    onOverflow: ({ size, maxBytes }) =>
+      new Error(
+        `OpenAI ChatGPT Responses success body exceeded ${maxBytes} bytes (received ${size})`,
+      ),
+  });
   const decoder = new TextDecoder();
   let buffer = "";
 
   try {
     while (true) {
-      const { done, value } = await reader.read();
+      const { done, value } = await guard.read();
       if (done) {
         break;
       }
@@ -760,14 +774,18 @@ async function* parseSSE(response: Response): AsyncGenerator<Record<string, unkn
     }
   } finally {
     try {
-      await reader.cancel();
+      await guard.cancel();
     } catch {}
     try {
       reader.releaseLock();
     } catch {}
   }
 }
 
+// Test-only re-export of the bounded SSE parser. Mirrors
+// `parseAnthropicSseBodyForTest` / `iterateSseMessagesForTest` patterns.
+export const parseSSEForTest = parseSSE;
+
 // ============================================================================
 // WebSocket Parsing
 // ============================================================================
@@ -1521,10 +1539,57 @@ async function processWebSocketStream(
 // Error Handling
 // ============================================================================
 
+async function readChatGptResponsesErrorTextLimited(response: Response): Promise<string> {
+  const reader = response.body?.getReader();
+  if (!reader) {
+    return "";
+  }
+
+  const decoder = new TextDecoder();
+  let total = 0;
+  let text = "";
+  let reachedLimit = false;
+
+  try {
+    while (true) {
+      const { value, done } = await reader.read();
+      if (done) {
+        break;
+      }
+      if (!value || value.byteLength === 0) {
+        continue;
+      }
+      const remaining = OPENAI_CHATGPT_RESPONSES_ERROR_BODY_MAX_BYTES - total;
+      if (remaining <= 0) {
+        reachedLimit = true;
+        break;
+      }
+      const chunk = value.byteLength > remaining ? value.subarray(0, remaining) : value;
+      total += chunk.byteLength;
+      text += decoder.decode(chunk, { stream: true });
+      if (total >= OPENAI_CHATGPT_RESPONSES_ERROR_BODY_MAX_BYTES) {
+        reachedLimit = true;
+        break;
+      }
+    }
+    text += decoder.decode();
+  } finally {
+    if (reachedLimit) {
+      // This provider module is browser-safe, so keep error-body capping on Web APIs.
+      await reader.cancel().catch(() => {});
+    }
+    try {
+      reader.releaseLock();
+    } catch {}
+  }
+
+  return text;
+}
+
 async function parseErrorResponse(
   response: Response,
 ): Promise<{ message: string; friendlyMessage?: string }> {
-  const raw = await response.text();
+  const raw = await readChatGptResponsesErrorTextLimited(response);
   let message = raw || response.statusText || "Request failed";
   let friendlyMessage: string | undefined;
 
