From 7700f48011c454e82d5f28541e1533dcc15109fc Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Wed, 23 Sep 2026 17:46:39 +0200 Subject: [PATCH] fix: bound incoming transport message sizes --- src/acp.ts | 10 ++++- src/http-stream.ts | 89 ++++++++++++++++++++++++++++++++++---- src/line-buffer.ts | 80 ++++++++++++++++++++-------------- src/sse.ts | 100 +++++++++++++++++++++++++++---------------- src/stream-limits.ts | 70 ++++++++++++++++++++++++++++++ src/stream.ts | 11 ++++- src/v2/acp.ts | 9 +++- typos.toml | 1 + 8 files changed, 291 insertions(+), 79 deletions(-) create mode 100644 src/stream-limits.ts diff --git a/src/acp.ts b/src/acp.ts index eb6aa189..373bab62 100644 --- a/src/acp.ts +++ b/src/acp.ts @@ -48,11 +48,17 @@ export type Stream = { export function ndJsonStream( output: WritableStream, input: ReadableStream, + options?: NdJsonStreamOptions, ): Stream { - return createJsonStream(output, input); + return createJsonStream(output, input, options); } export { RequestError } from "./jsonrpc.js"; +export { + DEFAULT_MAX_MESSAGE_BYTES, + MessageTooLargeError, +} from "./stream-limits.js"; +export type { NdJsonStreamOptions } from "./stream.js"; export type { AnyMessage, AnyNotification, @@ -65,7 +71,7 @@ export type { SendRequestOptions, } from "./jsonrpc.js"; -import type { WireStream } from "./stream.js"; +import type { NdJsonStreamOptions, WireStream } from "./stream.js"; import { Connection, Handled, HandlerRegistration } from "./jsonrpc.js"; import type { AnyWireMessage, diff --git a/src/http-stream.ts b/src/http-stream.ts index a8d8c562..7ff7a0d0 100644 --- a/src/http-stream.ts +++ b/src/http-stream.ts @@ -11,6 +11,11 @@ import { } from "./protocol.js"; import { MemoryAcpCookieStore } from "./cookie-store.js"; import { parseSseStream } from "./sse.js"; +import { + MessageBuffer, + MessageTooLargeError, + resolveMaxMessageBytes, +} from "./stream-limits.js"; import type { AcpCookieStore } from "./cookie-store.js"; import type { AnyMessage } from "./jsonrpc.js"; @@ -36,10 +41,15 @@ export interface HttpStreamOptions { * when the stream closes/errors. */ readonly cookieStore?: AcpCookieStore; + readonly maxMessageBytes?: number; } export { MemoryAcpCookieStore } from "./cookie-store.js"; export type { AcpCookieStore } from "./cookie-store.js"; +export { + DEFAULT_MAX_MESSAGE_BYTES, + MessageTooLargeError, +} from "./stream-limits.js"; /** * Creates an ACP Stream over Streamable HTTP. @@ -66,6 +76,7 @@ class HttpStreamTransport { private readonly fetchImpl: typeof globalThis.fetch; private readonly headers: Record; + private readonly maxMessageBytes: number; private readonly cookiePolicy: RequestCredentials; private readonly cookieStore: AcpCookieStore; private readonly ownsCookieStore: boolean; @@ -85,6 +96,7 @@ class HttpStreamTransport { private readonly serverUrl: string, options: HttpStreamOptions, ) { + this.maxMessageBytes = resolveMaxMessageBytes(options.maxMessageBytes); this.fetchImpl = resolveFetch(options.fetch); this.headers = options.headers ?? {}; this.cookiePolicy = options.cookies ?? "include"; @@ -148,7 +160,11 @@ class HttpStreamTransport { }); if (!response.ok) { - throw await httpError("ACP initialize failed", response); + throw await httpError( + "ACP initialize failed", + response, + this.maxMessageBytes, + ); } const connectionId = response.headers.get(HEADER_CONNECTION_ID); @@ -159,7 +175,9 @@ class HttpStreamTransport { cleanupConnectionId = connectionId; this.throwIfClosedDuringInitialize(); - const body: unknown = await response.json(); + const body: unknown = JSON.parse( + await readResponseText(response, this.maxMessageBytes), + ); this.throwIfClosedDuringInitialize(); if (!isResponseMessage(body)) { @@ -229,8 +247,13 @@ class HttpStreamTransport { }); if (!response.ok) { - throw await httpError("ACP POST failed", response); + throw await httpError( + "ACP POST failed", + response, + this.maxMessageBytes, + ); } + void response.body?.cancel().catch(() => {}); if (!("method" in message) && "id" in message) { const key = messageIdKey(message.id); @@ -353,7 +376,11 @@ class HttpStreamTransport { }); if (!response.ok) { - throw await httpError("ACP SSE connection failed", response); + throw await httpError( + "ACP SSE connection failed", + response, + this.maxMessageBytes, + ); } if (!response.body) { @@ -362,7 +389,10 @@ class HttpStreamTransport { lifecycle.onOpen?.(); - for await (const message of parseSseStream(response.body)) { + for await (const message of parseSseStream( + response.body, + this.maxMessageBytes, + )) { if (this.isClosed) { return; } @@ -509,8 +539,13 @@ class HttpStreamTransport { }); if (!response.ok) { - throw await httpError("ACP DELETE failed", response); + throw await httpError( + "ACP DELETE failed", + response, + this.maxMessageBytes, + ); } + void response.body?.cancel().catch(() => {}); } private clearOwnedCookieStore(): void { @@ -575,8 +610,46 @@ function resolveFetch( ); } -async function httpError(prefix: string, response: Response): Promise { - const text = await response.text().catch(() => ""); +async function readResponseText( + response: Response, + maxMessageBytes: number, +): Promise { + if (!response.body) { + return ""; + } + + const buffer = new MessageBuffer(maxMessageBytes); + const reader = response.body.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) { + return new TextDecoder().decode(buffer.take()); + } + buffer.append(value); + } + } catch (error) { + void reader.cancel(error).catch(() => {}); + throw error; + } finally { + buffer.clear(); + reader.releaseLock(); + } +} + +async function httpError( + prefix: string, + response: Response, + maxMessageBytes: number, +): Promise { + let text = ""; + try { + text = await readResponseText(response, maxMessageBytes); + } catch (error) { + if (error instanceof MessageTooLargeError) { + throw error; + } + } if (text) { return new Error( diff --git a/src/line-buffer.ts b/src/line-buffer.ts index c7bf5204..de421880 100644 --- a/src/line-buffer.ts +++ b/src/line-buffer.ts @@ -1,3 +1,9 @@ +import { + MessageBuffer, + MessageTooLargeError, + resolveMaxMessageBytes, +} from "./stream-limits.js"; + const newline = 0x0a; /** @@ -6,35 +12,38 @@ const newline = 0x0a; * Only the newly pushed chunk is scanned for newlines, so splitting costs * O(total bytes) no matter how many chunks a line spans. * - * Chunks passed to {@link push} must not be mutated afterwards; a chunk - * without a newline is retained until its line completes. + * Chunks passed to {@link push} must not be mutated while its iterator is + * in use. */ export class LineBuffer { /** Bytes of the current (incomplete) line, carried across chunks. */ - #pending: Uint8Array[] = []; + readonly #pending: MessageBuffer; + readonly #maxMessageBytes: number; + + constructor(maxMessageBytes?: number) { + this.#maxMessageBytes = resolveMaxMessageBytes(maxMessageBytes); + this.#pending = new MessageBuffer( + Math.min(Number.MAX_SAFE_INTEGER, this.#maxMessageBytes + 1), + ); + } /** - * Consumes a chunk, returning each complete line without its trailing - * newline. + * Consumes a chunk, yielding each complete line without its trailing + * LF or CRLF. */ - push(chunk: Uint8Array): Uint8Array[] { - const lines: Uint8Array[] = []; + *push(chunk: Uint8Array): Generator { let start = 0; let newlineIndex = chunk.indexOf(newline, start); while (newlineIndex !== -1) { - lines.push(this.#takeLine(chunk.subarray(start, newlineIndex))); + yield this.#takeLine(chunk.subarray(start, newlineIndex)); start = newlineIndex + 1; newlineIndex = chunk.indexOf(newline, start); } if (start < chunk.byteLength) { - // Copy a partial tail so a few carried-over bytes don't pin the whole - // chunk's buffer. The constructor guarantees a copy, unlike slice(), - // which Node's Buffer subclass overrides to return a view. - this.#pending.push( - start === 0 ? chunk : new Uint8Array(chunk.subarray(start)), - ); + const tail = chunk.subarray(start); + this.#checkLine(tail); + this.#pending.append(tail); } - return lines; } /** @@ -42,28 +51,37 @@ export class LineBuffer { * undefined if no bytes are buffered. */ flush(): Uint8Array | undefined { - if (this.#pending.length === 0) { + if (this.#pending.byteLength === 0) { return undefined; } - return this.#takeLine(new Uint8Array(0)); + return stripCarriageReturn(this.#pending.take()); + } + + clear(): void { + this.#pending.clear(); } #takeLine(tail: Uint8Array): Uint8Array { - if (this.#pending.length === 0) { - return tail; + this.#checkLine(tail); + if (this.#pending.byteLength === 0) { + return stripCarriageReturn(tail); } - let total = tail.byteLength; - for (const part of this.#pending) { - total += part.byteLength; - } - const line = new Uint8Array(total); - let offset = 0; - for (const part of this.#pending) { - line.set(part, offset); - offset += part.byteLength; + this.#pending.append(tail); + return stripCarriageReturn(this.#pending.take()); + } + + #checkLine(tail: Uint8Array): void { + const lastByte = + tail.byteLength > 0 ? tail[tail.byteLength - 1] : this.#pending.lastByte; + const byteLength = + this.#pending.byteLength + tail.byteLength - (lastByte === 0x0d ? 1 : 0); + if (byteLength > this.#maxMessageBytes) { + this.clear(); + throw new MessageTooLargeError(this.#maxMessageBytes); } - line.set(tail, offset); - this.#pending = []; - return line; } } + +function stripCarriageReturn(line: Uint8Array): Uint8Array { + return line[line.byteLength - 1] === 0x0d ? line.subarray(0, -1) : line; +} diff --git a/src/sse.ts b/src/sse.ts index 55c9fc8c..c8313f3f 100644 --- a/src/sse.ts +++ b/src/sse.ts @@ -1,6 +1,15 @@ import type { AnyMessage } from "./jsonrpc.js"; import { isRecord } from "./jsonrpc.js"; import { LineBuffer } from "./line-buffer.js"; +import { + DEFAULT_MAX_MESSAGE_BYTES, + MessageBuffer, + MessageTooLargeError, +} from "./stream-limits.js"; + +const dataPrefix = new TextEncoder().encode("data:"); +const dataSeparator = new Uint8Array([0x0a]); +const maxLineOverhead = 9; // UTF-8 BOM + "data: "; LineBuffer strips LF/CRLF. export function serializeSseEvent(msg: AnyMessage): string { return `data: ${JSON.stringify(msg)}\n\n`; @@ -12,25 +21,48 @@ export function serializeSseKeepAlive(): string { export async function* parseSseStream( body: ReadableStream, + maxMessageBytes = DEFAULT_MAX_MESSAGE_BYTES, ): AsyncIterable { - const decoder = new TextDecoder(); + const decoder = new TextDecoder("utf-8", { ignoreBOM: true }); + const eventData = new MessageBuffer(maxMessageBytes); + const lines = new LineBuffer( + Math.min(Number.MAX_SAFE_INTEGER, maxMessageBytes + maxLineOverhead), + ); const reader = body.getReader(); - const lines = new LineBuffer(); - let eventLines: string[] = []; - - const decodeLine = (lineBytes: Uint8Array): string => { - const line = decoder.decode(lineBytes); - return line.endsWith("\r") ? line.slice(0, -1) : line; - }; + let hasData = false; + let finished = false; + let cancelReason: unknown; // A blank line ends the current event. const takeEvent = (): AnyMessage | undefined => { - if (eventLines.length === 0) { + if (!hasData) { + return undefined; + } + hasData = false; + return parseSseEvent(decoder.decode(eventData.take())); + }; + + const consumeLine = (line: Uint8Array): AnyMessage | undefined => { + // Preserve the previous decoder's per-line BOM handling. + if (line[0] === 0xef && line[1] === 0xbb && line[2] === 0xbf) { + line = line.subarray(3); + } + if (line.byteLength === 0) { + return takeEvent(); + } + if (!dataPrefix.every((byte, index) => line[index] === byte)) { return undefined; } - const event = eventLines; - eventLines = []; - return parseSseEvent(event); + + const value = line.subarray( + dataPrefix.byteLength + (line[dataPrefix.byteLength] === 0x20 ? 1 : 0), + ); + if (hasData) { + eventData.append(dataSeparator); + } + eventData.append(value); + hasData = true; + return undefined; }; try { @@ -38,51 +70,47 @@ export async function* parseSseStream( const chunk = await reader.read(); if (chunk.done) { + finished = true; break; } for (const lineBytes of lines.push(chunk.value)) { - const line = decodeLine(lineBytes); - if (line === "") { - const msg = takeEvent(); - if (msg) { - yield msg; - } - } else { - eventLines.push(line); + const msg = consumeLine(lineBytes); + if (msg) { + yield msg; } } } const lastLine = lines.flush(); if (lastLine) { - const line = decodeLine(lastLine); - if (line !== "") { - eventLines.push(line); + const msg = consumeLine(lastLine); + if (msg) { + yield msg; } } const msg = takeEvent(); if (msg) { yield msg; } + } catch (error) { + cancelReason = + error instanceof MessageTooLargeError && + error.maxMessageBytes !== maxMessageBytes + ? new MessageTooLargeError(maxMessageBytes) + : error; + throw cancelReason; } finally { + lines.clear(); + eventData.clear(); + if (!finished) { + void reader.cancel(cancelReason).catch(() => {}); + } reader.releaseLock(); } } -function parseSseEvent(eventLines: string[]): AnyMessage | undefined { - const dataLines = eventLines - .filter((line) => line.startsWith("data:")) - .map((line) => { - const value = line.slice("data:".length); - return value.startsWith(" ") ? value.slice(1) : value; - }); - - if (dataLines.length === 0) { - return undefined; - } - - const data = dataLines.join("\n"); +function parseSseEvent(data: string): AnyMessage | undefined { if (!data.trim()) { return undefined; } diff --git a/src/stream-limits.ts b/src/stream-limits.ts new file mode 100644 index 00000000..ddfe13cb --- /dev/null +++ b/src/stream-limits.ts @@ -0,0 +1,70 @@ +export const DEFAULT_MAX_MESSAGE_BYTES = 32 * 1024 * 1024; + +export class MessageTooLargeError extends Error { + constructor(readonly maxMessageBytes: number) { + super( + `Incoming ACP data exceeds the configured ${maxMessageBytes} byte limit`, + ); + this.name = "MessageTooLargeError"; + } +} + +export function resolveMaxMessageBytes(value: number | undefined): number { + const maxMessageBytes = value ?? DEFAULT_MAX_MESSAGE_BYTES; + if (!Number.isSafeInteger(maxMessageBytes) || maxMessageBytes <= 0) { + throw new RangeError("maxMessageBytes must be a positive safe integer"); + } + return maxMessageBytes; +} + +export class MessageBuffer { + #buffer = new Uint8Array(0); + #length = 0; + readonly #maxMessageBytes: number; + + constructor(maxMessageBytes: number) { + this.#maxMessageBytes = resolveMaxMessageBytes(maxMessageBytes); + } + + get byteLength(): number { + return this.#length; + } + + get lastByte(): number | undefined { + return this.#buffer[this.#length - 1]; + } + + #checkAppend(byteLength: number): void { + if (byteLength > this.#maxMessageBytes - this.#length) { + this.clear(); + throw new MessageTooLargeError(this.#maxMessageBytes); + } + } + + append(bytes: Uint8Array): void { + this.#checkAppend(bytes.byteLength); + const length = this.#length + bytes.byteLength; + if (length > this.#buffer.byteLength) { + const capacity = Math.min( + this.#maxMessageBytes, + Math.max(length, this.#buffer.byteLength * 2, 1024), + ); + const buffer = new Uint8Array(capacity); + buffer.set(this.#buffer.subarray(0, this.#length)); + this.#buffer = buffer; + } + this.#buffer.set(bytes, this.#length); + this.#length = length; + } + + take(): Uint8Array { + const bytes = this.#buffer.subarray(0, this.#length); + this.clear(); + return bytes; + } + + clear(): void { + this.#buffer = new Uint8Array(0); + this.#length = 0; + } +} diff --git a/src/stream.ts b/src/stream.ts index 009c39c9..57efcb9f 100644 --- a/src/stream.ts +++ b/src/stream.ts @@ -1,6 +1,11 @@ import type { AnyMessage, AnyWireMessage } from "./jsonrpc.js"; import { RequestError, isRecord, protocolErrorResponse } from "./jsonrpc.js"; import { LineBuffer } from "./line-buffer.js"; +import { resolveMaxMessageBytes } from "./stream-limits.js"; + +export interface NdJsonStreamOptions { + readonly maxMessageBytes?: number; +} /** * Stream interface for ACP connections. @@ -40,7 +45,9 @@ export type WireStream = Stream; export function ndJsonStream( output: WritableStream, input: ReadableStream, + options: NdJsonStreamOptions = {}, ): Stream { + const maxMessageBytes = resolveMaxMessageBytes(options.maxMessageBytes); const textEncoder = new TextEncoder(); const textDecoder = new TextDecoder(); let cancelled = false; @@ -63,7 +70,7 @@ export function ndJsonStream( const readable = new ReadableStream({ async start(controller) { - const lines = new LineBuffer(); + const lines = new LineBuffer(maxMessageBytes); const enqueueLine = async (lineBytes: Uint8Array) => { const trimmedLine = textDecoder.decode(lineBytes).trim(); @@ -121,8 +128,10 @@ export function ndJsonStream( return; } controller.error(err); + void reader.cancel(err).catch(() => {}); return; } finally { + lines.clear(); if (inputReader === reader) { inputReader = undefined; } diff --git a/src/v2/acp.ts b/src/v2/acp.ts index ff06432e..f7efd381 100644 --- a/src/v2/acp.ts +++ b/src/v2/acp.ts @@ -15,6 +15,7 @@ import * as schema from "./schema/index.js"; import * as validate from "./schema/zod.gen.js"; import * as guards from "./schema/guards.gen.js"; import { ndJsonStream as createJsonStream } from "../stream.js"; +import type { NdJsonStreamOptions } from "../stream.js"; export type * from "./schema/types.gen.js"; // Runtime narrowing helpers for extensible unions, exposed as companion values // that merge (declaration merging) with the like-named types — e.g. @@ -81,11 +82,17 @@ export type Stream = WireStream; export function ndJsonStream( output: WritableStream, input: ReadableStream, + options?: NdJsonStreamOptions, ): Stream { - return createJsonStream(output, input); + return createJsonStream(output, input, options); } export { RequestError } from "../jsonrpc.js"; +export { + DEFAULT_MAX_MESSAGE_BYTES, + MessageTooLargeError, +} from "../stream-limits.js"; +export type { NdJsonStreamOptions } from "../stream.js"; export { AgentProtocolRouter, agentProtocolRouter, diff --git a/typos.toml b/typos.toml index 5ceb171e..934fe83f 100644 --- a/typos.toml +++ b/typos.toml @@ -29,3 +29,4 @@ check-filename = true [default.extend-identifiers] ndJsonStream = "ndJsonStream" +NdJsonStreamOptions = "NdJsonStreamOptions"