tor-browser

receiver.js (14511B)

---

'use strict';

const { Writable } = require('stream');

const PerMessageDeflate = require('./permessage-deflate');
const {
 BINARY_TYPES,
 EMPTY_BUFFER,
 kStatusCode,
 kWebSocket
} = require('./constants');
const { concat, toArrayBuffer, unmask } = require('./buffer-util');
const { isValidStatusCode, isValidUTF8 } = require('./validation');

const GET_INFO = 0;
const GET_PAYLOAD_LENGTH_16 = 1;
const GET_PAYLOAD_LENGTH_64 = 2;
const GET_MASK = 3;
const GET_DATA = 4;
const INFLATING = 5;

/**
* HyBi Receiver implementation.
*
* @extends Writable
*/
class Receiver extends Writable {
 /**
  * Creates a Receiver instance.
  *
  * @param {Object} [options] Options object
  * @param {String} [options.binaryType=nodebuffer] The type for binary data
  * @param {Object} [options.extensions] An object containing the negotiated
  *     extensions
  * @param {Boolean} [options.isServer=false] Specifies whether to operate in
  *     client or server mode
  * @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
  */
 constructor(options = {}) {
   super();

   this._binaryType = options.binaryType || BINARY_TYPES[0];
   this._extensions = options.extensions || {};
   this._isServer = !!options.isServer;
   this._maxPayload = options.maxPayload | 0;
   this._skipUTF8Validation = !!options.skipUTF8Validation;
   this[kWebSocket] = undefined;

   this._bufferedBytes = 0;
   this._buffers = [];

   this._compressed = false;
   this._payloadLength = 0;
   this._mask = undefined;
   this._fragmented = 0;
   this._masked = false;
   this._fin = false;
   this._opcode = 0;

   this._totalPayloadLength = 0;
   this._messageLength = 0;
   this._fragments = [];

   this._state = GET_INFO;
   this._loop = false;
 }

 /**
  * Implements `Writable.prototype._write()`.
  *
  * @param {Buffer} chunk The chunk of data to write
  * @param {String} encoding The character encoding of `chunk`
  * @param {Function} cb Callback
  * @private
  */
 _write(chunk, encoding, cb) {
   if (this._opcode === 0x08 && this._state == GET_INFO) return cb();

   this._bufferedBytes += chunk.length;
   this._buffers.push(chunk);
   this.startLoop(cb);
 }

 /**
  * Consumes `n` bytes from the buffered data.
  *
  * @param {Number} n The number of bytes to consume
  * @return {Buffer} The consumed bytes
  * @private
  */
 consume(n) {
   this._bufferedBytes -= n;

   if (n === this._buffers[0].length) return this._buffers.shift();

   if (n < this._buffers[0].length) {
     const buf = this._buffers[0];
     this._buffers[0] = buf.slice(n);
     return buf.slice(0, n);
   }

   const dst = Buffer.allocUnsafe(n);

   do {
     const buf = this._buffers[0];
     const offset = dst.length - n;

     if (n >= buf.length) {
       dst.set(this._buffers.shift(), offset);
     } else {
       dst.set(new Uint8Array(buf.buffer, buf.byteOffset, n), offset);
       this._buffers[0] = buf.slice(n);
     }

     n -= buf.length;
   } while (n > 0);

   return dst;
 }

 /**
  * Starts the parsing loop.
  *
  * @param {Function} cb Callback
  * @private
  */
 startLoop(cb) {
   let err;
   this._loop = true;

   do {
     switch (this._state) {
       case GET_INFO:
         err = this.getInfo();
         break;
       case GET_PAYLOAD_LENGTH_16:
         err = this.getPayloadLength16();
         break;
       case GET_PAYLOAD_LENGTH_64:
         err = this.getPayloadLength64();
         break;
       case GET_MASK:
         this.getMask();
         break;
       case GET_DATA:
         err = this.getData(cb);
         break;
       default:
         // `INFLATING`
         this._loop = false;
         return;
     }
   } while (this._loop);

   cb(err);
 }

 /**
  * Reads the first two bytes of a frame.
  *
  * @return {(RangeError|undefined)} A possible error
  * @private
  */
 getInfo() {
   if (this._bufferedBytes < 2) {
     this._loop = false;
     return;
   }

   const buf = this.consume(2);

   if ((buf[0] & 0x30) !== 0x00) {
     this._loop = false;
     return error(
       RangeError,
       'RSV2 and RSV3 must be clear',
       true,
       1002,
       'WS_ERR_UNEXPECTED_RSV_2_3'
     );
   }

   const compressed = (buf[0] & 0x40) === 0x40;

   if (compressed && !this._extensions[PerMessageDeflate.extensionName]) {
     this._loop = false;
     return error(
       RangeError,
       'RSV1 must be clear',
       true,
       1002,
       'WS_ERR_UNEXPECTED_RSV_1'
     );
   }

   this._fin = (buf[0] & 0x80) === 0x80;
   this._opcode = buf[0] & 0x0f;
   this._payloadLength = buf[1] & 0x7f;

   if (this._opcode === 0x00) {
     if (compressed) {
       this._loop = false;
       return error(
         RangeError,
         'RSV1 must be clear',
         true,
         1002,
         'WS_ERR_UNEXPECTED_RSV_1'
       );
     }

     if (!this._fragmented) {
       this._loop = false;
       return error(
         RangeError,
         'invalid opcode 0',
         true,
         1002,
         'WS_ERR_INVALID_OPCODE'
       );
     }

     this._opcode = this._fragmented;
   } else if (this._opcode === 0x01 || this._opcode === 0x02) {
     if (this._fragmented) {
       this._loop = false;
       return error(
         RangeError,
         `invalid opcode ${this._opcode}`,
         true,
         1002,
         'WS_ERR_INVALID_OPCODE'
       );
     }

     this._compressed = compressed;
   } else if (this._opcode > 0x07 && this._opcode < 0x0b) {
     if (!this._fin) {
       this._loop = false;
       return error(
         RangeError,
         'FIN must be set',
         true,
         1002,
         'WS_ERR_EXPECTED_FIN'
       );
     }

     if (compressed) {
       this._loop = false;
       return error(
         RangeError,
         'RSV1 must be clear',
         true,
         1002,
         'WS_ERR_UNEXPECTED_RSV_1'
       );
     }

     if (this._payloadLength > 0x7d) {
       this._loop = false;
       return error(
         RangeError,
         `invalid payload length ${this._payloadLength}`,
         true,
         1002,
         'WS_ERR_INVALID_CONTROL_PAYLOAD_LENGTH'
       );
     }
   } else {
     this._loop = false;
     return error(
       RangeError,
       `invalid opcode ${this._opcode}`,
       true,
       1002,
       'WS_ERR_INVALID_OPCODE'
     );
   }

   if (!this._fin && !this._fragmented) this._fragmented = this._opcode;
   this._masked = (buf[1] & 0x80) === 0x80;

   if (this._isServer) {
     if (!this._masked) {
       this._loop = false;
       return error(
         RangeError,
         'MASK must be set',
         true,
         1002,
         'WS_ERR_EXPECTED_MASK'
       );
     }
   } else if (this._masked) {
     this._loop = false;
     return error(
       RangeError,
       'MASK must be clear',
       true,
       1002,
       'WS_ERR_UNEXPECTED_MASK'
     );
   }

   if (this._payloadLength === 126) this._state = GET_PAYLOAD_LENGTH_16;
   else if (this._payloadLength === 127) this._state = GET_PAYLOAD_LENGTH_64;
   else return this.haveLength();
 }

 /**
  * Gets extended payload length (7+16).
  *
  * @return {(RangeError|undefined)} A possible error
  * @private
  */
 getPayloadLength16() {
   if (this._bufferedBytes < 2) {
     this._loop = false;
     return;
   }

   this._payloadLength = this.consume(2).readUInt16BE(0);
   return this.haveLength();
 }

 /**
  * Gets extended payload length (7+64).
  *
  * @return {(RangeError|undefined)} A possible error
  * @private
  */
 getPayloadLength64() {
   if (this._bufferedBytes < 8) {
     this._loop = false;
     return;
   }

   const buf = this.consume(8);
   const num = buf.readUInt32BE(0);

   //
   // The maximum safe integer in JavaScript is 2^53 - 1. An error is returned
   // if payload length is greater than this number.
   //
   if (num > Math.pow(2, 53 - 32) - 1) {
     this._loop = false;
     return error(
       RangeError,
       'Unsupported WebSocket frame: payload length > 2^53 - 1',
       false,
       1009,
       'WS_ERR_UNSUPPORTED_DATA_PAYLOAD_LENGTH'
     );
   }

   this._payloadLength = num * Math.pow(2, 32) + buf.readUInt32BE(4);
   return this.haveLength();
 }

 /**
  * Payload length has been read.
  *
  * @return {(RangeError|undefined)} A possible error
  * @private
  */
 haveLength() {
   if (this._payloadLength && this._opcode < 0x08) {
     this._totalPayloadLength += this._payloadLength;
     if (this._totalPayloadLength > this._maxPayload && this._maxPayload > 0) {
       this._loop = false;
       return error(
         RangeError,
         'Max payload size exceeded',
         false,
         1009,
         'WS_ERR_UNSUPPORTED_MESSAGE_LENGTH'
       );
     }
   }

   if (this._masked) this._state = GET_MASK;
   else this._state = GET_DATA;
 }

 /**
  * Reads mask bytes.
  *
  * @private
  */
 getMask() {
   if (this._bufferedBytes < 4) {
     this._loop = false;
     return;
   }

   this._mask = this.consume(4);
   this._state = GET_DATA;
 }

 /**
  * Reads data bytes.
  *
  * @param {Function} cb Callback
  * @return {(Error|RangeError|undefined)} A possible error
  * @private
  */
 getData(cb) {
   let data = EMPTY_BUFFER;

   if (this._payloadLength) {
     if (this._bufferedBytes < this._payloadLength) {
       this._loop = false;
       return;
     }

     data = this.consume(this._payloadLength);

     if (
       this._masked &&
       (this._mask[0] | this._mask[1] | this._mask[2] | this._mask[3]) !== 0
     ) {
       unmask(data, this._mask);
     }
   }

   if (this._opcode > 0x07) return this.controlMessage(data);

   if (this._compressed) {
     this._state = INFLATING;
     this.decompress(data, cb);
     return;
   }

   if (data.length) {
     //
     // This message is not compressed so its length is the sum of the payload
     // length of all fragments.
     //
     this._messageLength = this._totalPayloadLength;
     this._fragments.push(data);
   }

   return this.dataMessage();
 }

 /**
  * Decompresses data.
  *
  * @param {Buffer} data Compressed data
  * @param {Function} cb Callback
  * @private
  */
 decompress(data, cb) {
   const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];

   perMessageDeflate.decompress(data, this._fin, (err, buf) => {
     if (err) return cb(err);

     if (buf.length) {
       this._messageLength += buf.length;
       if (this._messageLength > this._maxPayload && this._maxPayload > 0) {
         return cb(
           error(
             RangeError,
             'Max payload size exceeded',
             false,
             1009,
             'WS_ERR_UNSUPPORTED_MESSAGE_LENGTH'
           )
         );
       }

       this._fragments.push(buf);
     }

     const er = this.dataMessage();
     if (er) return cb(er);

     this.startLoop(cb);
   });
 }

 /**
  * Handles a data message.
  *
  * @return {(Error|undefined)} A possible error
  * @private
  */
 dataMessage() {
   if (this._fin) {
     const messageLength = this._messageLength;
     const fragments = this._fragments;

     this._totalPayloadLength = 0;
     this._messageLength = 0;
     this._fragmented = 0;
     this._fragments = [];

     if (this._opcode === 2) {
       let data;

       if (this._binaryType === 'nodebuffer') {
         data = concat(fragments, messageLength);
       } else if (this._binaryType === 'arraybuffer') {
         data = toArrayBuffer(concat(fragments, messageLength));
       } else {
         data = fragments;
       }

       this.emit('message', data, true);
     } else {
       const buf = concat(fragments, messageLength);

       if (!this._skipUTF8Validation && !isValidUTF8(buf)) {
         this._loop = false;
         return error(
           Error,
           'invalid UTF-8 sequence',
           true,
           1007,
           'WS_ERR_INVALID_UTF8'
         );
       }

       this.emit('message', buf, false);
     }
   }

   this._state = GET_INFO;
 }

 /**
  * Handles a control message.
  *
  * @param {Buffer} data Data to handle
  * @return {(Error|RangeError|undefined)} A possible error
  * @private
  */
 controlMessage(data) {
   if (this._opcode === 0x08) {
     this._loop = false;

     if (data.length === 0) {
       this.emit('conclude', 1005, EMPTY_BUFFER);
       this.end();
     } else if (data.length === 1) {
       return error(
         RangeError,
         'invalid payload length 1',
         true,
         1002,
         'WS_ERR_INVALID_CONTROL_PAYLOAD_LENGTH'
       );
     } else {
       const code = data.readUInt16BE(0);

       if (!isValidStatusCode(code)) {
         return error(
           RangeError,
           `invalid status code ${code}`,
           true,
           1002,
           'WS_ERR_INVALID_CLOSE_CODE'
         );
       }

       const buf = data.slice(2);

       if (!this._skipUTF8Validation && !isValidUTF8(buf)) {
         return error(
           Error,
           'invalid UTF-8 sequence',
           true,
           1007,
           'WS_ERR_INVALID_UTF8'
         );
       }

       this.emit('conclude', code, buf);
       this.end();
     }
   } else if (this._opcode === 0x09) {
     this.emit('ping', data);
   } else {
     this.emit('pong', data);
   }

   this._state = GET_INFO;
 }
}

module.exports = Receiver;

/**
* Builds an error object.
*
* @param {function(new:Error|RangeError)} ErrorCtor The error constructor
* @param {String} message The error message
* @param {Boolean} prefix Specifies whether or not to add a default prefix to
*     `message`
* @param {Number} statusCode The status code
* @param {String} errorCode The exposed error code
* @return {(Error|RangeError)} The error
* @private
*/
function error(ErrorCtor, message, prefix, statusCode, errorCode) {
 const err = new ErrorCtor(
   prefix ? `Invalid WebSocket frame: ${message}` : message
 );

 Error.captureStackTrace(err, error);
 err.code = errorCode;
 err[kStatusCode] = statusCode;
 return err;
}