[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

@@ -54,6 +54,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)
 
@@ -77,6 +78,10 @@ This class represents a WebSocket server. It extends the `EventEmitter`.
   - `noServer` {Boolean} Enable no server mode.
   - `clientTracking` {Boolean} Specifies whether or not to track clients.
   - `perMessageDeflate` {Boolean|Object} Enable/disable permessage-deflate.
+  - `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.
 - `callback` {Function}
 
@@ -267,6 +272,10 @@ This class represents a WebSocket. It extends the `EventEmitter`.
   - `protocolVersion` {Number} Value of the `Sec-WebSocket-Version` header.
   - `origin` {String} Value of the `Origin` or `Sec-WebSocket-Origin` header
     depending on the `protocolVersion`.
+  - `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.
   - Any other option allowed in [http.request()][] or [https.request()][].
     Options given do not have any effect if parsed from the URL given with the
@@ -547,6 +556,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

@@ -33,14 +33,27 @@ class Receiver extends Writable {
    * @param {Boolean} [isServer=false] Specifies whether to operate in client or
    *     server mode
    * @param {Number} [maxPayload=0] The maximum allowed message length
+   * @param {Number} [maxBufferedChunks=0] The maximum number of
+   *     buffered data chunks
+   * @param {Number} [maxFragments=0] The maximum number of message
+   *     fragments
    */
-  constructor(binaryType, extensions, isServer, maxPayload) {
+  constructor(
+    binaryType,
+    extensions,
+    isServer,
+    maxPayload,
+    maxBufferedChunks,
+    maxFragments
+  ) {
     super();
 
     this._binaryType = binaryType || BINARY_TYPES[0];
     this[kWebSocket] = undefined;
     this._extensions = extensions || {};
     this._isServer = !!isServer;
+    this._maxBufferedChunks = maxBufferedChunks | 0;
+    this._maxFragments = maxFragments | 0;
     this._maxPayload = maxPayload | 0;
 
     this._bufferedBytes = 0;
@@ -73,6 +86,21 @@ 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
+    ) {
+      return cb(
+        error(
+          RangeError,
+          'Too many buffered chunks',
+          false,
+          1008,
+          'WS_ERR_TOO_MANY_BUFFERED_PARTS'
+        )
+      );
+    }
+
     this._bufferedBytes += chunk.length;
     this._buffers.push(chunk);
     this.startLoop(cb);
@@ -424,6 +452,20 @@ class Receiver extends Writable {
     }
 
     if (data.length) {
+      if (
+        this._maxFragments > 0 &&
+        this._fragments.length >= this._maxFragments
+      ) {
+        this._loop = false;
+        return error(
+          RangeError,
+          'Too many message fragments',
+          false,
+          1008,
+          'WS_ERR_TOO_MANY_BUFFERED_PARTS'
+        );
+      }
+
       //
       // This message is not compressed so its lenght is the sum of the payload
       // length of all fragments.
@@ -462,6 +504,21 @@ class Receiver extends Writable {
           );
         }
 
+        if (
+          this._maxFragments > 0 &&
+          this._fragments.length >= this._maxFragments
+        ) {
+          return cb(
+            error(
+              RangeError,
+              'Too many message fragments',
+              false,
+              1008,
+              'WS_ERR_TOO_MANY_BUFFERED_PARTS'
+            )
+          );
+        }
+
         this._fragments.push(buf);
       }
 

lib/websocket-server.js

@@ -36,6 +36,10 @@ class WebSocketServer extends EventEmitter {
    *     track clients
    * @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
@@ -52,6 +56,8 @@ class WebSocketServer extends EventEmitter {
     super();
 
     options = {
+      maxBufferedChunks: 1024 * 1024,
+      maxFragments: 128 * 1024,
       maxPayload: 100 * 1024 * 1024,
       perMessageDeflate: false,
       handleProtocols: null,
@@ -350,7 +356,13 @@ class WebSocketServer extends EventEmitter {
     socket.write(headers.concat('\r\n').join('\r\n'));
     socket.removeListener('error', socketOnError);
 
-    ws.setSocket(socket, head, this.options.maxPayload);
+    ws.setSocket(
+      socket,
+      head,
+      this.options.maxPayload,
+      this.options.maxBufferedChunks,
+      this.options.maxFragments
+    );
 
     if (this.clients) {
       this.clients.add(ws);

lib/websocket.js

@@ -187,14 +187,20 @@ class WebSocket extends EventEmitter {
    *     server and client
    * @param {Buffer} head The first packet of the upgraded stream
    * @param {Number} [maxPayload=0] The maximum allowed message size
+   * @param {Number} [maxBufferedChunks=0] The maximum number of
+   *     buffered data chunks
+   * @param {Number} [maxFragments=0] The maximum number of message
+   *     fragments
    * @private
    */
-  setSocket(socket, head, maxPayload) {
+  setSocket(socket, head, maxPayload, maxBufferedChunks, maxFragments) {
     const receiver = new Receiver(
       this.binaryType,
       this._extensions,
       this._isServer,
-      maxPayload
+      maxPayload,
+      maxBufferedChunks,
+      maxFragments
     );
 
     this._sender = new Sender(socket, this._extensions);
@@ -571,6 +577,10 @@ module.exports = WebSocket;
  *     `Sec-WebSocket-Version` header
  * @param {String} [options.origin] Value of the `Origin` or
  *     `Sec-WebSocket-Origin` header
+ * @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.followRedirects=false] Whether or not to follow
@@ -582,6 +592,8 @@ module.exports = WebSocket;
 function initAsClient(websocket, address, protocols, options) {
   const opts = {
     protocolVersion: protocolVersions[1],
+    maxBufferedChunks: 1024 * 1024,
+    maxFragments: 128 * 1024,
     maxPayload: 100 * 1024 * 1024,
     perMessageDeflate: true,
     followRedirects: false,
@@ -881,7 +893,13 @@ function initAsClient(websocket, address, protocols, options) {
       }
     }
 
-    websocket.setSocket(socket, head, opts.maxPayload);
+    websocket.setSocket(
+      socket,
+      head,
+      opts.maxPayload,
+      opts.maxBufferedChunks,
+      opts.maxFragments
+    );
   });
 }
 

test/receiver.test.js

@@ -955,6 +955,89 @@ describe('Receiver', () => {
     });
   });
 
+  it('emits an error if there are too many message fragments (1/2)', (done) => {
+    const receiver = new Receiver(undefined, {}, false, 0, 0, 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(
+      undefined,
+      {
+        'permessage-deflate': perMessageDeflate
+      },
+      false,
+      0,
+      0,
+      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(undefined, {}, false, 0, 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

@@ -65,11 +65,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
           },
@@ -79,6 +83,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

@@ -57,7 +57,9 @@ describe('WebSocket', () => {
         assert.strictEqual(count, 3);
       });
 
-      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(
           {
@@ -67,10 +69,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,