[security] Limit retained message parts

Previously, the receiver could retain one `Buffer` entry per buffered
chunk or message fragment until enough data was parsed or the message
completed. A peer could use many tiny fragments/chunks and make retained
memory scale with retained part count rather than message payload size.

Add configurable `maxBufferedChunks` and `maxFragments` options to bound
the number of retained parts. When either limit is exceeded, emit
a `WS_ERR_TOO_MANY_BUFFERED_PARTS` error and close the connection with
close code 1008.

Signed-off-by: Nadav0077 <18245584+Nadav0077@users.noreply.github.com>

Author: Nadav0077

doc/ws.md

@@ -62,6 +62,7 @@
   - [WS_ERR_UNEXPECTED_MASK](#ws_err_unexpected_mask)
   - [WS_ERR_UNEXPECTED_RSV_1](#ws_err_unexpected_rsv_1)
   - [WS_ERR_UNEXPECTED_RSV_2_3](#ws_err_unexpected_rsv_2_3)
+  - [WS_ERR_TOO_MANY_BUFFERED_PARTS](#ws_err_too_many_buffered_parts)
   - [WS_ERR_UNSUPPORTED_DATA_PAYLOAD_LENGTH](#ws_err_unsupported_data_payload_length)
   - [WS_ERR_UNSUPPORTED_MESSAGE_LENGTH](#ws_err_unsupported_message_length)
 
@@ -86,6 +87,10 @@ This class represents a WebSocket server. It extends the `EventEmitter`.
   - `handleProtocols` {Function} A function which can be used to handle the
     WebSocket subprotocols. See description below.
   - `host` {String} The hostname where to bind the server.
+  - `maxBufferedChunks` {Number} The maximum number of buffered data chunks.
+    Defaults to 1048576. Set to 0 to disable the limit.
+  - `maxFragments` {Number} The maximum number of fragments in a message.
+    Defaults to 131072. Set to 0 to disable the limit.
   - `maxPayload` {Number} The maximum allowed message size in bytes. Defaults to
     100 MiB (104857600 bytes).
   - `noServer` {Boolean} Enable no server mode.
@@ -320,6 +325,10 @@ This class represents a WebSocket. It extends the `EventEmitter`.
     cryptographically strong random bytes.
   - `handshakeTimeout` {Number} Timeout in milliseconds for the handshake
     request. This is reset after every redirection.
+  - `maxBufferedChunks` {Number} The maximum number of buffered data chunks.
+    Defaults to 1048576. Set to 0 to disable the limit.
+  - `maxFragments` {Number} The maximum number of fragments in a message.
+    Defaults to 131072. Set to 0 to disable the limit.
   - `maxPayload` {Number} The maximum allowed message size in bytes. Defaults to
     100 MiB (104857600 bytes).
   - `maxRedirects` {Number} The maximum number of redirects allowed. Defaults
@@ -690,6 +699,11 @@ A WebSocket frame was received with the RSV1 bit set unexpectedly.
 
 A WebSocket frame was received with the RSV2 or RSV3 bit set unexpectedly.
 
+### WS_ERR_TOO_MANY_BUFFERED_PARTS
+
+The configured maximum number of buffered data chunks or message fragments was
+exceeded.
+
 ### WS_ERR_UNSUPPORTED_DATA_PAYLOAD_LENGTH
 
 A data frame was received with a length longer than the max supported length

lib/receiver.js

@@ -40,6 +40,10 @@ class Receiver extends Writable {
    *     extensions
    * @param {Boolean} [options.isServer=false] Specifies whether to operate in
    *     client or server mode
+   * @param {Number} [options.maxBufferedChunks=0] The maximum number of
+   *     buffered data chunks
+   * @param {Number} [options.maxFragments=0] The maximum number of message
+   *     fragments
    * @param {Number} [options.maxPayload=0] The maximum allowed message length
    * @param {Boolean} [options.skipUTF8Validation=false] Specifies whether or
    *     not to skip UTF-8 validation for text and close messages
@@ -54,6 +58,8 @@ class Receiver extends Writable {
     this._binaryType = options.binaryType || BINARY_TYPES[0];
     this._extensions = options.extensions || {};
     this._isServer = !!options.isServer;
+    this._maxBufferedChunks = options.maxBufferedChunks | 0;
+    this._maxFragments = options.maxFragments | 0;
     this._maxPayload = options.maxPayload | 0;
     this._skipUTF8Validation = !!options.skipUTF8Validation;
     this[kWebSocket] = undefined;
@@ -89,6 +95,22 @@ class Receiver extends Writable {
   _write(chunk, encoding, cb) {
     if (this._opcode === 0x08 && this._state == GET_INFO) return cb();
 
+    if (
+      this._maxBufferedChunks > 0 &&
+      this._buffers.length >= this._maxBufferedChunks
+    ) {
+      cb(
+        this.createError(
+          RangeError,
+          'Too many buffered chunks',
+          false,
+          1008,
+          'WS_ERR_TOO_MANY_BUFFERED_PARTS'
+        )
+      );
+      return;
+    }
+
     this._bufferedBytes += chunk.length;
     this._buffers.push(chunk);
     this.startLoop(cb);
@@ -485,6 +507,22 @@ class Receiver extends Writable {
     }
 
     if (data.length) {
+      if (
+        this._maxFragments > 0 &&
+        this._fragments.length >= this._maxFragments
+      ) {
+        const error = this.createError(
+          RangeError,
+          'Too many message fragments',
+          false,
+          1008,
+          'WS_ERR_TOO_MANY_BUFFERED_PARTS'
+        );
+
+        cb(error);
+        return;
+      }
+
       //
       // This message is not compressed so its length is the sum of the payload
       // length of all fragments.
@@ -524,6 +562,22 @@ class Receiver extends Writable {
           return;
         }
 
+        if (
+          this._maxFragments > 0 &&
+          this._fragments.length >= this._maxFragments
+        ) {
+          const error = this.createError(
+            RangeError,
+            'Too many message fragments',
+            false,
+            1008,
+            'WS_ERR_TOO_MANY_BUFFERED_PARTS'
+          );
+
+          cb(error);
+          return;
+        }
+
         this._fragments.push(buf);
       }
 

lib/websocket-server.js

@@ -43,6 +43,10 @@ class WebSocketServer extends EventEmitter {
    *     called
    * @param {Function} [options.handleProtocols] A hook to handle protocols
    * @param {String} [options.host] The hostname where to bind the server
+   * @param {Number} [options.maxBufferedChunks=1048576] The maximum number of
+   *     buffered data chunks
+   * @param {Number} [options.maxFragments=131072] The maximum number of message
+   *     fragments
    * @param {Number} [options.maxPayload=104857600] The maximum allowed message
    *     size
    * @param {Boolean} [options.noServer=false] Enable no server mode
@@ -65,6 +69,8 @@ class WebSocketServer extends EventEmitter {
     options = {
       allowSynchronousEvents: true,
       autoPong: true,
+      maxBufferedChunks: 1024 * 1024,
+      maxFragments: 128 * 1024,
       maxPayload: 100 * 1024 * 1024,
       skipUTF8Validation: false,
       perMessageDeflate: false,
@@ -424,6 +430,8 @@ class WebSocketServer extends EventEmitter {
 
     ws.setSocket(socket, head, {
       allowSynchronousEvents: this.options.allowSynchronousEvents,
+      maxBufferedChunks: this.options.maxBufferedChunks,
+      maxFragments: this.options.maxFragments,
       maxPayload: this.options.maxPayload,
       skipUTF8Validation: this.options.skipUTF8Validation
     });

lib/websocket.js

@@ -201,6 +201,10 @@ class WebSocket extends EventEmitter {
    *     multiple times in the same tick
    * @param {Function} [options.generateMask] The function used to generate the
    *     masking key
+   * @param {Number} [options.maxBufferedChunks=0] The maximum number of
+   *     buffered data chunks
+   * @param {Number} [options.maxFragments=0] The maximum number of message
+   *     fragments
    * @param {Number} [options.maxPayload=0] The maximum allowed message size
    * @param {Boolean} [options.skipUTF8Validation=false] Specifies whether or
    *     not to skip UTF-8 validation for text and close messages
@@ -212,6 +216,8 @@ class WebSocket extends EventEmitter {
       binaryType: this.binaryType,
       extensions: this._extensions,
       isServer: this._isServer,
+      maxBufferedChunks: options.maxBufferedChunks,
+      maxFragments: options.maxFragments,
       maxPayload: options.maxPayload,
       skipUTF8Validation: options.skipUTF8Validation
     });
@@ -640,6 +646,10 @@ module.exports = WebSocket;
  *     masking key
  * @param {Number} [options.handshakeTimeout] Timeout in milliseconds for the
  *     handshake request
+ * @param {Number} [options.maxBufferedChunks=1048576] The maximum number of
+ *     buffered data chunks
+ * @param {Number} [options.maxFragments=131072] The maximum number of message
+ *     fragments
  * @param {Number} [options.maxPayload=104857600] The maximum allowed message
  *     size
  * @param {Number} [options.maxRedirects=10] The maximum number of redirects
@@ -660,6 +670,8 @@ function initAsClient(websocket, address, protocols, options) {
     autoPong: true,
     closeTimeout: CLOSE_TIMEOUT,
     protocolVersion: protocolVersions[1],
+    maxBufferedChunks: 1024 * 1024,
+    maxFragments: 128 * 1024,
     maxPayload: 100 * 1024 * 1024,
     skipUTF8Validation: false,
     perMessageDeflate: true,
@@ -1017,6 +1029,8 @@ function initAsClient(websocket, address, protocols, options) {
     websocket.setSocket(socket, head, {
       allowSynchronousEvents: opts.allowSynchronousEvents,
       generateMask: opts.generateMask,
+      maxBufferedChunks: opts.maxBufferedChunks,
+      maxFragments: opts.maxFragments,
       maxPayload: opts.maxPayload,
       skipUTF8Validation: opts.skipUTF8Validation
     });

test/receiver.test.js

@@ -988,6 +988,85 @@ describe('Receiver', () => {
     });
   });
 
+  it('emits an error if there are too many message fragments (1/2)', (done) => {
+    const receiver = new Receiver({ maxFragments: 2 });
+
+    receiver.on('error', (err) => {
+      assert.ok(err instanceof RangeError);
+      assert.strictEqual(err.code, 'WS_ERR_TOO_MANY_BUFFERED_PARTS');
+      assert.strictEqual(err.message, 'Too many message fragments');
+      assert.strictEqual(err[kStatusCode], 1008);
+      done();
+    });
+
+    receiver.write(
+      Buffer.from([
+        0x02,
+        0x01,
+        0x61, // First non-final binary fragment.
+        0x00,
+        0x01,
+        0x62, // Continuation fragment.
+        0x00,
+        0x01,
+        0x63 // Continuation fragment that exceeds the limit.
+      ])
+    );
+  });
+
+  it('emits an error if there are too many message fragments (2/2)', (done) => {
+    const perMessageDeflate = new PerMessageDeflate();
+    perMessageDeflate.accept([{}]);
+
+    const receiver = new Receiver({
+      extensions: {
+        'permessage-deflate': perMessageDeflate
+      },
+      maxFragments: 1
+    });
+    const fragment1 = Buffer.from('foo');
+    const fragment2 = Buffer.from('bar');
+
+    receiver.on('error', (err) => {
+      assert.ok(err instanceof RangeError);
+      assert.strictEqual(err.code, 'WS_ERR_TOO_MANY_BUFFERED_PARTS');
+      assert.strictEqual(err.message, 'Too many message fragments');
+      assert.strictEqual(err[kStatusCode], 1008);
+      done();
+    });
+
+    perMessageDeflate.compress(fragment1, false, (err, data) => {
+      if (err) return done(err);
+
+      receiver.write(Buffer.from([0x41, data.length]));
+      receiver.write(data);
+
+      perMessageDeflate.compress(fragment2, true, (err, data) => {
+        if (err) return done(err);
+
+        receiver.write(Buffer.from([0x80, data.length]));
+        receiver.write(data);
+      });
+    });
+  });
+
+  it('emits an error if there are too many buffered chunks', (done) => {
+    const receiver = new Receiver({ maxBufferedChunks: 2 });
+
+    receiver.on('error', (err) => {
+      assert.ok(err instanceof RangeError);
+      assert.strictEqual(err.code, 'WS_ERR_TOO_MANY_BUFFERED_PARTS');
+      assert.strictEqual(err.message, 'Too many buffered chunks');
+      assert.strictEqual(err[kStatusCode], 1008);
+      done();
+    });
+
+    receiver.write(Buffer.from([0x82, 0x05]));
+    receiver.write(Buffer.from([0x61]));
+    receiver.write(Buffer.from([0x62]));
+    receiver.write(Buffer.from([0x63]));
+  });
+
   it("honors the 'nodebuffer' binary type", (done) => {
     const receiver = new Receiver();
     const frags = [

test/websocket-server.test.js

@@ -66,11 +66,15 @@ describe('WebSocketServer', () => {
         });
       });
 
-      it('accepts the `maxPayload` option', (done) => {
+      it('accepts the receiver limit options', (done) => {
+        const maxBufferedChunks = 1024;
+        const maxFragments = 512;
         const maxPayload = 20480;
         const wss = new WebSocket.Server(
           {
             perMessageDeflate: true,
+            maxBufferedChunks,
+            maxFragments,
             maxPayload,
             port: 0
           },
@@ -82,6 +86,11 @@ describe('WebSocketServer', () => {
         );
 
         wss.on('connection', (ws) => {
+          assert.strictEqual(
+            ws._receiver._maxBufferedChunks,
+            maxBufferedChunks
+          );
+          assert.strictEqual(ws._receiver._maxFragments, maxFragments);
           assert.strictEqual(ws._receiver._maxPayload, maxPayload);
           assert.strictEqual(
             ws._receiver._extensions['permessage-deflate']._maxPayload,

test/websocket.test.js

@@ -135,7 +135,9 @@ describe('WebSocket', () => {
         assert.strictEqual(count, 2);
       });
 
-      it('accepts the `maxPayload` option', (done) => {
+      it('accepts the receiver limit options', (done) => {
+        const maxBufferedChunks = 1024;
+        const maxFragments = 512;
         const maxPayload = 20480;
         const wss = new WebSocket.Server(
           {
@@ -145,10 +147,17 @@ describe('WebSocket', () => {
           () => {
             const ws = new WebSocket(`ws://localhost:${wss.address().port}`, {
               perMessageDeflate: true,
+              maxBufferedChunks,
+              maxFragments,
               maxPayload
             });
 
             ws.on('open', () => {
+              assert.strictEqual(
+                ws._receiver._maxBufferedChunks,
+                maxBufferedChunks
+              );
+              assert.strictEqual(ws._receiver._maxFragments, maxFragments);
               assert.strictEqual(ws._receiver._maxPayload, maxPayload);
               assert.strictEqual(
                 ws._receiver._extensions['permessage-deflate']._maxPayload,