chore: build

2026-05-01 22:28:13 +02:00
parent ece977f268
commit 99ce3e4bb3
532 changed files with 5244 additions and 13124 deletions
@@ -379,6 +379,24 @@ class SecureProxyConnectionError extends UndiciError {
  [kSecureProxyConnectionError] = true
 }

+const kMessageSizeExceededError = Symbol.for('undici.error.UND_ERR_WS_MESSAGE_SIZE_EXCEEDED')
+class MessageSizeExceededError extends UndiciError {
+  constructor (message) {
+    super(message)
+    this.name = 'MessageSizeExceededError'
+    this.message = message || 'Max decompressed message size exceeded'
+    this.code = 'UND_ERR_WS_MESSAGE_SIZE_EXCEEDED'
+  }
+
+  static [Symbol.hasInstance] (instance) {
+    return instance && instance[kMessageSizeExceededError] === true
+  }
+
+  get [kMessageSizeExceededError] () {
+    return true
+  }
+}
+
 module.exports = {
  AbortError,
  HTTPParserError,
@@ -402,5 +420,6 @@ module.exports = {
  ResponseExceededMaxSizeError,
  RequestRetryError,
  ResponseError,
-  SecureProxyConnectionError
+  SecureProxyConnectionError,
+  MessageSizeExceededError
 }
@@ -66,6 +66,10 @@ class Request {
      throw new InvalidArgumentError('upgrade must be a string')
    }

+    if (upgrade && !isValidHeaderValue(upgrade)) {
+      throw new InvalidArgumentError('invalid upgrade header')
+    }
+
    if (headersTimeout != null && (!Number.isFinite(headersTimeout) || headersTimeout < 0)) {
      throw new InvalidArgumentError('invalid headersTimeout')
    }
@@ -360,13 +364,19 @@ function processHeader (request, key, val) {
    val = `${val}`
  }

-  if (request.host === null && headerName === 'host') {
+  if (headerName === 'host') {
+    if (request.host !== null) {
+      throw new InvalidArgumentError('duplicate host header')
+    }
    if (typeof val !== 'string') {
      throw new InvalidArgumentError('invalid host header')
    }
    // Consumed by Client
    request.host = val
-  } else if (request.contentLength === null && headerName === 'content-length') {
+  } else if (headerName === 'content-length') {
+    if (request.contentLength !== null) {
+      throw new InvalidArgumentError('duplicate content-length header')
+    }
    request.contentLength = parseInt(val, 10)
    if (!Number.isFinite(request.contentLength)) {
      throw new InvalidArgumentError('invalid content-length header')
@@ -24,7 +24,6 @@ function defaultFactory (origin, opts) {

 class Agent extends DispatcherBase {
  constructor ({ factory = defaultFactory, maxRedirections = 0, connect, ...options } = {}) {
-    super()

    if (typeof factory !== 'function') {
      throw new InvalidArgumentError('factory must be a function.')
@@ -38,6 +37,8 @@ class Agent extends DispatcherBase {
      throw new InvalidArgumentError('maxRedirections must be a positive number')
    }

+    super(options)
+
    if (connect && typeof connect !== 'function') {
      connect = { ...connect }
    }
@@ -106,9 +106,10 @@ class Client extends DispatcherBase {
    autoSelectFamilyAttemptTimeout,
    // h2
    maxConcurrentStreams,
-    allowH2
+    allowH2,
+    webSocket
  } = {}) {
-    super()
+    super({ webSocket })

    if (keepAlive !== undefined) {
      throw new InvalidArgumentError('unsupported keepAlive, use pipelining=0 instead')
@@ -11,15 +11,23 @@ const { kDestroy, kClose, kClosed, kDestroyed, kDispatch, kInterceptors } = requ
 const kOnDestroyed = Symbol('onDestroyed')
 const kOnClosed = Symbol('onClosed')
 const kInterceptedDispatch = Symbol('Intercepted Dispatch')
+const kWebSocketOptions = Symbol('webSocketOptions')

 class DispatcherBase extends Dispatcher {
-  constructor () {
+  constructor (opts) {
    super()

    this[kDestroyed] = false
    this[kOnDestroyed] = null
    this[kClosed] = false
    this[kOnClosed] = []
+    this[kWebSocketOptions] = opts?.webSocket ?? {}
+  }
+
+  get webSocketOptions () {
+    return {
+      maxPayloadSize: this[kWebSocketOptions].maxPayloadSize ?? 128 * 1024 * 1024
+    }
  }

  get destroyed () {
@@ -19,8 +19,8 @@ const kRemoveClient = Symbol('remove client')
 const kStats = Symbol('stats')

 class PoolBase extends DispatcherBase {
-  constructor () {
-    super()
+  constructor (opts) {
+    super(opts)

    this[kQueue] = new FixedQueue()
    this[kClients] = []
@@ -37,8 +37,6 @@ class Pool extends PoolBase {
    allowH2,
    ...options
  } = {}) {
-    super()
-
    if (connections != null && (!Number.isFinite(connections) || connections < 0)) {
      throw new InvalidArgumentError('invalid connections')
    }
@@ -63,6 +61,8 @@ class Pool extends PoolBase {
      })
    }

+    super(options)
+
    this[kInterceptors] = options.interceptors?.Pool && Array.isArray(options.interceptors.Pool)
      ? options.interceptors.Pool
      : []
@@ -2,6 +2,7 @@

 const { createInflateRaw, Z_DEFAULT_WINDOWBITS } = require('node:zlib')
 const { isValidClientWindowBits } = require('./util')
+const { MessageSizeExceededError } = require('../../core/errors')

 const tail = Buffer.from([0x00, 0x00, 0xff, 0xff])
 const kBuffer = Symbol('kBuffer')
@@ -13,17 +14,29 @@ class PerMessageDeflate {

  #options = {}

-  constructor (extensions) {
+  #maxPayloadSize = 0
+
+  /**
+   * @param {Map<string, string>} extensions
+   */
+  constructor (extensions, options) {
    this.#options.serverNoContextTakeover = extensions.has('server_no_context_takeover')
    this.#options.serverMaxWindowBits = extensions.get('server_max_window_bits')
+
+    this.#maxPayloadSize = options.maxPayloadSize
  }

+  /**
+   * Decompress a compressed payload.
+   * @param {Buffer} chunk Compressed data
+   * @param {boolean} fin Final fragment flag
+   * @param {Function} callback Callback function
+   */
  decompress (chunk, fin, callback) {
    // An endpoint uses the following algorithm to decompress a message.
    // 1.  Append 4 octets of 0x00 0x00 0xff 0xff to the tail end of the
    //     payload of the message.
    // 2.  Decompress the resulting data using DEFLATE.
-
    if (!this.#inflate) {
      let windowBits = Z_DEFAULT_WINDOWBITS

@@ -36,13 +49,26 @@ class PerMessageDeflate {
        windowBits = Number.parseInt(this.#options.serverMaxWindowBits)
      }

-      this.#inflate = createInflateRaw({ windowBits })
+      try {
+        this.#inflate = createInflateRaw({ windowBits })
+      } catch (err) {
+        callback(err)
+        return
+      }
      this.#inflate[kBuffer] = []
      this.#inflate[kLength] = 0

      this.#inflate.on('data', (data) => {
-        this.#inflate[kBuffer].push(data)
        this.#inflate[kLength] += data.length
+
+        if (this.#maxPayloadSize > 0 && this.#inflate[kLength] > this.#maxPayloadSize) {
+          callback(new MessageSizeExceededError())
+          this.#inflate.removeAllListeners()
+          this.#inflate = null
+          return
+        }
+
+        this.#inflate[kBuffer].push(data)
      })

      this.#inflate.on('error', (err) => {
@@ -57,6 +83,10 @@ class PerMessageDeflate {
    }

    this.#inflate.flush(() => {
+      if (!this.#inflate) {
+        return
+      }
+
      const full = Buffer.concat(this.#inflate[kBuffer], this.#inflate[kLength])

      this.#inflate[kBuffer].length = 0
@@ -18,6 +18,7 @@ const {
 const { WebsocketFrameSend } = require('./frame')
 const { closeWebSocketConnection } = require('./connection')
 const { PerMessageDeflate } = require('./permessage-deflate')
+const { MessageSizeExceededError } = require('../../core/errors')

 // This code was influenced by ws released under the MIT license.
 // Copyright (c) 2011 Einar Otto Stangvik <einaros@gmail.com>
@@ -26,6 +27,7 @@ const { PerMessageDeflate } = require('./permessage-deflate')

 class ByteParser extends Writable {
  #buffers = []
+  #fragmentsBytes = 0
  #byteOffset = 0
  #loop = false

@@ -37,14 +39,23 @@ class ByteParser extends Writable {
  /** @type {Map<string, PerMessageDeflate>} */
  #extensions

-  constructor (ws, extensions) {
+  /** @type {number} */
+  #maxPayloadSize
+
+  /**
+   * @param {import('./websocket').WebSocket} ws
+   * @param {Map<string, string>|null} extensions
+   * @param {{ maxPayloadSize?: number }} [options]
+   */
+  constructor (ws, extensions, options = {}) {
    super()

    this.ws = ws
    this.#extensions = extensions == null ? new Map() : extensions
+    this.#maxPayloadSize = options.maxPayloadSize ?? 0

    if (this.#extensions.has('permessage-deflate')) {
-      this.#extensions.set('permessage-deflate', new PerMessageDeflate(extensions))
+      this.#extensions.set('permessage-deflate', new PerMessageDeflate(extensions, options))
    }
  }

@@ -60,6 +71,19 @@ class ByteParser extends Writable {
    this.run(callback)
  }

+  #validatePayloadLength () {
+    if (
+      this.#maxPayloadSize > 0 &&
+      !isControlFrame(this.#info.opcode) &&
+      this.#info.payloadLength > this.#maxPayloadSize
+    ) {
+      failWebsocketConnection(this.ws, 'Payload size exceeds maximum allowed size')
+      return false
+    }
+
+    return true
+  }
+
  /**
   * Runs whenever a new chunk is received.
   * Callback is called whenever there are no more chunks buffering,
@@ -148,6 +172,10 @@ class ByteParser extends Writable {
        if (payloadLength <= 125) {
          this.#info.payloadLength = payloadLength
          this.#state = parserStates.READ_DATA
+
+          if (!this.#validatePayloadLength()) {
+            return
+          }
        } else if (payloadLength === 126) {
          this.#state = parserStates.PAYLOADLENGTH_16
        } else if (payloadLength === 127) {
@@ -172,6 +200,10 @@ class ByteParser extends Writable {

        this.#info.payloadLength = buffer.readUInt16BE(0)
        this.#state = parserStates.READ_DATA
+
+        if (!this.#validatePayloadLength()) {
+          return
+        }
      } else if (this.#state === parserStates.PAYLOADLENGTH_64) {
        if (this.#byteOffset < 8) {
          return callback()
@@ -179,6 +211,7 @@ class ByteParser extends Writable {

        const buffer = this.consume(8)
        const upper = buffer.readUInt32BE(0)
+        const lower = buffer.readUInt32BE(4)

        // 2^31 is the maximum bytes an arraybuffer can contain
        // on 32-bit systems. Although, on 64-bit systems, this is
@@ -186,15 +219,17 @@ class ByteParser extends Writable {
        // https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Errors/Invalid_array_length
        // https://source.chromium.org/chromium/chromium/src/+/main:v8/src/common/globals.h;drc=1946212ac0100668f14eb9e2843bdd846e510a1e;bpv=1;bpt=1;l=1275
        // https://source.chromium.org/chromium/chromium/src/+/main:v8/src/objects/js-array-buffer.h;l=34;drc=1946212ac0100668f14eb9e2843bdd846e510a1e
-        if (upper > 2 ** 31 - 1) {
+        if (upper !== 0 || lower > 2 ** 31 - 1) {
          failWebsocketConnection(this.ws, 'Received payload length > 2^31 bytes.')
          return
        }

-        const lower = buffer.readUInt32BE(4)
-
-        this.#info.payloadLength = (upper << 8) + lower
+        this.#info.payloadLength = lower
        this.#state = parserStates.READ_DATA
+
+        if (!this.#validatePayloadLength()) {
+          return
+        }
      } else if (this.#state === parserStates.READ_DATA) {
        if (this.#byteOffset < this.#info.payloadLength) {
          return callback()
@@ -207,42 +242,53 @@ class ByteParser extends Writable {
          this.#state = parserStates.INFO
        } else {
          if (!this.#info.compressed) {
-            this.#fragments.push(body)
+            this.writeFragments(body)
+
+            if (this.#maxPayloadSize > 0 && this.#fragmentsBytes > this.#maxPayloadSize) {
+              failWebsocketConnection(this.ws, new MessageSizeExceededError().message)
+              return
+            }

            // If the frame is not fragmented, a message has been received.
            // If the frame is fragmented, it will terminate with a fin bit set
            // and an opcode of 0 (continuation), therefore we handle that when
            // parsing continuation frames, not here.
            if (!this.#info.fragmented && this.#info.fin) {
-              const fullMessage = Buffer.concat(this.#fragments)
-              websocketMessageReceived(this.ws, this.#info.binaryType, fullMessage)
-              this.#fragments.length = 0
+              websocketMessageReceived(this.ws, this.#info.binaryType, this.consumeFragments())
            }

            this.#state = parserStates.INFO
          } else {
-            this.#extensions.get('permessage-deflate').decompress(body, this.#info.fin, (error, data) => {
-              if (error) {
-                closeWebSocketConnection(this.ws, 1007, error.message, error.message.length)
-                return
-              }
+            this.#extensions.get('permessage-deflate').decompress(
+              body,
+              this.#info.fin,
+              (error, data) => {
+                if (error) {
+                  failWebsocketConnection(this.ws, error.message)
+                  return
+                }

-              this.#fragments.push(data)
+                this.writeFragments(data)
+
+                if (this.#maxPayloadSize > 0 && this.#fragmentsBytes > this.#maxPayloadSize) {
+                  failWebsocketConnection(this.ws, new MessageSizeExceededError().message)
+                  return
+                }
+
+                if (!this.#info.fin) {
+                  this.#state = parserStates.INFO
+                  this.#loop = true
+                  this.run(callback)
+                  return
+                }
+
+                websocketMessageReceived(this.ws, this.#info.binaryType, this.consumeFragments())

-              if (!this.#info.fin) {
-                this.#state = parserStates.INFO
                this.#loop = true
+                this.#state = parserStates.INFO
                this.run(callback)
-                return
              }
-
-              websocketMessageReceived(this.ws, this.#info.binaryType, Buffer.concat(this.#fragments))
-
-              this.#loop = true
-              this.#state = parserStates.INFO
-              this.#fragments.length = 0
-              this.run(callback)
-            })
+            )

            this.#loop = false
            break
@@ -294,6 +340,26 @@ class ByteParser extends Writable {
    return buffer
  }

+  writeFragments (fragment) {
+    this.#fragmentsBytes += fragment.length
+    this.#fragments.push(fragment)
+  }
+
+  consumeFragments () {
+    const fragments = this.#fragments
+
+    if (fragments.length === 1) {
+      this.#fragmentsBytes = 0
+      return fragments.shift()
+    }
+
+    const output = Buffer.concat(fragments, this.#fragmentsBytes)
+    this.#fragments = []
+    this.#fragmentsBytes = 0
+
+    return output
+  }
+
  parseCloseBody (data) {
    assert(data.length !== 1)

@@ -266,6 +266,12 @@ function parseExtensions (extensions) {
 * @param {string} value
 */
 function isValidClientWindowBits (value) {
+  // Must have at least one character
+  if (value.length === 0) {
+    return false
+  }
+
+  // Check all characters are ASCII digits
  for (let i = 0; i < value.length; i++) {
    const byte = value.charCodeAt(i)

@@ -274,7 +280,9 @@ function isValidClientWindowBits (value) {
    }
  }

-  return true
+  // Check numeric range: zlib requires windowBits in range 8-15
+  const num = Number.parseInt(value, 10)
+  return num >= 8 && num <= 15
 }

 // https://nodejs.org/api/intl.html#detecting-internationalization-support
@@ -431,11 +431,15 @@ class WebSocket extends EventTarget {
   * @see https://websockets.spec.whatwg.org/#feedback-from-the-protocol
   */
  #onConnectionEstablished (response, parsedExtensions) {
-    // processResponse is called when the "response’s header list has been received and initialized."
+    // processResponse is called when the "response's header list has been received and initialized."
    // once this happens, the connection is open
    this[kResponse] = response

-    const parser = new ByteParser(this, parsedExtensions)
+    const maxPayloadSize = this[kController]?.dispatcher?.webSocketOptions?.maxPayloadSize
+
+    const parser = new ByteParser(this, parsedExtensions, {
+      maxPayloadSize
+    })
    parser.on('drain', onParserDrain)
    parser.on('error', onParserError.bind(this))