proxy-packer.js/index.js

'use strict';

var Packer = module.exports;

var serviceEvents = {
  default: 'tunnelData'
, connection: 'tunnelConnection'
, control: 'tunnelControl'
, error:   'tunnelError'
, end:     'tunnelEnd'
, pause:   'tunnelPause'
, resume:  'tunnelResume'
};
var serviceFuncs = {
  default: 'onmessage'
, connection: 'onconnection'
, control: 'oncontrol'
, error:   'onerror'
, end:     'onend'
, pause:   'onpause'
, resume:  'onresume'
};

Packer.create = function (opts) {
  var machine;

  if (!opts.onMessage && !opts.onmessage) {
    machine = new (require('events').EventEmitter)();
  } else {
    machine = {};
  }

  machine.onmessage    = opts.onmessage    || opts.onMessage;
  machine.oncontrol    = opts.oncontrol    || opts.onControl;
  machine.onconnection = opts.onconnection || opts.onConnection || function () {};
  machine.onerror      = opts.onerror      || opts.onError;
  machine.onend        = opts.onend        || opts.onEnd;
  machine.onpause      = opts.onpause      || opts.onPause;
  machine.onresume     = opts.onresume     || opts.onResume;

  machine._version = 1;
  machine.fns = {};

  machine.chunkIndex = 0;
  machine.buf = null;
  machine.bufIndex = 0;
  machine.fns.collectData = function (chunk, size) {
    var chunkLeft = chunk.length - machine.chunkIndex;
    var hasLen = (size > 0);

    if (!hasLen) {
      return Buffer.alloc(0);
    }

    // First handle case where we don't have all the data we need yet. We need to save
    // what we have in a buffer, and increment the index for both the buffer and the chunk.
    if (machine.bufIndex + chunkLeft < size) {
      if (!machine.buf) {
        machine.buf = Buffer.alloc(size);
      }
      chunk.copy(machine.buf, machine.bufIndex, machine.chunkIndex);
      machine.bufIndex += chunkLeft;
      machine.chunkIndex += chunkLeft;

      return null;
    }

    // Read and mark as read however much data we need from the chunk to complete our buffer.
    var partLen = size - machine.bufIndex;
    var part = chunk.slice(machine.chunkIndex, machine.chunkIndex+partLen);
    machine.chunkIndex += partLen;

    // If we had nothing buffered than the part of the chunk we just read is all we need.
    if (!machine.buf) {
      return part;
    }

    // Otherwise we need to copy the new data into the buffer.
    part.copy(machine.buf, machine.bufIndex);
    // Before returning the buffer we need to clear our reference to it.
    var buf = machine.buf;
    machine.buf = null;
    machine.bufIndex = 0;
    return buf;
  };

  machine.fns.version = function (chunk) {
    //console.log('');
    //console.log('[version]');
    if ((255 - machine._version) !== chunk[machine.chunkIndex]) {
      console.error("not v" + machine._version + " (or data is corrupt)");
      // no idea how to fix this yet
    }
    machine.chunkIndex += 1;

    return true;
  };


  machine.headerLen = 0;
  machine.fns.headerLength = function (chunk) {
    //console.log('');
    //console.log('[headerLength]');
    machine.headerLen = chunk[machine.chunkIndex];
    machine.chunkIndex += 1;

    return true;
  };

  machine.fns.header = function (chunk) {
    //console.log('');
    //console.log('[header]');
    var header = machine.fns.collectData(chunk, machine.headerLen);

    // We don't have the entire header yet so return false.
    if (!header) {
      return false;
    }

    machine._headers = header.toString().split(/,/g);

    machine.family      = machine._headers[0];
    machine.address     = machine._headers[1];
    machine.port        = machine._headers[2];
    machine.bodyLen     = parseInt(machine._headers[3], 10) || 0;
    machine.service     = machine._headers[4];
    machine.serviceport = machine._headers[5];
    machine.name        = machine._headers[6];
    //console.log('machine.service', machine.service);

    return true;
  };

  machine.fns.data = function (chunk) {
    //console.log('');
    //console.log('[data]');
    var data;
    // The 'connection' event may not have a body
    // Other events may not have a body either
    if (machine.bodyLen) {
      data = machine.fns.collectData(chunk, machine.bodyLen);
      // We don't have the entire body yet so return false.
      if (!data) {
        return false;
      }
    }


    //
    // data, end, error
    //
    var msg = {};
    if ('error' === machine.service) {
      try {
        msg = JSON.parse(data.toString());
      } catch(e) {
        msg.message = data.toString();
        msg.code = 'E_UNKNOWN_ERR';
      }
    }

    msg.family      = machine.family;
    msg.address     = machine.address;
    msg.port        = machine.port;
    msg.service     = machine.service;
    msg.serviceport = machine.serviceport;
    msg.name        = machine.name;
    msg.data        = data;

    if (machine.emit) {
      machine.emit(serviceEvents[msg.service] || serviceEvents.default);
    } else {
      (machine[serviceFuncs[msg.service]] || machine[serviceFuncs.default])(msg);
    }

    return true;
  };

  machine.state = 0;
  machine.states = ['version', 'headerLength', 'header', 'data'];
  machine.fns.addChunk = function (chunk) {
    //console.log('');
    //console.log('[addChunk]');
    machine.chunkIndex = 0;
    while (machine.chunkIndex < chunk.length) {
      //console.log('chunkIndex:', machine.chunkIndex, 'state:', machine.state);

      if (true === machine.fns[machine.states[machine.state]](chunk)) {
        machine.state += 1;
        machine.state %= machine.states.length;
      }
    }
  };

  return machine;
};

Packer.packHeader = function (meta, data, service, andBody, oldways) {
  if (oldways) {
    data = data || Buffer.from(' ');
  }
  if (data && !Buffer.isBuffer(data)) {
    data = new Buffer(JSON.stringify(data));
  }
  if (oldways && !data.byteLength) {
    data = Buffer.from(' ');
  }

  if (service && service !== 'control') {
    meta.service = service;
  }

  var size = data && data.byteLength || 0;
  var sizeReserve = andBody ? size : 0;
  var version = 1;
  var header;
  if (service === 'control') {
    header = Buffer.from(['', '', '', size, service].join(','));
  }
  else {
    header = Buffer.from([
      meta.family, meta.address, meta.port, size,
      (meta.service || ''), (meta.serviceport || ''), (meta.name || '')
    ].join(','));
  }
  var metaBuf = Buffer.from([ 255 - version, header.length ]);
  var buf = Buffer.alloc(metaBuf.byteLength + header.byteLength + sizeReserve);

  metaBuf.copy(buf, 0);
  header.copy(buf, 2);
  if (sizeReserve) { data.copy(buf, 2 + header.byteLength); }

  return buf;
};
Packer.pack = function (meta, data, service) {
  return Packer.packHeader(meta, data, service, true, true);
};

function extractSocketProps(socket, propNames) {
  var props = {};

  if (socket.remotePort) {
    propNames.forEach(function (propName) {
      props[propName] = socket[propName];
    });
  } else if (socket._remotePort) {
    propNames.forEach(function (propName) {
      props[propName] = socket['_' + propName];
    });
  } else if (
    socket._handle
    && socket._handle._parent
    && socket._handle._parent.owner
    && socket._handle._parent.owner.stream
    && socket._handle._parent.owner.stream.remotePort
  ) {
    propNames.forEach(function (propName) {
      props[propName] = socket._handle._parent.owner.stream[propName];
    });
  } else if (
    socket._handle._parentWrap
    && socket._handle._parentWrap
    && socket._handle._parentWrap.remotePort
  ) {
    propNames.forEach(function (propName) {
      props[propName] = socket._handle._parentWrap[propName];
    });
  } else if (
    socket._handle._parentWrap
    && socket._handle._parentWrap._handle
    && socket._handle._parentWrap._handle.owner
    && socket._handle._parentWrap._handle.owner.stream
    && socket._handle._parentWrap._handle.owner.stream.remotePort
  ) {
    propNames.forEach(function (propName) {
      props[propName] = socket._handle._parentWrap._handle.owner.stream[propName];
    });
  }
  return props;
}
function extractSocketProp(socket, propName) {
  // remoteAddress, remotePort... ugh... https://github.com/nodejs/node/issues/8854
  var value = socket[propName] || socket['_' + propName];
  try {
    value = value || socket._handle._parent.owner.stream[propName];
  } catch (e) {}

  try {
    value = value || socket._handle._parentWrap[propName];
    value = value || socket._handle._parentWrap._handle.owner.stream[propName];
  } catch (e) {}

  return value || '';
}
Packer.socketToAddr = function (socket) {
  // TODO BUG XXX
  // https://github.com/nodejs/node/issues/8854
  // tlsSocket.remoteAddress = remoteAddress; // causes core dump
  // console.log(tlsSocket.remoteAddress);

  var props = extractSocketProps(socket, [ 'remoteFamily', 'remoteAddress', 'remotePort', 'localPort' ]);
  return {
    family:  props.remoteFamily
  , address: props.remoteAddress
  , port:    props.remotePort
  , serviceport: props.localPort
  };
};

Packer.addrToId = function (address) {
  return address.family + ',' + address.address + ',' + address.port;
};

Packer.socketToId = function (socket) {
  return Packer.addrToId(Packer.socketToAddr(socket));
};


var addressNames = [
  'remoteAddress'
, 'remotePort'
, 'remoteFamily'
, 'localAddress'
, 'localPort'
];
var sockFuncs = [
  'address'
, 'destroy'
, 'ref'
, 'unref'
, 'setEncoding'
, 'setKeepAlive'
, 'setNoDelay'
, 'setTimeout'
];
// Unlike Packer.Stream.create this should handle all of the events needed to make everything work.
Packer.wrapSocket = function (socket) {
  // node v10.2+ doesn't need a workaround for  https://github.com/nodejs/node/issues/8854
  addressNames.forEach(function (name) {
    Object.defineProperty(socket, name, {
      enumerable: false,
      configurable: true,
      get: function() {
        return extractSocketProp(socket, name);
      }
    });
  });
  return socket;
  // Improved workaround for  https://github.com/nodejs/node/issues/8854
  /*
  // TODO use defineProperty to override remotePort, etc
  var myDuplex = new require('stream').Duplex();
  addressNames.forEach(function (name) {
    myDuplex[name] = extractSocketProp(socket, name);
  });

  // Handle everything needed for the write part of the Duplex. We need to overwrite the
  // `end` function because there is no other way to know when the other side calls it.
  myDuplex._write = socket.write.bind(socket);
  myDuplex.end = socket.end.bind(socket);

  // Handle everything needed for the read part of the Duplex. See the example under
  // https://nodejs.org/api/stream.html#stream_readable_push_chunk_encoding.
  myDuplex._read = function () {
    socket.resume();
  };
  socket.on('data', function (chunk) {
    if (!myDuplex.push(chunk)) {
      socket.pause();
    }
  });
  socket.on('end', function () {
    myDuplex.push(null);
  });

  // Handle the the things not directly related to reading or writing
  socket.on('error', function (err) {
    console.error('[error] wrapped socket errored - ' + err.toString());
    myDuplex.emit('error', err);
  });
  socket.on('close', function () {
    myDuplex.emit('close');
  });
  sockFuncs.forEach(function (name) {
    if (typeof socket[name] !== 'function') {
      console.warn('expected `'+name+'` to be a function on wrapped socket');
    } else {
      myDuplex[name] = socket[name].bind(socket);
    }
  });

  return myDuplex;
  */
};

var Transform = require('stream').Transform;
var util = require('util');

function MyTransform(options) {
  if (!(this instanceof MyTransform)) {
    return new MyTransform(options);
  }
  this.__my_addr = options.address;
  this.__my_service = options.service;
  this.__my_serviceport = options.serviceport;
  this.__my_name = options.name;
  Transform.call(this, options);
}
util.inherits(MyTransform, Transform);

MyTransform.prototype._transform = function (data, encoding, callback) {
  var address = this.__my_addr;

  address.service = address.service || this.__my_service;
  address.serviceport = address.serviceport || this.__my_serviceport;
  address.name = address.name || this.__my_name;
  this.push(Packer.pack(address, data));
  callback();
};

Packer.Stream = {};
var Dup = {
  write: function (chunk, encoding, cb) {
    //console.log('_write', chunk.byteLength);
    this.__my_socket.write(chunk, encoding, cb);
  }
, read: function (size) {
    //console.log('_read');
    var x = this.__my_socket.read(size);
    if (x) {
      console.log('_read', size);
      this.push(x);
    }
  }
};
Packer.Stream.create = function (socket) {
  if (!Packer.Stream.warned) {
    console.warn('`Stream.create` deprecated, use `wrapSocket` instead');
    Packer.Stream.warned = true;
  }

  // Workaround for
  // https://github.com/nodejs/node/issues/8854

  // https://www.google.com/#q=get+socket+address+from+file+descriptor
  // TODO try creating a new net.Socket({ handle: socket._handle, fd: socket._handle.fd })
  // from the old one and then adding back the data with
  // sock.push(firstChunk)
  var Duplex = require('stream').Duplex;
  var myDuplex = new Duplex();

  myDuplex.__my_socket = socket;
  myDuplex._write = Dup.write;
  myDuplex._read = Dup.read;
  //console.log('plainSocket.*Address');
  //console.log('remote:', socket.remoteAddress);
  //console.log('local:', socket.localAddress);
  //console.log('address():', socket.address());
  myDuplex.remoteFamily = socket.remoteFamily;
  myDuplex.remoteAddress = socket.remoteAddress;
  myDuplex.remotePort = socket.remotePort;
  myDuplex.localFamily = socket.localFamily;
  myDuplex.localAddress = socket.localAddress;
  myDuplex.localPort = socket.localPort;

  return myDuplex;
};

Packer.Transform = {};
Packer.Transform.create = function (opts) {
  // Note: service refers to the port that the incoming request was from,
  // if known (smtps, smtp, https, http, etc)
  // { address: '127.0.0.1', service: 'https' }
  return new MyTransform(opts);
};