fix(websocket,eventsource): keep receive limits behind dispatcher wrappers

lib/core/constants.js

@@ -114,7 +114,18 @@ for (let i = 0; i < wellknownHeaderNames.length; ++i) {
     lowerCasedKey
 }
 
+/**
+ * Default WebSocket receive limits (see `webSocket` in Client.md). Used by
+ * DispatcherBase#webSocketOptions and by the WebSocket parser when the
+ * dispatcher doesn't provide a value, so a dispatcher without
+ * `webSocketOptions` (a custom one, or a wrapper) doesn't remove the limits.
+ */
+const defaultWebSocketMaxFragments = 131072
+const defaultWebSocketMaxPayloadSize = 128 * 1024 * 1024 // 128 MB
+
 module.exports = {
   wellknownHeaderNames,
-  headerNameLowerCasedRecord
+  headerNameLowerCasedRecord,
+  defaultWebSocketMaxFragments,
+  defaultWebSocketMaxPayloadSize
 }

lib/dispatcher/dispatcher-base.js

@@ -8,6 +8,7 @@ const {
   InvalidArgumentError
 } = require('../core/errors')
 const { kDestroy, kClose, kClosed, kDestroyed, kDispatch } = require('../core/symbols')
+const { defaultWebSocketMaxFragments, defaultWebSocketMaxPayloadSize } = require('../core/constants')
 
 const kOnDestroyed = Symbol('onDestroyed')
 const kOnClosed = Symbol('onClosed')
@@ -41,8 +42,8 @@ class DispatcherBase extends Dispatcher {
    */
   get webSocketOptions () {
     return {
-      maxFragments: this[kWebSocketOptions].maxFragments ?? 131072,
-      maxPayloadSize: this[kWebSocketOptions].maxPayloadSize ?? 128 * 1024 * 1024 // 128 MB default
+      maxFragments: this[kWebSocketOptions].maxFragments ?? defaultWebSocketMaxFragments,
+      maxPayloadSize: this[kWebSocketOptions].maxPayloadSize ?? defaultWebSocketMaxPayloadSize
     }
   }
 

lib/dispatcher/dispatcher1-wrapper.js

@@ -104,6 +104,17 @@ class Dispatcher1Wrapper extends Dispatcher {
     return this.#dispatcher.dispatch(opts, Dispatcher1Wrapper.wrapHandler(handler))
   }
 
+  // Older undici versions read WebSocket and EventSource receive limits from
+  // the (legacy) global dispatcher, which is this wrapper, so report the
+  // wrapped dispatcher's settings.
+  get webSocketOptions () {
+    return this.#dispatcher.webSocketOptions
+  }
+
+  get eventSourceOptions () {
+    return this.#dispatcher.eventSourceOptions
+  }
+
   close (...args) {
     return this.#dispatcher.close(...args)
   }

lib/dispatcher/retry-agent.js

@@ -26,6 +26,16 @@ class RetryAgent extends Dispatcher {
     return this.#agent.dispatch(opts, retry)
   }
 
+  // WebSocket and EventSource read their receive limits from the dispatcher
+  // they are given, so report the wrapped agent's settings rather than none.
+  get webSocketOptions () {
+    return this.#agent.webSocketOptions
+  }
+
+  get eventSourceOptions () {
+    return this.#agent.eventSourceOptions
+  }
+
   close () {
     return this.#agent.close()
   }

lib/mock/mock-agent.js

@@ -55,6 +55,17 @@ class MockAgent extends Dispatcher {
     }
   }
 
+  // WebSocket and EventSource read their receive limits from the dispatcher
+  // they are given, so report the settings of the agent that makes real
+  // connections.
+  get webSocketOptions () {
+    return this[kAgent].webSocketOptions
+  }
+
+  get eventSourceOptions () {
+    return this[kAgent].eventSourceOptions
+  }
+
   get (origin) {
     // Normalize origin to handle URL objects and case-insensitive hostnames
     const normalizedOrigin = normalizeOrigin(origin)

lib/web/websocket/receiver.js

@@ -16,6 +16,7 @@ const { failWebsocketConnection } = require('./connection')
 const { WebsocketFrameSend } = require('./frame')
 const { PerMessageDeflate } = require('./permessage-deflate')
 const { MessageSizeExceededError } = require('../../core/errors')
+const { defaultWebSocketMaxFragments, defaultWebSocketMaxPayloadSize } = require('../../core/constants')
 
 // This code was influenced by ws released under the MIT license.
 // Copyright (c) 2011 Einar Otto Stangvik <einaros@gmail.com>
@@ -48,18 +49,25 @@ class ByteParser extends Writable {
   /**
    * @param {import('./websocket').Handler} handler
    * @param {Map<string, string>|null} extensions
-   * @param {{ maxPayloadSize?: number }} [options]
+   * @param {{ maxFragments?: number, maxPayloadSize?: number }} [options]
+   *   A missing value uses the default limit; `0` disables the limit.
    */
   constructor (handler, extensions, options = {}) {
     super()
 
     this.#handler = handler
     this.#extensions = extensions == null ? new Map() : extensions
-    this.#maxFragments = options.maxFragments ?? 0
-    this.#maxPayloadSize = options.maxPayloadSize ?? 0
+    // Options come from the dispatcher's `webSocketOptions`, which only
+    // DispatcherBase provides. Fall back to the defaults rather than to no
+    // limit, so RetryAgent, MockAgent or a custom dispatcher can't turn off
+    // the payload and decompression limits by accident.
+    this.#maxFragments = options.maxFragments ?? defaultWebSocketMaxFragments
+    this.#maxPayloadSize = options.maxPayloadSize ?? defaultWebSocketMaxPayloadSize
 
     if (this.#extensions.has('permessage-deflate')) {
-      this.#extensions.set('permessage-deflate', new PerMessageDeflate(extensions, options))
+      this.#extensions.set('permessage-deflate', new PerMessageDeflate(extensions, {
+        maxPayloadSize: this.#maxPayloadSize
+      }))
     }
   }
 

test/eventsource/eventsource-dispatcher-options.js

@@ -1,8 +1,11 @@
 'use strict'
 
 const buffer = require('node:buffer')
+const http = require('node:http')
+const { once } = require('node:events')
 const { test } = require('node:test')
-const { Agent, Client, Pool } = require('../..')
+const { Agent, Client, Dispatcher1Wrapper, MockAgent, Pool, RetryAgent } = require('../..')
+const { EventSource } = require('../../lib/web/eventsource/eventsource')
 
 test('Agent eventSourceOptions.maxEventSize is read correctly', async (t) => {
   const customLimit = 32 * 1024 * 1024
@@ -50,3 +53,70 @@ test('Pool eventSourceOptions.maxEventSize is read correctly', async (t) => {
 
   t.assert.strictEqual(pool.eventSourceOptions.maxEventSize, customLimit)
 })
+
+// Wrappers report the eventSourceOptions of the dispatcher they wrap, so a
+// configured maxEventSize isn't lost behind them.
+const eventSource = { maxEventSize: 5 }
+const wrappers = {
+  RetryAgent: () => new RetryAgent(new Agent({ eventSource })),
+  MockAgent: () => {
+    const agent = new MockAgent({ eventSource })
+    agent.enableNetConnect()
+    return agent
+  },
+  'MockAgent({ agent })': () => {
+    const agent = new MockAgent({ agent: new Agent({ eventSource }) })
+    agent.enableNetConnect()
+    return agent
+  },
+  Dispatcher1Wrapper: () => new Dispatcher1Wrapper(new Agent({ eventSource }))
+}
+
+// Resolves with 'message' or 'error', whichever the EventSource emits first,
+// for a server that sends one event with `data`.
+async function firstEvent (t, dispatcher, data) {
+  const server = http.createServer((req, res) => {
+    res.writeHead(200, { 'Content-Type': 'text/event-stream' })
+    res.write(`data: ${data}\n\n`)
+  })
+  await once(server.listen(0), 'listening')
+
+  const es = new EventSource(`http://localhost:${server.address().port}`, {
+    node: { dispatcher }
+  })
+  t.after(async () => {
+    es.close()
+    server.closeAllConnections()
+    server.close()
+    await dispatcher.close()
+  })
+
+  const type = await new Promise((resolve) => {
+    es.onmessage = () => resolve('message')
+    es.onerror = () => resolve('error')
+  })
+  return { type, readyState: es.readyState }
+}
+
+for (const [name, createDispatcher] of Object.entries(wrappers)) {
+  test(`${name} reports the wrapped agent's eventSourceOptions`, async (t) => {
+    const dispatcher = createDispatcher()
+    t.after(() => dispatcher.close())
+
+    t.assert.deepStrictEqual(dispatcher.eventSourceOptions, eventSource)
+  })
+
+  test(`${name} applies the wrapped agent's maxEventSize`, async (t) => {
+    // 11 bytes of data against a limit of 5.
+    const { type, readyState } = await firstEvent(t, createDispatcher(), 'hello world')
+
+    t.assert.strictEqual(type, 'error')
+    t.assert.strictEqual(readyState, EventSource.CLOSED)
+  })
+
+  test(`${name} delivers events within the wrapped agent's maxEventSize`, async (t) => {
+    const { type } = await firstEvent(t, createDispatcher(), 'hello')
+
+    t.assert.strictEqual(type, 'message')
+  })
+}

test/websocket/dispatcher-options.js

@@ -2,9 +2,13 @@
 
 const { test } = require('node:test')
 const {
+  Agent,
   BalancedPool,
+  Dispatcher1Wrapper,
   EnvHttpProxyAgent,
+  MockAgent,
   ProxyAgent,
+  RetryAgent,
   RoundRobinPool,
   Socks5ProxyAgent
 } = require('../..')
@@ -27,7 +31,12 @@ const dispatchers = {
     webSocket
   }),
   RoundRobinPool: () => new RoundRobinPool('http://localhost', { webSocket }),
-  Socks5ProxyAgent: () => new Socks5ProxyAgent('socks5://localhost', { webSocket })
+  Socks5ProxyAgent: () => new Socks5ProxyAgent('socks5://localhost', { webSocket }),
+  // Wrappers report the options of the dispatcher they wrap.
+  RetryAgent: () => new RetryAgent(new Agent({ webSocket })),
+  MockAgent: () => new MockAgent({ webSocket }),
+  'MockAgent({ agent })': () => new MockAgent({ agent: new Agent({ webSocket }) }),
+  Dispatcher1Wrapper: () => new Dispatcher1Wrapper(new Agent({ webSocket }))
 }
 
 for (const [name, createDispatcher] of Object.entries(dispatchers)) {