fix(client): bound SSE event size to prevent client-side memory exhaustion (#582)

# Description

`parseSseStream` (shared by the JSON-RPC and REST client transports)
accumulates two buffers with **no size bound**:

- the line `buffer` — `buffer += value` on every chunk
- an event's joined `eventData` — consecutive `data:` lines are appended

Because A2A clients stream SSE from remote, potentially untrusted agent
servers, a malicious or broken server can exploit this as a
**denial-of-service
against the client**:

- stream bytes that **never contain a `\n`** → the line `buffer` grows
without bound
- stream endless `data:` lines with **no terminating blank line** →
`eventData` grows without bound

Either way the client keeps allocating without bound. Today that can
result in
a process-level OOM crash that the caller cannot handle as a normal
stream
error.

## Fix

Cap both accumulation points at `maxEventSizeBytes` and throw when
exceeded.
The throw runs the async generator's teardown, whose `finally` now
cancels the
reader (#580), so the offending connection is also closed. Net effect: a
**process-level crash becomes a catchable `Error`** that flows into the
transports' existing SSE-parse error handling.

The same limit applies to the unterminated line buffer because a single
SSE
field line is necessarily part of an event and should not be allowed to
exceed
the maximum event budget on its own.

- Default is **4 MiB** (matching gRPC's default max message size):
realistic
  A2A events (Message/Task JSON) are KB-scale, and large files should be
  referenced via `FileWithUri` parts rather than inlined.
- It is **configurable** via the new optional `maxEventSizeBytes`
argument for
callers that must inline larger payloads, and the error message points
at
  both options.

## Open question — is this the right layer?

I'm **not sure `parseSseStream` is the right place** for this, and would
appreciate maintainer guidance. Alternatives worth considering:

- enforcing it at the transport layer instead of the shared parser,
- deriving/validating against `Content-Length` where present,
- or leaving DoS mitigation entirely to the runtime / a reverse proxy.

Happy to change the default, make it opt-in, or move it elsewhere based
on
what you prefer.

## Tests

Adds regression tests for the unterminated-line,
terminated-oversized-line
(single-chunk), and unterminated-event cases — all would grow unbounded
without the cap — plus a within-limit sanity check.
`test/sse_utils.spec.ts`
is green.

- [x] Follows the `CONTRIBUTING` guide
- [x] PR title uses Conventional Commits (`fix:`)
- [x] Tests and linter pass
- [ ] Docs updated (not necessary)

Author: RenKoya1

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<SseEvent, void, undefined> {
   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

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<SseEvent[]> {
+      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);
+    });
+  });
 });