From 25415b6a003dad7e457c9da553f3c458902ae70e Mon Sep 17 00:00:00 2001 From: James M Snell Date: Thu, 24 Sep 2026 18:29:40 +0000 Subject: [PATCH 1/2] fix(websocket): keep receive limits behind dispatcher wrappers WebSocket and WebSocketStream read maxPayloadSize and maxFragments from the dispatcher's webSocketOptions, which only DispatcherBase provides. For any other dispatcher the parser got undefined and treated it as 0, i.e. no limit. That covered RetryAgent, MockAgent/SnapshotAgent, Dispatcher1Wrapper and custom dispatchers, including when one of them is set as the global dispatcher. A configured limit was ignored, and even the 128 MB default was lost, so a ~128 KB compressed frame could inflate to an unbounded message. - The parser falls back to the documented defaults (131072 fragments, 128 MB) when no value is provided, and passes the resolved limit to permessage-deflate. An explicit 0 still disables a limit. - The defaults live in lib/core/constants.js, shared with DispatcherBase. - RetryAgent, MockAgent and Dispatcher1Wrapper report the wrapped dispatcher's webSocketOptions. Signed-off-by: James M Snell --- lib/core/constants.js | 13 +- lib/dispatcher/dispatcher-base.js | 5 +- lib/dispatcher/dispatcher1-wrapper.js | 7 + lib/dispatcher/retry-agent.js | 6 + lib/mock/mock-agent.js | 6 + lib/web/websocket/receiver.js | 16 +- test/websocket/dispatcher-options.js | 11 +- test/websocket/dispatcher-wrapper-limits.js | 197 ++++++++++++++++++++ 8 files changed, 253 insertions(+), 8 deletions(-) create mode 100644 test/websocket/dispatcher-wrapper-limits.js diff --git a/lib/core/constants.js b/lib/core/constants.js index 000c0194909..ef2d80653c8 100644 --- a/lib/core/constants.js +++ b/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 } diff --git a/lib/dispatcher/dispatcher-base.js b/lib/dispatcher/dispatcher-base.js index d9d15d6441a..97a4db6e321 100644 --- a/lib/dispatcher/dispatcher-base.js +++ b/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 } } diff --git a/lib/dispatcher/dispatcher1-wrapper.js b/lib/dispatcher/dispatcher1-wrapper.js index 53193121ef3..a6d211e1477 100644 --- a/lib/dispatcher/dispatcher1-wrapper.js +++ b/lib/dispatcher/dispatcher1-wrapper.js @@ -104,6 +104,13 @@ class Dispatcher1Wrapper extends Dispatcher { return this.#dispatcher.dispatch(opts, Dispatcher1Wrapper.wrapHandler(handler)) } + // Older undici versions read WebSocket receive limits from the (legacy) + // global dispatcher, which is this wrapper, so report the wrapped + // dispatcher's settings. + get webSocketOptions () { + return this.#dispatcher.webSocketOptions + } + close (...args) { return this.#dispatcher.close(...args) } diff --git a/lib/dispatcher/retry-agent.js b/lib/dispatcher/retry-agent.js index 0b50f22a318..ec4cc9c3ee9 100644 --- a/lib/dispatcher/retry-agent.js +++ b/lib/dispatcher/retry-agent.js @@ -26,6 +26,12 @@ class RetryAgent extends Dispatcher { return this.#agent.dispatch(opts, retry) } + // WebSocket reads its receive limits from the dispatcher it is given, so + // report the wrapped agent's settings rather than none at all. + get webSocketOptions () { + return this.#agent.webSocketOptions + } + close () { return this.#agent.close() } diff --git a/lib/mock/mock-agent.js b/lib/mock/mock-agent.js index 6051d5aab7c..6f8aedbf150 100644 --- a/lib/mock/mock-agent.js +++ b/lib/mock/mock-agent.js @@ -55,6 +55,12 @@ class MockAgent extends Dispatcher { } } + // WebSocket reads its receive limits from the dispatcher it is given, so + // report the settings of the agent that makes real connections. + get webSocketOptions () { + return this[kAgent].webSocketOptions + } + get (origin) { // Normalize origin to handle URL objects and case-insensitive hostnames const normalizedOrigin = normalizeOrigin(origin) diff --git a/lib/web/websocket/receiver.js b/lib/web/websocket/receiver.js index 63a5a2e7a52..a3da3f9e44a 100644 --- a/lib/web/websocket/receiver.js +++ b/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 @@ -48,18 +49,25 @@ class ByteParser extends Writable { /** * @param {import('./websocket').Handler} handler * @param {Map|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 + })) } } diff --git a/test/websocket/dispatcher-options.js b/test/websocket/dispatcher-options.js index cfec9933575..b902412e3aa 100644 --- a/test/websocket/dispatcher-options.js +++ b/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)) { diff --git a/test/websocket/dispatcher-wrapper-limits.js b/test/websocket/dispatcher-wrapper-limits.js new file mode 100644 index 00000000000..2ae35105fc7 --- /dev/null +++ b/test/websocket/dispatcher-wrapper-limits.js @@ -0,0 +1,197 @@ +'use strict' + +// WebSocket and WebSocketStream read their receive limits (maxPayloadSize, +// maxFragments) from the dispatcher's `webSocketOptions`. Wrappers such as +// RetryAgent and MockAgent must report the wrapped agent's limits, and a +// dispatcher that doesn't provide any must get the default limits, not none. +// +// Each test waits for the close code the server observes, so nothing depends +// on timing. If a message gets through instead, the client closes normally +// and the test fails on the close code rather than hanging. + +const { test } = require('node:test') +const { once } = require('node:events') +const { createDeflateRaw, constants } = require('node:zlib') +const { WebSocketServer } = require('ws') +const { + Agent, + Dispatcher1Wrapper, + MockAgent, + RetryAgent, + WebSocket, + WebSocketStream, + getGlobalDispatcher, + setGlobalDispatcher +} = require('../..') + +const webSocket = { maxPayloadSize: 1024, maxFragments: 8 } + +const wrappers = { + RetryAgent: (opts) => new RetryAgent(new Agent(opts)), + MockAgent: (opts) => { + const agent = new MockAgent(opts) + agent.enableNetConnect() + return agent + }, + Dispatcher1Wrapper: (opts) => new Dispatcher1Wrapper(new Agent(opts)) +} + +// A dispatcher that only implements `dispatch`, so it has no webSocketOptions. +function customDispatcher (t) { + const agent = new Agent() + t.after(() => agent.close()) + return { dispatch: (opts, handler) => agent.dispatch(opts, handler) } +} + +// Starts a server that runs `send(ws)` for each connection and resolves +// `closed` with the close code and reason it receives. +async function startServer (t, send) { + const server = new WebSocketServer({ port: 0, perMessageDeflate: true }) + t.after(() => server.close()) + await once(server, 'listening') + + const closed = new Promise((resolve) => { + server.on('connection', (ws) => { + ws.on('close', (code, reason) => resolve({ code, reason: reason.toString() })) + send(ws) + }) + }) + + return { url: `ws://127.0.0.1:${server.address().port}`, closed } +} + +// Resolves with the size of the first message, closing the connection. +function connect (t, url, dispatcher) { + const ws = new WebSocket(url, { dispatcher }) + t.after(() => ws.close()) + return new Promise((resolve) => { + ws.onmessage = (event) => { + resolve(event.data.size) + ws.close() + } + }) +} + +function sendCompressed (size) { + return (ws) => ws.send(Buffer.alloc(size, 0x41), { binary: true, compress: true }) +} + +function sendFragments (count) { + return (ws) => { + const fragment = Buffer.from('a') + for (let i = 0; i < count; i++) { + ws.send(fragment, { binary: true, compress: false, fin: i === count - 1 }) + } + } +} + +// One compressed frame that inflates to just over the 128 MB default, while +// only about 128 KB goes over the wire (a decompression bomb). +function sendDecompressionBomb (ws) { + const deflate = createDeflateRaw() + const chunks = [] + deflate.on('data', (chunk) => chunks.push(chunk)) + const megabyte = Buffer.alloc(1024 * 1024) + for (let i = 0; i < 129; i++) { + deflate.write(megabyte) + } + deflate.flush(constants.Z_SYNC_FLUSH, () => { + // permessage-deflate drops the trailing 00 00 ff ff of the sync flush. + const stream = Buffer.concat(chunks) + const payload = stream.subarray(0, stream.length - 4) + + // Server -> client binary frame, FIN=1, RSV1=1 (compressed), 64-bit length + const header = Buffer.alloc(10) + header[0] = 0xC2 + header[1] = 0x7f + header.writeUInt32BE(0, 2) + header.writeUInt32BE(payload.length, 6) + ws._socket.write(Buffer.concat([header, payload])) + }) +} + +for (const [name, wrap] of Object.entries(wrappers)) { + test(`${name} applies the wrapped agent's maxPayloadSize`, async (t) => { + const dispatcher = wrap({ webSocket }) + t.after(() => dispatcher.close()) + const server = await startServer(t, sendCompressed(64 * 1024)) + + connect(t, server.url, dispatcher) + const { code } = await server.closed + t.assert.strictEqual(code, 1009) + }) + + test(`${name} applies the wrapped agent's maxFragments`, async (t) => { + const dispatcher = wrap({ webSocket }) + t.after(() => dispatcher.close()) + const server = await startServer(t, sendFragments(20)) + + connect(t, server.url, dispatcher) + t.assert.deepStrictEqual(await server.closed, { code: 1008, reason: 'Too many message fragments' }) + }) + + test(`${name} delivers messages within the wrapped agent's limits`, async (t) => { + const dispatcher = wrap({ webSocket }) + t.after(() => dispatcher.close()) + const server = await startServer(t, sendFragments(8)) + + t.assert.strictEqual(await connect(t, server.url, dispatcher), 8) + t.assert.strictEqual((await server.closed).code, 1005) + }) +} + +test('a dispatcher without webSocketOptions gets the default maxFragments', async (t) => { + // One more than the default of 131072. + const server = await startServer(t, sendFragments(131073)) + + connect(t, server.url, customDispatcher(t)) + t.assert.deepStrictEqual(await server.closed, { code: 1008, reason: 'Too many message fragments' }) +}) + +test('a dispatcher without webSocketOptions gets the default maxPayloadSize for decompression', async (t) => { + const server = await startServer(t, sendDecompressionBomb) + + connect(t, server.url, customDispatcher(t)) + const { code } = await server.closed + t.assert.strictEqual(code, 1009) +}) + +test('an explicit 0 still disables the limits through a wrapper', async (t) => { + const dispatcher = new RetryAgent(new Agent({ webSocket: { maxPayloadSize: 0, maxFragments: 0 } })) + t.after(() => dispatcher.close()) + // More fragments than the default limit, so a 0 replaced by the default + // would fail this. + const server = await startServer(t, sendFragments(131073)) + + t.assert.strictEqual(await connect(t, server.url, dispatcher), 131073) + t.assert.strictEqual((await server.closed).code, 1005) +}) + +test('WebSocketStream applies the limits of a wrapped global dispatcher', async (t) => { + // WebSocketStream doesn't take a dispatcher option; it uses the global one. + const previous = getGlobalDispatcher() + const dispatcher = new RetryAgent(new Agent({ webSocket })) + setGlobalDispatcher(dispatcher) + t.after(async () => { + setGlobalDispatcher(previous) + await dispatcher.close() + }) + + const server = await startServer(t, sendFragments(20)) + const wss = new WebSocketStream(server.url) + // Observe both promises right away: `closed` rejects when the connection + // fails, and an unobserved rejection would fail the test process. + const settled = Promise.allSettled([wss.opened, wss.closed]) + + // Read the first message; if it arrives, close so the test fails on the code. + wss.opened.then(async ({ readable }) => { + const reader = readable.getReader() + const { done } = await reader.read() + if (!done) { + wss.close() + } + }).catch(() => {}) + + t.assert.deepStrictEqual(await server.closed, { code: 1008, reason: 'Too many message fragments' }) + await settled +}) From 11e3107847db4e2975878188c2602a99ceba88c4 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Thu, 24 Sep 2026 18:31:26 +0000 Subject: [PATCH 2/2] fix(eventsource): keep maxEventSize behind dispatcher wrappers EventSource reads maxEventSize from the dispatcher's eventSourceOptions, which only DispatcherBase provides. Behind RetryAgent, MockAgent/ SnapshotAgent or Dispatcher1Wrapper the configured limit was ignored and the default (buffer.kStringMaxLength) applied instead. Forward eventSourceOptions to the wrapped dispatcher, as is now done for webSocketOptions. Signed-off-by: James M Snell --- lib/dispatcher/dispatcher1-wrapper.js | 10 ++- lib/dispatcher/retry-agent.js | 8 ++- lib/mock/mock-agent.js | 9 ++- .../eventsource-dispatcher-options.js | 72 ++++++++++++++++++- 4 files changed, 91 insertions(+), 8 deletions(-) diff --git a/lib/dispatcher/dispatcher1-wrapper.js b/lib/dispatcher/dispatcher1-wrapper.js index a6d211e1477..87e60c40a6a 100644 --- a/lib/dispatcher/dispatcher1-wrapper.js +++ b/lib/dispatcher/dispatcher1-wrapper.js @@ -104,13 +104,17 @@ class Dispatcher1Wrapper extends Dispatcher { return this.#dispatcher.dispatch(opts, Dispatcher1Wrapper.wrapHandler(handler)) } - // Older undici versions read WebSocket receive limits from the (legacy) - // global dispatcher, which is this wrapper, so report the wrapped - // dispatcher's settings. + // 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) } diff --git a/lib/dispatcher/retry-agent.js b/lib/dispatcher/retry-agent.js index ec4cc9c3ee9..e0374f80c2d 100644 --- a/lib/dispatcher/retry-agent.js +++ b/lib/dispatcher/retry-agent.js @@ -26,12 +26,16 @@ class RetryAgent extends Dispatcher { return this.#agent.dispatch(opts, retry) } - // WebSocket reads its receive limits from the dispatcher it is given, so - // report the wrapped agent's settings rather than none at all. + // 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() } diff --git a/lib/mock/mock-agent.js b/lib/mock/mock-agent.js index 6f8aedbf150..3868415698e 100644 --- a/lib/mock/mock-agent.js +++ b/lib/mock/mock-agent.js @@ -55,12 +55,17 @@ class MockAgent extends Dispatcher { } } - // WebSocket reads its receive limits from the dispatcher it is given, so - // report the settings of the agent that makes real connections. + // 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) diff --git a/test/eventsource/eventsource-dispatcher-options.js b/test/eventsource/eventsource-dispatcher-options.js index 4d98c8f3fcd..421bc88dec6 100644 --- a/test/eventsource/eventsource-dispatcher-options.js +++ b/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') + }) +}