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>

Author: 2 people

docs/docs/api/Client.md

@@ -25,7 +25,7 @@ Returns: `Client`
 * **maxHeaderSize** `number | null` (optional) - Default: `--max-http-header-size` or `16384` - The maximum length of request headers in bytes. Defaults to Node.js' --max-http-header-size or 16KiB.
 * **maxResponseSize** `number | null` (optional) - Default: `-1` - The maximum length of response body in bytes. Set to `-1` to disable.
 * **webSocket** `WebSocketOptions` (optional) - WebSocket-specific configuration options.
-  * **maxFragments** `number` (optional) - Defailt: `131072` - Maximum number of fragments in a message. Set to 0 to disable the limit.
+  * **maxFragments** `number` (optional) - Default: `131072` - Maximum number of fragments in a message. Set to 0 to disable the limit.
   * **maxPayloadSize** `number` (optional) - Default: `134217728` (128 MB) - Maximum allowed payload size in bytes for WebSocket messages. Applied to uncompressed messages, compressed frame payloads, and decompressed (permessage-deflate) messages. Set to 0 to disable the limit.
 * **pipelining** `number | null` (optional) - Default: `1` - The amount of concurrent requests to be sent over the single TCP/TLS connection according to [RFC7230](https://tools.ietf.org/html/rfc7230#section-6.3.2). Carefully consider your workload and environment before enabling concurrent requests as pipelining may reduce performance if used incorrectly. Pipelining is sensitive to network stack settings as well as head of line blocking caused by e.g. long running requests. Set to `0` to disable keep-alive connections. This option has no effect once HTTP/2 is negotiated — see `maxConcurrentStreams` for the h2 dispatch ceiling.
 * **connect** `ConnectOptions | Function | null` (optional) - Default: `null` - Configures how undici establishes TCP/TLS connections. Accepts two forms:

lib/web/websocket/receiver.js

@@ -39,7 +39,6 @@ class ByteParser extends Writable {
   /** @type {import('./websocket').Handler} */
   #handler
 
-
   /** @type {number} */
   #maxFragments
 
@@ -247,7 +246,7 @@ class ByteParser extends Writable {
           this.#state = parserStates.INFO
         } else {
           if (!this.#info.compressed) {
-            if (body.length && !this.writeFragments(body)) {
+            if (!this.writeFragments(body)) {
               return
             }
 
@@ -271,7 +270,7 @@ class ByteParser extends Writable {
                   return
                 }
 
-                if (data.length && !this.writeFragments(data)) {
+                if (!this.writeFragments(data)) {
                   return
                 }
 

lib/web/websocket/stream/websocketstream.js

@@ -258,7 +258,14 @@ class WebSocketStream {
   #onConnectionEstablished (response, parsedExtensions) {
     this.#handler.socket = response.socket
 
-    const parser = new ByteParser(this.#handler, parsedExtensions)
+    // Get options from dispatcher options
+    const maxFragments = this.#handler.controller.dispatcher?.webSocketOptions?.maxFragments
+    const maxPayloadSize = this.#handler.controller.dispatcher?.webSocketOptions?.maxPayloadSize
+
+    const parser = new ByteParser(this.#handler, parsedExtensions, {
+      maxFragments,
+      maxPayloadSize
+    })
     parser.on('drain', () => this.#handler.onParserDrain())
     parser.on('error', (err) => this.#handler.onParserError(err))
 

test/websocket/fragments.js

@@ -43,6 +43,15 @@ test('Fragmented frame with a ping frame in the middle of it', (t) => {
 test('Too many fragments (uncompressed)', (t, done) => {
   t.plan(4)
 
+  function maybeDone () {
+    if (++maybeDone.callCount === 2) {
+      agent.close()
+      server.close(done)
+    }
+  }
+
+  maybeDone.callCount = 0
+
   const agent = new Agent({
     webSocket: {
       maxFragments: 3
@@ -61,15 +70,15 @@ test('Too many fragments (uncompressed)', (t, done) => {
 
     client.addEventListener('close', (event) => {
       t.assert.deepStrictEqual(event.code, 1006)
+      maybeDone()
     })
   })
 
   server.on('connection', (ws) => {
     ws.on('close', (code, reason) => {
       t.assert.deepStrictEqual(code, 1008)
       t.assert.deepStrictEqual(reason.toString(), 'Too many message fragments')
-      agent.close()
-      server.close(done)
+      maybeDone()
     })
 
     const fragment = Buffer.from('a')
@@ -85,6 +94,15 @@ test('Too many fragments (uncompressed)', (t, done) => {
 test('Too many fragments (compressed)', (t, done) => {
   t.plan(4)
 
+  function maybeDone () {
+    if (++maybeDone.callCount === 2) {
+      agent.close()
+      server.close(done)
+    }
+  }
+
+  maybeDone.callCount = 0
+
   const agent = new Agent({
     webSocket: {
       maxFragments: 3
@@ -106,15 +124,15 @@ test('Too many fragments (compressed)', (t, done) => {
 
     client.addEventListener('close', (event) => {
       t.assert.deepStrictEqual(event.code, 1006)
+      maybeDone()
     })
   })
 
   server.on('connection', (ws) => {
     ws.on('close', (code, reason) => {
       t.assert.deepStrictEqual(code, 1008)
       t.assert.deepStrictEqual(reason.toString(), 'Too many message fragments')
-      agent.close()
-      server.close(done)
+      maybeDone()
     })
 
     const fragment = Buffer.from('a')
@@ -126,3 +144,87 @@ test('Too many fragments (compressed)', (t, done) => {
     ws.send(fragment, options)
   })
 })
+
+test('Empty first fragment followed by non-empty continuation delivers the message', (t) => {
+  // RFC 6455 §5.4 allows zero-byte fragments. A conforming server that opens
+  // a fragmented message with an empty frame must be honored: the parser must
+  // recognize the in-progress fragmented message when the continuation arrives.
+  const server = new WebSocketServer({ port: 0 })
+
+  server.on('connection', (ws) => {
+    ws.send('', { fin: false })  // Text frame fin=0, len=0
+    ws.send('hello', { fin: true }) // Continuation fin=1, "hello"
+  })
+
+  after(() => {
+    for (const client of server.clients) {
+      client.close()
+    }
+
+    server.close()
+  })
+
+  const ws = new WebSocket(`ws://localhost:${server.address().port}`)
+
+  return new Promise((resolve) => {
+    ws.addEventListener('message', ({ data }) => {
+      t.assert.strictEqual(data, 'hello')
+
+      ws.close()
+      resolve()
+    })
+  })
+})
+
+test('Too many empty fragments triggers close 1008', (t, done) => {
+  // Empty fragments must still count toward maxFragments; otherwise a
+  // peer can flood zero-byte continuation frames forever.
+  t.plan(4)
+
+  function maybeDone () {
+    if (++maybeDone.callCount === 2) {
+      agent.close()
+      server.close(done)
+    }
+  }
+
+  maybeDone.callCount = 0
+
+  const agent = new Agent({
+    webSocket: {
+      maxFragments: 3
+    }
+  })
+
+  const server = new WebSocketServer({ port: 0 }, () => {
+    const { port } = server.address()
+    const client = new WebSocket(`ws://127.0.0.1:${port}`, {
+      dispatcher: agent
+    })
+
+    client.addEventListener('error', (event) => {
+      t.assert.ok(true)
+    })
+
+    client.addEventListener('close', (event) => {
+      t.assert.deepStrictEqual(event.code, 1006)
+      maybeDone()
+    })
+  })
+
+  server.on('connection', (ws) => {
+    ws.on('close', (code, reason) => {
+      t.assert.deepStrictEqual(code, 1008)
+      t.assert.deepStrictEqual(reason.toString(), 'Too many message fragments')
+      maybeDone()
+    })
+
+    const fragment = ''
+    const options = { fin: false }
+
+    ws.send(fragment, options) // Text frame fin=0, len=0
+    ws.send(fragment, options) // Continuation fin=0, len=0
+    ws.send(fragment, options) // Continuation fin=0, len=0
+    ws.send(fragment, options) // Continuation fin=0, len=0
+  })
+})

test/websocket/stream/too-many-fragments.js

@@ -0,0 +1,58 @@
+'use strict'
+
+const { test } = require('node:test')
+const { WebSocketServer } = require('ws')
+const {
+  Agent,
+  WebSocketStream,
+  getGlobalDispatcher,
+  setGlobalDispatcher
+} = require('../../..')
+
+test('Too many fragments via WebSocketStream triggers close 1008', async (t) => {
+  // WebSocketStream reads its dispatcher from the global one (its options
+  // dictionary doesn't accept a dispatcher), so we swap it for the duration
+  // of this test. The fragment-count limit must apply to WebSocketStream
+  // the same way it applies to the WebSocket API.
+  const previous = getGlobalDispatcher()
+  const agent = new Agent({
+    webSocket: {
+      maxFragments: 3
+    }
+  })
+  setGlobalDispatcher(agent)
+
+  const server = new WebSocketServer({ port: 0 })
+
+  t.after(async () => {
+    setGlobalDispatcher(previous)
+    await agent.close()
+    server.close()
+  })
+
+  const serverClose = new Promise((resolve) => {
+    server.on('connection', (ws) => {
+      ws.on('close', (code, reason) => {
+        resolve({ code, reason: reason.toString() })
+      })
+
+      const fragment = Buffer.from('a')
+      const options = { fin: false }
+
+      ws.send(fragment, options)
+      ws.send(fragment, options)
+      ws.send(fragment, options)
+      ws.send(fragment, options)
+    })
+  })
+
+  const wss = new WebSocketStream(`ws://localhost:${server.address().port}`)
+
+  // The connection will be failed by the parser; both `opened` and `closed`
+  // settle. We only care that the server observed the policy-violation close.
+  await Promise.allSettled([wss.opened, wss.closed])
+
+  const observed = await serverClose
+  t.assert.deepStrictEqual(observed.code, 1008)
+  t.assert.deepStrictEqual(observed.reason, 'Too many message fragments')
+})