From feea2b714789aa4644f12f8eff3380d088c0e9e2 Mon Sep 17 00:00:00 2001 From: Ren Koya Date: Thu, 16 Jul 2026 15:43:18 +0900 Subject: [PATCH 1/4] fix(client): bound SSE event size to prevent client-side memory exhaustion parseSseStream accumulated both the line buffer and an event's joined data with no size limit. A2A clients stream from remote, potentially untrusted agent servers; a malicious or broken server could stream bytes that never form a complete line, or endless `data:` lines with no terminating blank line, growing an in-memory buffer without bound until the client process runs out of memory (CWE-400). Cap both accumulation points at a configurable maxEventSizeBytes (default 20 MiB, generous enough for legitimate base64 file parts) and throw when exceeded. The throw triggers the generator's teardown, whose finally now cancels the reader (#580), so the offending connection is also closed. Adds regression tests for the unterminated-line and unterminated-event cases plus a within-limit sanity check. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/sse_utils.ts | 35 ++++++++++++++++++++++++++++++++++- test/sse_utils.spec.ts | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/src/sse_utils.ts b/src/sse_utils.ts index 7ce0796c..32f46878 100644 --- a/src/sse_utils.ts +++ b/src/sse_utils.ts @@ -60,13 +60,34 @@ export function formatSSEErrorEvent(error: unknown): string { // SSE Event Parsing (Client-side) // ============================================================================ +/** + * Upper bound (in UTF-16 code units, a conservative over-approximation of + * byte length) on how large a single unterminated line or a single SSE event + * may grow before {@link parseSseStream} aborts. + * + * A2A clients stream from remote, potentially untrusted agent servers. + * Without this cap a malicious or broken server can stream bytes that never + * form a complete line — or `data:` lines whose blank-line terminator never + * arrives — growing an in-memory buffer without limit until the client + * process exhausts memory (CWE-400, uncontrolled resource consumption). The + * default is deliberately generous so that legitimate large JSON events + * (e.g. base64 file parts) still pass; callers handling larger payloads can + * raise it via the `maxEventSizeBytes` argument. + */ +export const DEFAULT_MAX_SSE_EVENT_SIZE_BYTES = 20 * 1024 * 1024; // 20 MiB + /** * Parses an SSE stream from a `Response`, yielding events as they arrive. * Expects well-formed SSE events with single-line JSON data, matching the * format produced by {@link formatSSEEvent} and {@link formatSSEErrorEvent}. + * + * @param maxEventSizeBytes - Aborts the stream if a single line or event + * exceeds this size, bounding memory against a hostile server. Defaults to + * {@link DEFAULT_MAX_SSE_EVENT_SIZE_BYTES}. */ export async function* parseSseStream( - response: Response + response: Response, + maxEventSizeBytes: number = DEFAULT_MAX_SSE_EVENT_SIZE_BYTES ): AsyncGenerator { if (!response.body) { throw new Error('SSE response body is undefined. Cannot read stream.'); @@ -108,8 +129,20 @@ export async function* parseSseStream( // so append instead of overwriting. const fieldValue = stripOptionalLeadingSpace(line.substring('data:'.length)); eventData = eventData === '' ? fieldValue : `${eventData}\n${fieldValue}`; + if (eventData.length > maxEventSizeBytes) { + throw new Error( + `SSE event data exceeded the maximum allowed size of ${maxEventSizeBytes} bytes.` + ); + } } } + + // A hostile/broken server can stream bytes that never form a complete + // line, leaving the residual partial line in `buffer` to grow without + // bound. Cap it here (the drained portion above never triggers this). + if (buffer.length > maxEventSizeBytes) { + throw new Error(`SSE line exceeded the maximum allowed size of ${maxEventSizeBytes} bytes.`); + } } // Yield any pending event at stream end. diff --git a/test/sse_utils.spec.ts b/test/sse_utils.spec.ts index 1bc19ed4..7926a8cf 100644 --- a/test/sse_utils.spec.ts +++ b/test/sse_utils.spec.ts @@ -335,4 +335,44 @@ describe('SSE Utils', () => { expect(JSON.parse(parsedEvents[1].data)).toEqual(errorEvent); }); }); + + // A2A clients stream from remote, potentially untrusted servers. A hostile + // server that never terminates a line — or an event — would grow an + // in-memory buffer without bound (CWE-400). parseSseStream caps both. + describe('parseSseStream size bound (DoS guard)', () => { + async function drain(response: Response, maxEventSizeBytes?: number): Promise { + const events: SseEvent[] = []; + for await (const event of parseSseStream(response, maxEventSizeBytes)) { + events.push(event); + } + return events; + } + + it('throws when a single line never terminates and exceeds the limit', async () => { + // No newline anywhere: the residual partial line grows past the cap. + const response = createMockResponse('data: ' + 'A'.repeat(1000)); + + await expect(drain(response, 100)).rejects.toThrow(/SSE line exceeded the maximum/); + }); + + it('throws when accumulated data lines exceed the limit before a blank line', async () => { + // Many consecutive `data:` lines with no terminating blank line: the + // joined event data grows past the cap. + let sse = ''; + for (let i = 0; i < 200; i++) sse += `data: ${'A'.repeat(20)}\n`; + const response = createMockResponse(sse); + + await expect(drain(response, 100)).rejects.toThrow(/SSE event data exceeded the maximum/); + }); + + it('parses a well-formed event that stays within the limit', async () => { + const event = { kind: 'message', text: 'hello' }; + const response = createMockResponse(formatSSEEvent(event)); + + const events = await drain(response, 1024); + + expect(events).toHaveLength(1); + expect(JSON.parse(events[0].data)).toEqual(event); + }); + }); }); From f16160c1f1e702d26a752f79f88df1f7e49e2ba1 Mon Sep 17 00:00:00 2001 From: Ren Koya Date: Thu, 16 Jul 2026 15:56:48 +0900 Subject: [PATCH 2/4] fix(client): also bound terminated SSE lines, not just unterminated ones The residual buffer check ran after the line-split loop, so a large but terminated line (e.g. a huge `:` comment or `event:` line ending in \n) was drained by the loop and slipped past the cap. Enforce the limit inside the loop on the line length before extracting it, and add a regression test that delivers such a line in a single chunk (the 2-byte mock chunking would otherwise trip the residual check first). Addresses gemini-code-assist review on #582. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/sse_utils.ts | 10 +++++++--- test/sse_utils.spec.ts | 11 +++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/src/sse_utils.ts b/src/sse_utils.ts index 32f46878..86a03c40 100644 --- a/src/sse_utils.ts +++ b/src/sse_utils.ts @@ -104,6 +104,11 @@ export async function* parseSseStream( let lineEndIndex: number; while ((lineEndIndex = buffer.indexOf('\n')) >= 0) { + if (lineEndIndex > maxEventSizeBytes) { + throw new Error( + `SSE line exceeded the maximum allowed size of ${maxEventSizeBytes} bytes.` + ); + } // Per the SSE spec lines may end with `\r\n`, `\r`, or `\n`. We // strip a trailing `\r` explicitly rather than calling `.trim()`, // which would also eat whitespace inside JSON-formatted `data:` @@ -137,9 +142,8 @@ export async function* parseSseStream( } } - // A hostile/broken server can stream bytes that never form a complete - // line, leaving the residual partial line in `buffer` to grow without - // bound. Cap it here (the drained portion above never triggers this). + // Same cap for a line that never terminates: the loop above never runs, + // so the residual buffer would otherwise grow without bound. if (buffer.length > maxEventSizeBytes) { throw new Error(`SSE line exceeded the maximum allowed size of ${maxEventSizeBytes} bytes.`); } diff --git a/test/sse_utils.spec.ts b/test/sse_utils.spec.ts index 7926a8cf..de6f3eba 100644 --- a/test/sse_utils.spec.ts +++ b/test/sse_utils.spec.ts @@ -355,6 +355,17 @@ describe('SSE Utils', () => { await expect(drain(response, 100)).rejects.toThrow(/SSE line exceeded the maximum/); }); + it('throws when an oversized line is terminated and arrives in one chunk', async () => { + // Delivered whole so the newline is present before the residual check — + // exercises the in-loop guard the residual check alone would miss. + const stream = createStream([new TextEncoder().encode(': ' + 'A'.repeat(1000) + '\n')]); + const response = new Response(stream, { + headers: { 'Content-Type': 'text/event-stream' }, + }); + + await expect(drain(response, 100)).rejects.toThrow(/SSE line exceeded the maximum/); + }); + it('throws when accumulated data lines exceed the limit before a blank line', async () => { // Many consecutive `data:` lines with no terminating blank line: the // joined event data grows past the cap. From 69418231d812a26884e10ad8f69bf270e9a0bc41 Mon Sep 17 00:00:00 2001 From: Ren Koya Date: Thu, 16 Jul 2026 16:09:29 +0900 Subject: [PATCH 3/4] docs(client): correct size-limit unit description String .length under-counts UTF-8 wire bytes for non-ASCII, so calling it a conservative over-approximation was backwards. State the actual unit and that memory stays bounded within a constant factor. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/sse_utils.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/sse_utils.ts b/src/sse_utils.ts index 86a03c40..980a7251 100644 --- a/src/sse_utils.ts +++ b/src/sse_utils.ts @@ -61,9 +61,11 @@ export function formatSSEErrorEvent(error: unknown): string { // ============================================================================ /** - * Upper bound (in UTF-16 code units, a conservative over-approximation of - * byte length) on how large a single unterminated line or a single SSE event - * may grow before {@link parseSseStream} aborts. + * Upper bound on how large a single unterminated line or a single SSE event + * may grow before {@link parseSseStream} aborts. Measured in UTF-16 code + * units (JS string length), which equals byte length for ASCII payloads such + * as JSON-encoded events; memory stays bounded within a small constant + * factor of this value either way. * * A2A clients stream from remote, potentially untrusted agent servers. * Without this cap a malicious or broken server can stream bytes that never From ee27bac0b05ac6bb95912e25111e233ddf1e0419 Mon Sep 17 00:00:00 2001 From: Ren Koya Date: Thu, 16 Jul 2026 17:47:10 +0900 Subject: [PATCH 4/4] fix(client): lower default SSE event size limit to 4 MiB Realistic A2A events (Message/Task JSON) are KB-scale and large files should be referenced via FileWithUri parts rather than inlined, so 4 MiB (matching gRPC's default max message size) leaves ample headroom. Also make the error actionable: point at maxEventSizeBytes and FileWithUri. Addresses review feedback from JakubWorek on #582. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/sse_utils.ts | 30 ++++++++++++++++++------------ 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/src/sse_utils.ts b/src/sse_utils.ts index 980a7251..95e9d96f 100644 --- a/src/sse_utils.ts +++ b/src/sse_utils.ts @@ -71,12 +71,15 @@ export function formatSSEErrorEvent(error: unknown): string { * Without this cap a malicious or broken server can stream bytes that never * form a complete line — or `data:` lines whose blank-line terminator never * arrives — growing an in-memory buffer without limit until the client - * process exhausts memory (CWE-400, uncontrolled resource consumption). The - * default is deliberately generous so that legitimate large JSON events - * (e.g. base64 file parts) still pass; callers handling larger payloads can - * raise it via the `maxEventSizeBytes` argument. + * process exhausts memory (CWE-400, uncontrolled resource consumption). + * + * Realistic A2A events (Message/Task JSON) are KB-scale, and large files + * should be referenced via `FileWithUri` parts rather than inlined, so the + * 4 MiB default (matching gRPC's default max message size) leaves ample + * headroom. Callers that must inline larger payloads can raise it via the + * `maxEventSizeBytes` argument. */ -export const DEFAULT_MAX_SSE_EVENT_SIZE_BYTES = 20 * 1024 * 1024; // 20 MiB +export const DEFAULT_MAX_SSE_EVENT_SIZE_BYTES = 4 * 1024 * 1024; // 4 MiB /** * Parses an SSE stream from a `Response`, yielding events as they arrive. @@ -107,9 +110,7 @@ export async function* parseSseStream( while ((lineEndIndex = buffer.indexOf('\n')) >= 0) { if (lineEndIndex > maxEventSizeBytes) { - throw new Error( - `SSE line exceeded the maximum allowed size of ${maxEventSizeBytes} bytes.` - ); + throw sseSizeError('SSE line', maxEventSizeBytes); } // Per the SSE spec lines may end with `\r\n`, `\r`, or `\n`. We // strip a trailing `\r` explicitly rather than calling `.trim()`, @@ -137,9 +138,7 @@ export async function* parseSseStream( const fieldValue = stripOptionalLeadingSpace(line.substring('data:'.length)); eventData = eventData === '' ? fieldValue : `${eventData}\n${fieldValue}`; if (eventData.length > maxEventSizeBytes) { - throw new Error( - `SSE event data exceeded the maximum allowed size of ${maxEventSizeBytes} bytes.` - ); + throw sseSizeError('SSE event data', maxEventSizeBytes); } } } @@ -147,7 +146,7 @@ export async function* parseSseStream( // Same cap for a line that never terminates: the loop above never runs, // so the residual buffer would otherwise grow without bound. if (buffer.length > maxEventSizeBytes) { - throw new Error(`SSE line exceeded the maximum allowed size of ${maxEventSizeBytes} bytes.`); + throw sseSizeError('SSE line', maxEventSizeBytes); } } @@ -157,6 +156,13 @@ export async function* parseSseStream( } } +function sseSizeError(what: string, maxEventSizeBytes: number): Error { + return new Error( + `${what} exceeded the maximum allowed size of ${maxEventSizeBytes} bytes. ` + + `Pass maxEventSizeBytes to raise the limit, or prefer FileWithUri parts for large payloads.` + ); +} + /** * Per the SSE spec, the optional single leading space after the field-name * colon is consumed by the parser; embedded and trailing whitespace are