fix: limit EventSource event size

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

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

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({

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]

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,13 +300,21 @@ class EventSourceStream extends Transform {
     }
 
     if (isFieldName(line, fieldLength, DATA)) {
+      const valueBytes = line.length - valueStart
+      const eventDataSize = this.eventDataSize + (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) {
         event.data = value
       } else {
         event.data += `\n${value}`
       }
+      this.eventDataSize = eventDataSize
       return
     }
 
@@ -345,6 +371,7 @@ class EventSourceStream extends Transform {
     this.event.event = undefined
     this.event.id = undefined
     this.event.retry = undefined
+    this.eventDataSize = 0
   }
 
   hasPendingEvent () {

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
 
@@ -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,
@@ -465,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

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' })

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)
+})

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()
 

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<EventSource>(new EventSource('https://example.com', {
 expectAssignable<EventSourceInit>({ dispatcher: agent })
 expectAssignable<EventSourceInit>({ withCredentials: true })
 expectAssignable<EventSourceInit>({})
+
+expectType<Agent>(new Agent({ eventSource: { maxEventSize: 1024 } }))

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