Backport WebSocket maxPayloadSize fixes to v7.x (#5423) · nodejs/undici@a027a4a · GitHub
Skip to content

Commit a027a4a

Browse files
mcollinaKhafraDevlpincaUlisesGascon
authored
Backport WebSocket maxPayloadSize fixes to v7.x (#5423)
* feat: add configurable maxPayloadSize for WebSocket (#4955) (cherry picked from commit bd91f86) Signed-off-by: Matteo Collina <hello@matteocollina.com> * test: fix flaky permessage-deflate limit timeout (#5229) (cherry picked from commit 9d82667) Signed-off-by: Matteo Collina <hello@matteocollina.com> * fix(websocket): enforce max payload size across fragments Account for previously received fragment bytes when checking WebSocket payload size limits, so fragmented messages cannot exceed maxPayloadSize by splitting the payload across frames. Add coverage for cumulative fragmented payload size enforcement. Co-authored-by: Matthew Aitken <maitken033380023@gmail.com> (cherry picked from commit b4c287b) Signed-off-by: Matteo Collina <hello@matteocollina.com> * websocket: handle empty fragments and stream limits Treat zero-byte frames as real fragments so fragmented messages can start with an empty frame and empty continuations still count toward maxFragments. Pass dispatcher WebSocket limits through to WebSocketStream's parser, add regression coverage for WebSocket and WebSocketStream fragment limits, make the fragment close tests wait for both endpoints, and fix the Client docs typo for maxFragments. Co-authored-by: Ulises Gascon <ulisesgascongonzalez@gmail.com> (cherry picked from commit c5ed787) Signed-off-by: Matteo Collina <hello@matteocollina.com> --------- Signed-off-by: Matteo Collina <hello@matteocollina.com> Co-authored-by: Matthew Aitken <maitken033380023@gmail.com> Co-authored-by: Luigi Pinca <luigipinca@gmail.com> Co-authored-by: Ulises Gascon <ulisesgascongonzalez@gmail.com>
1 parent 8cb10f9 commit a027a4a

11 files changed

Lines changed: 828 additions & 69 deletions

File tree

‎docs/docs/api/Client.md‎

Lines changed: 2 additions & 1 deletion

‎lib/dispatcher/dispatcher-base.js‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ const { kDestroy, kClose, kClosed, kDestroyed, kDispatch } = require('../core/sy
1111

1212
const kOnDestroyed = Symbol('onDestroyed')
1313
const kOnClosed = Symbol('onClosed')
14-
const kWebSocketOptions = Symbol('web socket options')
14+
const kWebSocketOptions = Symbol('webSocketOptions')
1515

1616
class DispatcherBase extends Dispatcher {
1717
/** @type {boolean} */
@@ -27,16 +27,20 @@ class DispatcherBase extends Dispatcher {
2727
[kOnClosed] = null
2828

2929
/**
30-
* @param {{ webSocket?: { maxFragments?: number } }} [opts]
30+
* @param {import('../../types/dispatcher').DispatcherOptions} [opts]
3131
*/
3232
constructor (opts) {
3333
super()
3434
this[kWebSocketOptions] = opts?.webSocket ?? {}
3535
}
3636

37+
/**
38+
* @returns {import('../../types/dispatcher').WebSocketOptions}
39+
*/
3740
get webSocketOptions () {
3841
return {
39-
maxFragments: this[kWebSocketOptions].maxFragments ?? 131072
42+
maxFragments: this[kWebSocketOptions].maxFragments ?? 131072,
43+
maxPayloadSize: this[kWebSocketOptions].maxPayloadSize ?? 128 * 1024 * 1024 // 128 MB default
4044
}
4145
}
4246

‎lib/web/websocket/permessage-deflate.js‎

Lines changed: 13 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -8,40 +8,35 @@ const tail = Buffer.from([0x00, 0x00, 0xff, 0xff])
88
const kBuffer = Symbol('kBuffer')
99
const kLength = Symbol('kLength')
1010

11-
// Default maximum decompressed message size: 4 MB
12-
const kDefaultMaxDecompressedSize = 4 * 1024 * 1024
13-
1411
class PerMessageDeflate {
1512
/** @type {import('node:zlib').InflateRaw} */
1613
#inflate
1714

1815
#options = {}
1916

20-
/** @type {boolean} */
21-
#aborted = false
22-
23-
/** @type {Function|null} */
24-
#currentCallback = null
17+
#maxPayloadSize = 0
2518

2619
/**
2720
* @param {Map<string, string>} extensions
2821
*/
29-
constructor (extensions) {
22+
constructor (extensions, options) {
3023
this.#options.serverNoContextTakeover = extensions.has('server_no_context_takeover')
3124
this.#options.serverMaxWindowBits = extensions.get('server_max_window_bits')
25+
26+
this.#maxPayloadSize = options.maxPayloadSize
3227
}
3328

29+
/**
30+
* Decompress a compressed payload.
31+
* @param {Buffer} chunk Compressed data
32+
* @param {boolean} fin Final fragment flag
33+
* @param {Function} callback Callback function
34+
*/
3435
decompress (chunk, fin, callback) {
3536
// An endpoint uses the following algorithm to decompress a message.
3637
// 1. Append 4 octets of 0x00 0x00 0xff 0xff to the tail end of the
3738
// payload of the message.
3839
// 2. Decompress the resulting data using DEFLATE.
39-
40-
if (this.#aborted) {
41-
callback(new MessageSizeExceededError())
42-
return
43-
}
44-
4540
if (!this.#inflate) {
4641
let windowBits = Z_DEFAULT_WINDOWBITS
4742

@@ -64,23 +59,12 @@ class PerMessageDeflate {
6459
this.#inflate[kLength] = 0
6560

6661
this.#inflate.on('data', (data) => {
67-
if (this.#aborted) {
68-
return
69-
}
70-
7162
this.#inflate[kLength] += data.length
7263

73-
if (this.#inflate[kLength] > kDefaultMaxDecompressedSize) {
74-
this.#aborted = true
64+
if (this.#maxPayloadSize > 0 && this.#inflate[kLength] > this.#maxPayloadSize) {
65+
callback(new MessageSizeExceededError())
7566
this.#inflate.removeAllListeners()
76-
this.#inflate.destroy()
7767
this.#inflate = null
78-
79-
if (this.#currentCallback) {
80-
const cb = this.#currentCallback
81-
this.#currentCallback = null
82-
cb(new MessageSizeExceededError())
83-
}
8468
return
8569
}
8670

@@ -93,22 +77,20 @@ class PerMessageDeflate {
9377
})
9478
}
9579

96-
this.#currentCallback = callback
9780
this.#inflate.write(chunk)
9881
if (fin) {
9982
this.#inflate.write(tail)
10083
}
10184

10285
this.#inflate.flush(() => {
103-
if (this.#aborted || !this.#inflate) {
86+
if (!this.#inflate) {
10487
return
10588
}
10689

10790
const full = Buffer.concat(this.#inflate[kBuffer], this.#inflate[kLength])
10891

10992
this.#inflate[kBuffer].length = 0
11093
this.#inflate[kLength] = 0
111-
this.#currentCallback = null
11294

11395
callback(null, full)
11496
})

‎lib/web/websocket/receiver.js‎

Lines changed: 65 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -42,20 +42,24 @@ class ByteParser extends Writable {
4242
/** @type {number} */
4343
#maxFragments
4444

45+
/** @type {number} */
46+
#maxPayloadSize
47+
4548
/**
4649
* @param {import('./websocket').Handler} handler
4750
* @param {Map<string, string>|null} extensions
48-
* @param {{ maxFragments?: number }} [options]
51+
* @param {{ maxFragments?: number, maxPayloadSize?: number }} [options]
4952
*/
5053
constructor (handler, extensions, options = {}) {
5154
super()
5255

5356
this.#handler = handler
5457
this.#extensions = extensions == null ? new Map() : extensions
5558
this.#maxFragments = options.maxFragments ?? 0
59+
this.#maxPayloadSize = options.maxPayloadSize ?? 0
5660

5761
if (this.#extensions.has('permessage-deflate')) {
58-
this.#extensions.set('permessage-deflate', new PerMessageDeflate(extensions))
62+
this.#extensions.set('permessage-deflate', new PerMessageDeflate(extensions, options))
5963
}
6064
}
6165

@@ -71,6 +75,19 @@ class ByteParser extends Writable {
7175
this.run(callback)
7276
}
7377

78+
#validatePayloadLength () {
79+
if (
80+
this.#maxPayloadSize > 0 &&
81+
!isControlFrame(this.#info.opcode) &&
82+
this.#info.payloadLength + this.#fragmentsBytes > this.#maxPayloadSize
83+
) {
84+
failWebsocketConnection(this.#handler, 1009, 'Payload size exceeds maximum allowed size')
85+
return false
86+
}
87+
88+
return true
89+
}
90+
7491
/**
7592
* Runs whenever a new chunk is received.
7693
* Callback is called whenever there are no more chunks buffering,
@@ -159,6 +176,10 @@ class ByteParser extends Writable {
159176
if (payloadLength <= 125) {
160177
this.#info.payloadLength = payloadLength
161178
this.#state = parserStates.READ_DATA
179+
180+
if (!this.#validatePayloadLength()) {
181+
return
182+
}
162183
} else if (payloadLength === 126) {
163184
this.#state = parserStates.PAYLOADLENGTH_16
164185
} else if (payloadLength === 127) {
@@ -183,6 +204,10 @@ class ByteParser extends Writable {
183204

184205
this.#info.payloadLength = buffer.readUInt16BE(0)
185206
this.#state = parserStates.READ_DATA
207+
208+
if (!this.#validatePayloadLength()) {
209+
return
210+
}
186211
} else if (this.#state === parserStates.PAYLOADLENGTH_64) {
187212
if (this.#byteOffset < 8) {
188213
return callback()
@@ -205,6 +230,10 @@ class ByteParser extends Writable {
205230

206231
this.#info.payloadLength = lower
207232
this.#state = parserStates.READ_DATA
233+
234+
if (!this.#validatePayloadLength()) {
235+
return
236+
}
208237
} else if (this.#state === parserStates.READ_DATA) {
209238
if (this.#byteOffset < this.#info.payloadLength) {
210239
return callback()
@@ -217,7 +246,7 @@ class ByteParser extends Writable {
217246
this.#state = parserStates.INFO
218247
} else {
219248
if (!this.#info.compressed) {
220-
if (body.length && !this.writeFragments(body)) {
249+
if (!this.writeFragments(body)) {
221250
return
222251
}
223252

@@ -231,31 +260,41 @@ class ByteParser extends Writable {
231260

232261
this.#state = parserStates.INFO
233262
} else {
234-
this.#extensions.get('permessage-deflate').decompress(body, this.#info.fin, (error, data) => {
235-
if (error) {
236-
// Use 1009 (Message Too Big) for decompression size limit errors
237-
const code = error instanceof MessageSizeExceededError ? 1009 : 1007
238-
failWebsocketConnection(this.#handler, code, error.message)
239-
return
240-
}
241-
242-
if (data.length && !this.writeFragments(data)) {
243-
return
244-
}
245-
246-
if (!this.#info.fin) {
247-
this.#state = parserStates.INFO
263+
this.#extensions.get('permessage-deflate').decompress(
264+
body,
265+
this.#info.fin,
266+
(error, data) => {
267+
if (error) {
268+
const code = error instanceof MessageSizeExceededError ? 1009 : 1007
269+
failWebsocketConnection(this.#handler, code, error.message)
270+
return
271+
}
272+
273+
if (!this.writeFragments(data)) {
274+
return
275+
}
276+
277+
// Check cumulative fragment size
278+
if (this.#maxPayloadSize > 0 && this.#fragmentsBytes > this.#maxPayloadSize) {
279+
failWebsocketConnection(this.#handler, 1009, new MessageSizeExceededError().message)
280+
return
281+
}
282+
283+
if (!this.#info.fin) {
284+
this.#state = parserStates.INFO
285+
this.#loop = true
286+
this.run(callback)
287+
return
288+
}
289+
290+
websocketMessageReceived(this.#handler, this.#info.binaryType, this.consumeFragments())
291+
248292
this.#loop = true
293+
this.#state = parserStates.INFO
249294
this.run(callback)
250-
return
251-
}
252-
253-
websocketMessageReceived(this.#handler, this.#info.binaryType, this.consumeFragments())
254-
255-
this.#loop = true
256-
this.#state = parserStates.INFO
257-
this.run(callback)
258-
})
295+
},
296+
this.#fragmentsBytes
297+
)
259298

260299
this.#loop = false
261300
break

‎lib/web/websocket/stream/websocketstream.js‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -258,7 +258,14 @@ class WebSocketStream {
258258
#onConnectionEstablished (response, parsedExtensions) {
259259
this.#handler.socket = response.socket
260260

261-
const parser = new ByteParser(this.#handler, parsedExtensions)
261+
// Get options from dispatcher options
262+
const maxFragments = this.#handler.controller.dispatcher?.webSocketOptions?.maxFragments
263+
const maxPayloadSize = this.#handler.controller.dispatcher?.webSocketOptions?.maxPayloadSize
264+
265+
const parser = new ByteParser(this.#handler, parsedExtensions, {
266+
maxFragments,
267+
maxPayloadSize
268+
})
262269
parser.on('drain', () => this.#handler.onParserDrain())
263270
parser.on('error', (err) => this.#handler.onParserError(err))
264271

‎lib/web/websocket/websocket.js‎

Lines changed: 5 additions & 2 deletions

0 commit comments

Comments
 (0)