twitst4tz

receiver.js (12664B)

---

 'use strict';
 
 const stream = require('stream');
 
 const PerMessageDeflate = require('./permessage-deflate');
 const bufferUtil = require('./buffer-util');
 const validation = require('./validation');
 const constants = require('./constants');
 
 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 stream.Writable
  */
 class Receiver extends stream.Writable {
   /**
    * Creates a Receiver instance.
    *
    * @param {String} binaryType The type for binary data
    * @param {Object} extensions An object containing the negotiated extensions
    * @param {Number} maxPayload The maximum allowed message length
    */
   constructor(binaryType, extensions, maxPayload) {
     super();
 
     this._binaryType = binaryType || constants.BINARY_TYPES[0];
     this[constants.kWebSocket] = undefined;
     this._extensions = extensions || {};
     this._maxPayload = maxPayload | 0;
 
     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
    */
   _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];
 
       if (n >= buf.length) {
         this._buffers.shift().copy(dst, dst.length - n);
       } else {
         buf.copy(dst, dst.length - n, 0, n);
         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) {
     var 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);
     }
 
     const compressed = (buf[0] & 0x40) === 0x40;
 
     if (compressed && !this._extensions[PerMessageDeflate.extensionName]) {
       this._loop = false;
       return error(RangeError, 'RSV1 must be clear', true, 1002);
     }
 
     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);
       }
 
       if (!this._fragmented) {
         this._loop = false;
         return error(RangeError, 'invalid opcode 0', true, 1002);
       }
 
       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);
       }
 
       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);
       }
 
       if (compressed) {
         this._loop = false;
         return error(RangeError, 'RSV1 must be clear', true, 1002);
       }
 
       if (this._payloadLength > 0x7d) {
         this._loop = false;
         return error(
           RangeError,
           `invalid payload length ${this._payloadLength}`,
           true,
           1002
         );
       }
     } else {
       this._loop = false;
       return error(RangeError, `invalid opcode ${this._opcode}`, true, 1002);
     }
 
     if (!this._fin && !this._fragmented) this._fragmented = this._opcode;
     this._masked = (buf[1] & 0x80) === 0x80;
 
     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
       );
     }
 
     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);
       }
     }
 
     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) {
     var data = constants.EMPTY_BUFFER;
 
     if (this._payloadLength) {
       if (this._bufferedBytes < this._payloadLength) {
         this._loop = false;
         return;
       }
 
       data = this.consume(this._payloadLength);
       if (this._masked) bufferUtil.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 lenght 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)
           );
         }
 
         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) {
         var data;
 
         if (this._binaryType === 'nodebuffer') {
           data = toBuffer(fragments, messageLength);
         } else if (this._binaryType === 'arraybuffer') {
           data = toArrayBuffer(toBuffer(fragments, messageLength));
         } else {
           data = fragments;
         }
 
         this.emit('message', data);
       } else {
         const buf = toBuffer(fragments, messageLength);
 
         if (!validation.isValidUTF8(buf)) {
           this._loop = false;
           return error(Error, 'invalid UTF-8 sequence', true, 1007);
         }
 
         this.emit('message', buf.toString());
       }
     }
 
     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, '');
         this.end();
       } else if (data.length === 1) {
         return error(RangeError, 'invalid payload length 1', true, 1002);
       } else {
         const code = data.readUInt16BE(0);
 
         if (!validation.isValidStatusCode(code)) {
           return error(RangeError, `invalid status code ${code}`, true, 1002);
         }
 
         const buf = data.slice(2);
 
         if (!validation.isValidUTF8(buf)) {
           return error(Error, 'invalid UTF-8 sequence', true, 1007);
         }
 
         this.emit('conclude', code, buf.toString());
         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 {(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
  * @return {(Error|RangeError)} The error
  * @private
  */
 function error(ErrorCtor, message, prefix, statusCode) {
   const err = new ErrorCtor(
     prefix ? `Invalid WebSocket frame: ${message}` : message
   );
 
   Error.captureStackTrace(err, error);
   err[constants.kStatusCode] = statusCode;
   return err;
 }
 
 /**
  * Makes a buffer from a list of fragments.
  *
  * @param {Buffer[]} fragments The list of fragments composing the message
  * @param {Number} messageLength The length of the message
  * @return {Buffer}
  * @private
  */
 function toBuffer(fragments, messageLength) {
   if (fragments.length === 1) return fragments[0];
   if (fragments.length > 1) return bufferUtil.concat(fragments, messageLength);
   return constants.EMPTY_BUFFER;
 }
 
 /**
  * Converts a buffer to an `ArrayBuffer`.
  *
  * @param {Buffer} buf The buffer to convert
  * @return {ArrayBuffer} Converted buffer
  */
 function toArrayBuffer(buf) {
   if (buf.byteLength === buf.buffer.byteLength) {
     return buf.buffer;
   }
 
   return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
 }