From ac34171a295d0b86440498eb6619119087ae5a57 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Tue, 23 Jun 2026 13:33:12 +0200 Subject: [PATCH 1/2] fix: limit EventSource event size Signed-off-by: Matteo Collina --- docs/docs/api/Client.md | 4 ++ docs/docs/api/EventSource.md | 4 ++ lib/dispatcher/client.js | 5 +- lib/dispatcher/dispatcher-base.js | 14 ++++- lib/web/eventsource/eventsource-stream.js | 46 +++++++++++++++- lib/web/eventsource/eventsource.js | 4 +- test/eventsource/eventsource-constructor.js | 35 +++++++++++++ .../eventsource-dispatcher-options.js | 52 +++++++++++++++++++ test/eventsource/eventsource-stream.js | 25 +++++++++ test/types/event-source-d.ts | 4 +- types/client.d.ts | 10 ++++ 11 files changed, 196 insertions(+), 7 deletions(-) create mode 100644 test/eventsource/eventsource-dispatcher-options.js diff --git a/docs/docs/api/Client.md b/docs/docs/api/Client.md index dc6ab7a6d5f..a0be3b49df7 100644 --- a/docs/docs/api/Client.md +++ b/docs/docs/api/Client.md @@ -134,6 +134,10 @@ added: v1.0.0 WebSocket messages. Applied to uncompressed messages, compressed frame payloads, and decompressed (`permessage-deflate`) messages. Set to `0` to disable the limit. **Default:** `134217728`. + * `eventSource` {Object} (optional) EventSource-specific configuration. + * `maxEventSize` {number} The maximum allowed event size, in bytes, for + EventSource messages. Set to `0` to disable the limit. + **Default:** `buffer.kStringMaxLength`. * Returns: {Client} Instantiating a `Client` does not open a connection; the connection is diff --git a/docs/docs/api/EventSource.md b/docs/docs/api/EventSource.md index 938a7a8c6f2..285a983379f 100644 --- a/docs/docs/api/EventSource.md +++ b/docs/docs/api/EventSource.md @@ -66,6 +66,9 @@ added: v6.5.0 wait before re-establishing a dropped connection. The server may override this value with a `retry` field. **Default:** `3000`. +EventSource-specific limits can be configured on the dispatcher using the +`eventSource` option. See [`Client`][] for details. + Creates a new `EventSource` and immediately begins connecting to `url`. The request is sent with the `Accept: text/event-stream` header, a cache mode of `no-store`, and an initiator type of `other`. @@ -349,6 +352,7 @@ eventSource.onerror = () => { ``` [WHATWG-conformant]: https://html.spec.whatwg.org/multipage/server-sent-events.html#server-sent-events +[`Client`]: Client.md#new-clienturl-options [`Dispatcher`]: Dispatcher.md#class-dispatcher [`addEventListener()`]: https://developer.mozilla.org/en-US/docs/Web/API/EventTarget/addEventListener [server-sent events]: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events diff --git a/lib/dispatcher/client.js b/lib/dispatcher/client.js index 8a4f65171bd..581e9aad6ac 100644 --- a/lib/dispatcher/client.js +++ b/lib/dispatcher/client.js @@ -128,7 +128,8 @@ class Client extends DispatcherBase { initialWindowSize, connectionWindowSize, pingInterval, - webSocket + webSocket, + eventSource } = {}) { if (keepAlive !== undefined) { throw new InvalidArgumentError('unsupported keepAlive, use pipelining=0 instead') @@ -236,7 +237,7 @@ class Client extends DispatcherBase { throw new InvalidArgumentError('pingInterval must be a positive integer, greater or equal to 0') } - super({ webSocket }) + super({ webSocket, eventSource }) if (typeof connect !== 'function') { connect = buildConnector({ diff --git a/lib/dispatcher/dispatcher-base.js b/lib/dispatcher/dispatcher-base.js index 21b5e346aca..d9d15d6441a 100644 --- a/lib/dispatcher/dispatcher-base.js +++ b/lib/dispatcher/dispatcher-base.js @@ -1,5 +1,6 @@ 'use strict' +const buffer = require('node:buffer') const Dispatcher = require('./dispatcher') const { ClientDestroyedError, @@ -11,6 +12,7 @@ const { kDestroy, kClose, kClosed, kDestroyed, kDispatch } = require('../core/sy const kOnDestroyed = Symbol('onDestroyed') const kOnClosed = Symbol('onClosed') const kWebSocketOptions = Symbol('webSocketOptions') +const kEventSourceOptions = Symbol('eventSourceOptions') class DispatcherBase extends Dispatcher { /** @type {boolean} */ @@ -31,10 +33,11 @@ class DispatcherBase extends Dispatcher { constructor (opts) { super() this[kWebSocketOptions] = opts?.webSocket ?? {} + this[kEventSourceOptions] = opts?.eventSource ?? {} } /** - * @returns {import('../../types/dispatcher').WebSocketOptions} + * @returns {import('../../types/client').Client.WebSocketOptions} */ get webSocketOptions () { return { @@ -43,6 +46,15 @@ class DispatcherBase extends Dispatcher { } } + /** + * @returns {import('../../types/client').Client.EventSourceOptions} + */ + get eventSourceOptions () { + return { + maxEventSize: this[kEventSourceOptions].maxEventSize ?? buffer.kStringMaxLength + } + } + /** @returns {boolean} */ get destroyed () { return this[kDestroyed] diff --git a/lib/web/eventsource/eventsource-stream.js b/lib/web/eventsource/eventsource-stream.js index 7b9e2f8cbba..12db53984c5 100644 --- a/lib/web/eventsource/eventsource-stream.js +++ b/lib/web/eventsource/eventsource-stream.js @@ -1,4 +1,5 @@ 'use strict' +const buffer = require('node:buffer') const { Transform } = require('node:stream') const { isASCIINumber, isValidLastEventId } = require('./util') @@ -23,6 +24,8 @@ const COLON = 0x3A */ const SPACE = 0x20 +const defaultMaxEventSize = buffer.kStringMaxLength + const DATA = Buffer.from('data') const EVENT = Buffer.from('event') const ID = Buffer.from('id') @@ -66,6 +69,12 @@ function isFieldName (line, length, field) { return true } +function createMaxEventSizeExceededError () { + const error = new Error('EventSource message size exceeded') + error.aborted = false + return error +} + /** * @typedef {object} EventSourceStreamEvent * @type {object} @@ -114,6 +123,8 @@ class EventSourceStream extends Transform { pos = 0 lineChunkIndex = 0 linePos = 0 + eventDataSize = 0 + maxEventSize event = { data: undefined, @@ -125,6 +136,7 @@ class EventSourceStream extends Transform { /** * @param {object} options * @param {boolean} [options.readableObjectMode] + * @param {number} [options.maxEventSize] * @param {eventSourceSettings} [options.eventSourceSettings] * @param {(chunk: any, encoding?: BufferEncoding | undefined) => boolean} [options.push] */ @@ -136,6 +148,7 @@ class EventSourceStream extends Transform { super(options) this.state = options.eventSourceSettings || {} + this.maxEventSize = options.maxEventSize ?? defaultMaxEventSize if (options.push) { this.push = options.push } @@ -231,7 +244,12 @@ class EventSourceStream extends Transform { // In any case, we can process the line as we reached an // end-of-line character - this.parseLine(this.readLine(), this.event) + try { + this.parseLine(this.readLine(), this.event) + } catch (error) { + callback(error) + return + } this.consumeCurrentByte() // A line was processed and this could be the end of the event. We need // to check if the next line is empty to determine if the event is @@ -282,6 +300,13 @@ class EventSourceStream extends Transform { } if (isFieldName(line, fieldLength, DATA)) { + const valueBytes = line.length - valueStart + const eventDataSize = this.getEventDataSize(event) + (event.data === undefined ? 0 : 1) + valueBytes + + if (this.maxEventSize > 0 && eventDataSize > this.maxEventSize) { + throw createMaxEventSizeExceededError() + } + const value = line.toString('utf8', valueStart) if (event.data === undefined) { @@ -289,6 +314,7 @@ class EventSourceStream extends Transform { } else { event.data += `\n${value}` } + this.setEventDataSize(event, eventDataSize) return } @@ -345,6 +371,21 @@ class EventSourceStream extends Transform { this.event.event = undefined this.event.id = undefined this.event.retry = undefined + this.eventDataSize = 0 + } + + getEventDataSize (event) { + if (event === this.event) { + return this.eventDataSize + } + + return event.data === undefined ? 0 : Buffer.byteLength(event.data) + } + + setEventDataSize (event, eventDataSize) { + if (event === this.event) { + this.eventDataSize = eventDataSize + } } hasPendingEvent () { @@ -490,5 +531,6 @@ class EventSourceStream extends Transform { } module.exports = { - EventSourceStream + EventSourceStream, + defaultMaxEventSize } diff --git a/lib/web/eventsource/eventsource.js b/lib/web/eventsource/eventsource.js index 17a1de7b7ba..163aa3fa2aa 100644 --- a/lib/web/eventsource/eventsource.js +++ b/lib/web/eventsource/eventsource.js @@ -10,6 +10,7 @@ const { isNetworkError } = require('../fetch/response') const { kEnumerableProperty } = require('../../core/util') const { environmentSettingsObject } = require('../fetch/util') const { createPotentialCORSRequest } = require('./util') +const { getGlobalDispatcher } = require('../../global') let experimentalWarned = false @@ -123,7 +124,7 @@ class EventSource extends EventTarget { url = webidl.converters.USVString(url) eventSourceInitDict = webidl.converters.EventSourceInitDict(eventSourceInitDict, prefix, 'eventSourceInitDict') - this.#dispatcher = eventSourceInitDict.node.dispatcher || eventSourceInitDict.dispatcher + this.#dispatcher = eventSourceInitDict.node.dispatcher || eventSourceInitDict.dispatcher || getGlobalDispatcher() this.#state = { lastEventId: '', reconnectionTime: eventSourceInitDict.node.reconnectionTime @@ -281,6 +282,7 @@ class EventSource extends EventTarget { const eventSourceStream = new EventSourceStream({ eventSourceSettings: this.#state, + maxEventSize: this.#dispatcher.eventSourceOptions?.maxEventSize, push: (event) => { this.dispatchEvent(createFastMessageEvent( event.type, diff --git a/test/eventsource/eventsource-constructor.js b/test/eventsource/eventsource-constructor.js index 00a33965ed6..9b7048fedf2 100644 --- a/test/eventsource/eventsource-constructor.js +++ b/test/eventsource/eventsource-constructor.js @@ -4,8 +4,43 @@ const { once } = require('node:events') const http = require('node:http') const { test, describe } = require('node:test') const { EventSource } = require('../../lib/web/eventsource/eventsource') +const { Agent } = require('../..') describe('EventSource - withCredentials', () => { + test('dispatcher eventSource.maxEventSize closes the connection when exceeded', async (t) => { + const server = http.createServer({ joinDuplicateHeaders: true }, (req, res) => { + res.writeHead(200, 'OK', { 'Content-Type': 'text/event-stream' }) + setImmediate(() => { + res.write('data: hello\n') + res.write('data: world\n') + }) + }) + + await once(server.listen(0), 'listening') + const port = server.address().port + + const agent = new Agent({ + eventSource: { + maxEventSize: 5 + } + }) + const eventSourceInstance = new EventSource(`http://localhost:${port}`, { + node: { + dispatcher: agent + } + }) + t.after(async () => { + eventSourceInstance.close() + server.close() + await agent.close() + }) + + await once(eventSourceInstance, 'open') + await once(eventSourceInstance, 'error', { signal: AbortSignal.timeout(1000) }) + + t.assert.strictEqual(eventSourceInstance.readyState, EventSource.CLOSED) + }) + test('withCredentials should be false by default', async (t) => { const server = http.createServer({ joinDuplicateHeaders: true }, (req, res) => { res.writeHead(200, 'OK', { 'Content-Type': 'text/event-stream' }) diff --git a/test/eventsource/eventsource-dispatcher-options.js b/test/eventsource/eventsource-dispatcher-options.js new file mode 100644 index 00000000000..4d98c8f3fcd --- /dev/null +++ b/test/eventsource/eventsource-dispatcher-options.js @@ -0,0 +1,52 @@ +'use strict' + +const buffer = require('node:buffer') +const { test } = require('node:test') +const { Agent, Client, Pool } = require('../..') + +test('Agent eventSourceOptions.maxEventSize is read correctly', async (t) => { + const customLimit = 32 * 1024 * 1024 + const agent = new Agent({ + eventSource: { + maxEventSize: customLimit + } + }) + + t.after(() => agent.close()) + + t.assert.strictEqual(agent.eventSourceOptions.maxEventSize, customLimit) +}) + +test('Agent with default eventSourceOptions uses buffer.kStringMaxLength', async (t) => { + const agent = new Agent() + + t.after(() => agent.close()) + + t.assert.strictEqual(agent.eventSourceOptions.maxEventSize, buffer.kStringMaxLength) +}) + +test('Client eventSourceOptions.maxEventSize is read correctly', async (t) => { + const customLimit = 16 * 1024 * 1024 + const client = new Client('http://localhost', { + eventSource: { + maxEventSize: customLimit + } + }) + + t.after(() => client.close()) + + t.assert.strictEqual(client.eventSourceOptions.maxEventSize, customLimit) +}) + +test('Pool eventSourceOptions.maxEventSize is read correctly', async (t) => { + const customLimit = 8 * 1024 * 1024 + const pool = new Pool('http://localhost', { + eventSource: { + maxEventSize: customLimit + } + }) + + t.after(() => pool.close()) + + t.assert.strictEqual(pool.eventSourceOptions.maxEventSize, customLimit) +}) diff --git a/test/eventsource/eventsource-stream.js b/test/eventsource/eventsource-stream.js index 7763f46ba29..d6dc277a5e1 100644 --- a/test/eventsource/eventsource-stream.js +++ b/test/eventsource/eventsource-stream.js @@ -4,6 +4,31 @@ const { test, describe } = require('node:test') const { EventSourceStream } = require('../../lib/web/eventsource/eventsource-stream') describe('EventSourceStream', () => { + test('limits accumulated event data', (t) => { + const stream = new EventSourceStream({ maxEventSize: 5 }) + const event = {} + + stream.parseLine(Buffer.from('data: hello', 'utf8'), event) + + t.assert.throws(() => { + stream.parseLine(Buffer.from('data: world', 'utf8'), event) + }, { + name: 'Error', + message: 'EventSource message size exceeded' + }) + t.assert.strictEqual(event.data, 'hello') + }) + + test('can disable event data limit', (t) => { + const stream = new EventSourceStream({ maxEventSize: 0 }) + const event = {} + + stream.parseLine(Buffer.from('data: hello', 'utf8'), event) + stream.parseLine(Buffer.from('data: world', 'utf8'), event) + + t.assert.strictEqual(event.data, 'hello\nworld') + }) + test('ignore empty chunks', (t) => { const stream = new EventSourceStream() diff --git a/test/types/event-source-d.ts b/test/types/event-source-d.ts index e84bce2bcfb..8527fee7985 100644 --- a/test/types/event-source-d.ts +++ b/test/types/event-source-d.ts @@ -1,7 +1,7 @@ import { URL } from 'node:url' import { expectType, expectAssignable } from 'tsd' -import { EventSource, EventSourceInit, Dispatcher } from '../../' +import { Agent, EventSource, EventSourceInit, Dispatcher } from '../../' declare const eventSource: EventSource declare const agent: Dispatcher @@ -21,3 +21,5 @@ expectType(new EventSource('https://example.com', { expectAssignable({ dispatcher: agent }) expectAssignable({ withCredentials: true }) expectAssignable({}) + +expectType(new Agent({ eventSource: { maxEventSize: 1024 } })) diff --git a/types/client.d.ts b/types/client.d.ts index e3b121962ef..7db7b8c4044 100644 --- a/types/client.d.ts +++ b/types/client.d.ts @@ -75,6 +75,8 @@ export declare namespace Client { maxResponseSize?: number; /** WebSocket-specific options */ webSocket?: Client.WebSocketOptions; + /** EventSource-specific options */ + eventSource?: Client.EventSourceOptions; /** Enables a family autodetection algorithm that loosely implements section 5 of RFC 8305. */ autoSelectFamily?: boolean; /** The amount of time in milliseconds to wait for a connection attempt to finish before trying the next address when using the `autoSelectFamily` option. */ @@ -129,6 +131,14 @@ export declare namespace Client { */ maxPayloadSize?: number; } + export interface EventSourceOptions { + /** + * Maximum allowed event size in bytes for EventSource messages. + * Set to 0 to disable the limit. + * @default buffer.kStringMaxLength + */ + maxEventSize?: number; + } } export default Client From df7b4aed0536d8f6700703268396aa912c2de70e Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 16 Jul 2026 16:41:32 +0000 Subject: [PATCH 2/2] fix: address EventSource review comments Signed-off-by: Matteo Collina --- lib/web/eventsource/eventsource-stream.js | 21 +++------------------ lib/web/eventsource/eventsource.js | 5 +++-- 2 files changed, 6 insertions(+), 20 deletions(-) diff --git a/lib/web/eventsource/eventsource-stream.js b/lib/web/eventsource/eventsource-stream.js index 12db53984c5..03000e67dd0 100644 --- a/lib/web/eventsource/eventsource-stream.js +++ b/lib/web/eventsource/eventsource-stream.js @@ -301,7 +301,7 @@ class EventSourceStream extends Transform { if (isFieldName(line, fieldLength, DATA)) { const valueBytes = line.length - valueStart - const eventDataSize = this.getEventDataSize(event) + (event.data === undefined ? 0 : 1) + valueBytes + const eventDataSize = this.eventDataSize + (event.data === undefined ? 0 : 1) + valueBytes if (this.maxEventSize > 0 && eventDataSize > this.maxEventSize) { throw createMaxEventSizeExceededError() @@ -314,7 +314,7 @@ class EventSourceStream extends Transform { } else { event.data += `\n${value}` } - this.setEventDataSize(event, eventDataSize) + this.eventDataSize = eventDataSize return } @@ -374,20 +374,6 @@ class EventSourceStream extends Transform { this.eventDataSize = 0 } - getEventDataSize (event) { - if (event === this.event) { - return this.eventDataSize - } - - return event.data === undefined ? 0 : Buffer.byteLength(event.data) - } - - setEventDataSize (event, eventDataSize) { - if (event === this.event) { - this.eventDataSize = eventDataSize - } - } - hasPendingEvent () { return this.event.data !== undefined || this.event.event !== undefined || @@ -531,6 +517,5 @@ class EventSourceStream extends Transform { } module.exports = { - EventSourceStream, - defaultMaxEventSize + EventSourceStream } diff --git a/lib/web/eventsource/eventsource.js b/lib/web/eventsource/eventsource.js index 163aa3fa2aa..b34c89c1648 100644 --- a/lib/web/eventsource/eventsource.js +++ b/lib/web/eventsource/eventsource.js @@ -124,7 +124,7 @@ class EventSource extends EventTarget { url = webidl.converters.USVString(url) eventSourceInitDict = webidl.converters.EventSourceInitDict(eventSourceInitDict, prefix, 'eventSourceInitDict') - this.#dispatcher = eventSourceInitDict.node.dispatcher || eventSourceInitDict.dispatcher || getGlobalDispatcher() + this.#dispatcher = eventSourceInitDict.node.dispatcher || eventSourceInitDict.dispatcher this.#state = { lastEventId: '', reconnectionTime: eventSourceInitDict.node.reconnectionTime @@ -467,7 +467,8 @@ webidl.converters.EventSourceInitDict = webidl.dictionaryConverter([ }, { key: 'dispatcher', // undici only - converter: webidl.converters.any + converter: webidl.converters.any, + defaultValue: () => getGlobalDispatcher() }, { key: 'node', // undici only