123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492 |
- '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;
- class Receiver extends Writable {
-
- constructor(binaryType, extensions, maxPayload) {
- super();
- this._binaryType = binaryType || BINARY_TYPES[0];
- this[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;
- }
-
- _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);
- }
-
- 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;
- }
-
- 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:
-
- this._loop = false;
- return;
- }
- } while (this._loop);
- cb(err);
- }
-
- 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();
- }
-
- getPayloadLength16() {
- if (this._bufferedBytes < 2) {
- this._loop = false;
- return;
- }
- this._payloadLength = this.consume(2).readUInt16BE(0);
- return this.haveLength();
- }
-
- getPayloadLength64() {
- if (this._bufferedBytes < 8) {
- this._loop = false;
- return;
- }
- const buf = this.consume(8);
- const num = buf.readUInt32BE(0);
-
-
-
-
- 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();
- }
-
- 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;
- }
-
- getMask() {
- if (this._bufferedBytes < 4) {
- this._loop = false;
- return;
- }
- this._mask = this.consume(4);
- this._state = GET_DATA;
- }
-
- getData(cb) {
- var data = EMPTY_BUFFER;
- if (this._payloadLength) {
- if (this._bufferedBytes < this._payloadLength) {
- this._loop = false;
- return;
- }
- data = this.consume(this._payloadLength);
- if (this._masked) 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._messageLength = this._totalPayloadLength;
- this._fragments.push(data);
- }
- return this.dataMessage();
- }
-
- 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);
- });
- }
-
- 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 = concat(fragments, messageLength);
- } else if (this._binaryType === 'arraybuffer') {
- data = toArrayBuffer(concat(fragments, messageLength));
- } else {
- data = fragments;
- }
- this.emit('message', data);
- } else {
- const buf = concat(fragments, messageLength);
- if (!isValidUTF8(buf)) {
- this._loop = false;
- return error(Error, 'invalid UTF-8 sequence', true, 1007);
- }
- this.emit('message', buf.toString());
- }
- }
- this._state = GET_INFO;
- }
-
- 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 (!isValidStatusCode(code)) {
- return error(RangeError, `invalid status code ${code}`, true, 1002);
- }
- const buf = data.slice(2);
- if (!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;
- function error(ErrorCtor, message, prefix, statusCode) {
- const err = new ErrorCtor(
- prefix ? `Invalid WebSocket frame: ${message}` : message
- );
- Error.captureStackTrace(err, error);
- err[kStatusCode] = statusCode;
- return err;
- }
|