diff --git a/src/sse_utils.ts b/src/sse_utils.ts index 7ce0796c..95e9d96f 100644 --- a/src/sse_utils.ts +++ b/src/sse_utils.ts @@ -60,13 +60,39 @@ export function formatSSEErrorEvent(error: unknown): string { // SSE Event Parsing (Client-side) // ============================================================================ +/** + * 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 + * 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). + * + * 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 = 4 * 1024 * 1024; // 4 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.'); @@ -83,6 +109,9 @@ export async function* parseSseStream( let lineEndIndex: number; while ((lineEndIndex = buffer.indexOf('\n')) >= 0) { + if (lineEndIndex > maxEventSizeBytes) { + 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()`, // which would also eat whitespace inside JSON-formatted `data:` @@ -108,8 +137,17 @@ 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 sseSizeError('SSE event data', maxEventSizeBytes); + } } } + + // 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 sseSizeError('SSE line', maxEventSizeBytes); + } } // Yield any pending event at stream end. @@ -118,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 diff --git a/test/sse_utils.spec.ts b/test/sse_utils.spec.ts index 1bc19ed4..de6f3eba 100644 --- a/test/sse_utils.spec.ts +++ b/test/sse_utils.spec.ts @@ -335,4 +335,55 @@ 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 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. + 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); + }); + }); });