'use strict'; const safeBuffer = require('safe-buffer'); const Limiter = require('async-limiter'); const zlib = require('zlib'); const bufferUtil = require('./buffer-util'); const Buffer = safeBuffer.Buffer; const TRAILER = Buffer.from([0x00, 0x00, 0xff, 0xff]); const EMPTY_BLOCK = Buffer.from([0x00]); const kWriteInProgress = Symbol('write-in-progress'); const kPendingClose = Symbol('pending-close'); const kTotalLength = Symbol('total-length'); const kCallback = Symbol('callback'); const kBuffers = Symbol('buffers'); const kError = Symbol('error'); const kOwner = Symbol('owner'); // // We limit zlib concurrency, which prevents severe memory fragmentation // as documented in https://github.com/nodejs/node/issues/8871#issuecomment-250915913 // and https://github.com/websockets/ws/issues/1202 // // Intentionally global; it's the global thread pool that's an issue. // let zlibLimiter; /** * permessage-deflate implementation. */ class PerMessageDeflate { /** * Creates a PerMessageDeflate instance. * * @param {Object} options Configuration options * @param {Boolean} options.serverNoContextTakeover Request/accept disabling * of server context takeover * @param {Boolean} options.clientNoContextTakeover Advertise/acknowledge * disabling of client context takeover * @param {(Boolean|Number)} options.serverMaxWindowBits Request/confirm the * use of a custom server window size * @param {(Boolean|Number)} options.clientMaxWindowBits Advertise support * for, or request, a custom client window size * @param {Object} options.zlibDeflateOptions Options to pass to zlib on deflate * @param {Object} options.zlibInflateOptions Options to pass to zlib on inflate * @param {Number} options.threshold Size (in bytes) below which messages * should not be compressed * @param {Number} options.concurrencyLimit The number of concurrent calls to * zlib * @param {Boolean} isServer Create the instance in either server or client * mode * @param {Number} maxPayload The maximum allowed message length */ constructor (options, isServer, maxPayload) { this._maxPayload = maxPayload | 0; this._options = options || {}; this._threshold = this._options.threshold !== undefined ? this._options.threshold : 1024; this._isServer = !!isServer; this._deflate = null; this._inflate = null; this.params = null; if (!zlibLimiter) { const concurrency = this._options.concurrencyLimit !== undefined ? this._options.concurrencyLimit : 10; zlibLimiter = new Limiter({ concurrency }); } } /** * @type {String} */ static get extensionName () { return 'permessage-deflate'; } /** * Create an extension negotiation offer. * * @return {Object} Extension parameters * @public */ offer () { const params = {}; if (this._options.serverNoContextTakeover) { params.server_no_context_takeover = true; } if (this._options.clientNoContextTakeover) { params.client_no_context_takeover = true; } if (this._options.serverMaxWindowBits) { params.server_max_window_bits = this._options.serverMaxWindowBits; } if (this._options.clientMaxWindowBits) { params.client_max_window_bits = this._options.clientMaxWindowBits; } else if (this._options.clientMaxWindowBits == null) { params.client_max_window_bits = true; } return params; } /** * Accept an extension negotiation offer/response. * * @param {Array} configurations The extension negotiation offers/reponse * @return {Object} Accepted configuration * @public */ accept (configurations) { configurations = this.normalizeParams(configurations); this.params = this._isServer ? this.acceptAsServer(configurations) : this.acceptAsClient(configurations); return this.params; } /** * Releases all resources used by the extension. * * @public */ cleanup () { if (this._inflate) { if (this._inflate[kWriteInProgress]) { this._inflate[kPendingClose] = true; } else { this._inflate.close(); this._inflate = null; } } if (this._deflate) { if (this._deflate[kWriteInProgress]) { this._deflate[kPendingClose] = true; } else { this._deflate.close(); this._deflate = null; } } } /** * Accept an extension negotiation offer. * * @param {Array} offers The extension negotiation offers * @return {Object} Accepted configuration * @private */ acceptAsServer (offers) { const opts = this._options; const accepted = offers.find((params) => { if ( (opts.serverNoContextTakeover === false && params.server_no_context_takeover) || (params.server_max_window_bits && (opts.serverMaxWindowBits === false || (typeof opts.serverMaxWindowBits === 'number' && opts.serverMaxWindowBits > params.server_max_window_bits))) || (typeof opts.clientMaxWindowBits === 'number' && !params.client_max_window_bits) ) { return false; } return true; }); if (!accepted) { throw new Error('None of the extension offers can be accepted'); } if (opts.serverNoContextTakeover) { accepted.server_no_context_takeover = true; } if (opts.clientNoContextTakeover) { accepted.client_no_context_takeover = true; } if (typeof opts.serverMaxWindowBits === 'number') { accepted.server_max_window_bits = opts.serverMaxWindowBits; } if (typeof opts.clientMaxWindowBits === 'number') { accepted.client_max_window_bits = opts.clientMaxWindowBits; } else if ( accepted.client_max_window_bits === true || opts.clientMaxWindowBits === false ) { delete accepted.client_max_window_bits; } return accepted; } /** * Accept the extension negotiation response. * * @param {Array} response The extension negotiation response * @return {Object} Accepted configuration * @private */ acceptAsClient (response) { const params = response[0]; if ( this._options.clientNoContextTakeover === false && params.client_no_context_takeover ) { throw new Error('Unexpected parameter "client_no_context_takeover"'); } if (!params.client_max_window_bits) { if (typeof this._options.clientMaxWindowBits === 'number') { params.client_max_window_bits = this._options.clientMaxWindowBits; } } else if ( this._options.clientMaxWindowBits === false || (typeof this._options.clientMaxWindowBits === 'number' && params.client_max_window_bits > this._options.clientMaxWindowBits) ) { throw new Error( 'Unexpected or invalid parameter "client_max_window_bits"' ); } return params; } /** * Normalize parameters. * * @param {Array} configurations The extension negotiation offers/reponse * @return {Array} The offers/response with normalized parameters * @private */ normalizeParams (configurations) { configurations.forEach((params) => { Object.keys(params).forEach((key) => { var value = params[key]; if (value.length > 1) { throw new Error(`Parameter "${key}" must have only a single value`); } value = value[0]; if (key === 'client_max_window_bits') { if (value !== true) { const num = +value; if (!Number.isInteger(num) || num < 8 || num > 15) { throw new TypeError( `Invalid value for parameter "${key}": ${value}` ); } value = num; } else if (!this._isServer) { throw new TypeError( `Invalid value for parameter "${key}": ${value}` ); } } else if (key === 'server_max_window_bits') { const num = +value; if (!Number.isInteger(num) || num < 8 || num > 15) { throw new TypeError( `Invalid value for parameter "${key}": ${value}` ); } value = num; } else if ( key === 'client_no_context_takeover' || key === 'server_no_context_takeover' ) { if (value !== true) { throw new TypeError( `Invalid value for parameter "${key}": ${value}` ); } } else { throw new Error(`Unknown parameter "${key}"`); } params[key] = value; }); }); return configurations; } /** * Decompress data. Concurrency limited by async-limiter. * * @param {Buffer} data Compressed data * @param {Boolean} fin Specifies whether or not this is the last fragment * @param {Function} callback Callback * @public */ decompress (data, fin, callback) { zlibLimiter.push((done) => { this._decompress(data, fin, (err, result) => { done(); callback(err, result); }); }); } /** * Compress data. Concurrency limited by async-limiter. * * @param {Buffer} data Data to compress * @param {Boolean} fin Specifies whether or not this is the last fragment * @param {Function} callback Callback * @public */ compress (data, fin, callback) { zlibLimiter.push((done) => { this._compress(data, fin, (err, result) => { done(); callback(err, result); }); }); } /** * Decompress data. * * @param {Buffer} data Compressed data * @param {Boolean} fin Specifies whether or not this is the last fragment * @param {Function} callback Callback * @private */ _decompress (data, fin, callback) { const endpoint = this._isServer ? 'client' : 'server'; if (!this._inflate) { const key = `${endpoint}_max_window_bits`; const windowBits = typeof this.params[key] !== 'number' ? zlib.Z_DEFAULT_WINDOWBITS : this.params[key]; this._inflate = zlib.createInflateRaw( Object.assign( {}, this._options.zlibInflateOptions, { windowBits } ) ); this._inflate[kTotalLength] = 0; this._inflate[kBuffers] = []; this._inflate[kOwner] = this; this._inflate.on('error', inflateOnError); this._inflate.on('data', inflateOnData); } this._inflate[kCallback] = callback; this._inflate[kWriteInProgress] = true; this._inflate.write(data); if (fin) this._inflate.write(TRAILER); this._inflate.flush(() => { const err = this._inflate[kError]; if (err) { this._inflate.close(); this._inflate = null; callback(err); return; } const data = bufferUtil.concat( this._inflate[kBuffers], this._inflate[kTotalLength] ); if ( (fin && this.params[`${endpoint}_no_context_takeover`]) || this._inflate[kPendingClose] ) { this._inflate.close(); this._inflate = null; } else { this._inflate[kWriteInProgress] = false; this._inflate[kTotalLength] = 0; this._inflate[kBuffers] = []; } callback(null, data); }); } /** * Compress data. * * @param {Buffer} data Data to compress * @param {Boolean} fin Specifies whether or not this is the last fragment * @param {Function} callback Callback * @private */ _compress (data, fin, callback) { if (!data || data.length === 0) { process.nextTick(callback, null, EMPTY_BLOCK); return; } const endpoint = this._isServer ? 'server' : 'client'; if (!this._deflate) { const key = `${endpoint}_max_window_bits`; const windowBits = typeof this.params[key] !== 'number' ? zlib.Z_DEFAULT_WINDOWBITS : this.params[key]; this._deflate = zlib.createDeflateRaw( Object.assign( // TODO deprecate memLevel/level and recommend zlibDeflateOptions instead { memLevel: this._options.memLevel, level: this._options.level }, this._options.zlibDeflateOptions, { windowBits } ) ); this._deflate[kTotalLength] = 0; this._deflate[kBuffers] = []; // // `zlib.DeflateRaw` emits an `'error'` event only when an attempt to use // it is made after it has already been closed. This cannot happen here, // so we only add a listener for the `'data'` event. // this._deflate.on('data', deflateOnData); } this._deflate[kWriteInProgress] = true; this._deflate.write(data); this._deflate.flush(zlib.Z_SYNC_FLUSH, () => { var data = bufferUtil.concat( this._deflate[kBuffers], this._deflate[kTotalLength] ); if (fin) data = data.slice(0, data.length - 4); if ( (fin && this.params[`${endpoint}_no_context_takeover`]) || this._deflate[kPendingClose] ) { this._deflate.close(); this._deflate = null; } else { this._deflate[kWriteInProgress] = false; this._deflate[kTotalLength] = 0; this._deflate[kBuffers] = []; } callback(null, data); }); } } module.exports = PerMessageDeflate; /** * The listener of the `zlib.DeflateRaw` stream `'data'` event. * * @param {Buffer} chunk A chunk of data * @private */ function deflateOnData (chunk) { this[kBuffers].push(chunk); this[kTotalLength] += chunk.length; } /** * The listener of the `zlib.InflateRaw` stream `'data'` event. * * @param {Buffer} chunk A chunk of data * @private */ function inflateOnData (chunk) { this[kTotalLength] += chunk.length; if ( this[kOwner]._maxPayload < 1 || this[kTotalLength] <= this[kOwner]._maxPayload ) { this[kBuffers].push(chunk); return; } this[kError] = new RangeError('Max payload size exceeded'); this[kError].closeCode = 1009; this.removeListener('data', inflateOnData); this.reset(); } /** * The listener of the `zlib.InflateRaw` stream `'error'` event. * * @param {Error} err The emitted error * @private */ function inflateOnError (err) { // // There is no need to call `Zlib#close()` as the handle is automatically // closed when an error is emitted. // this[kOwner]._inflate = null; this[kCallback](err); }