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];
		machine.servicename = machine._headers[7];
		//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 = 'e:' + JSON.stringify(data);
				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 ('connection' === machine.service) {
			msg.service = machine.servicename;
		}

		//console.log('msn', machine.service);
		if (machine.emit) {
			machine.emit(
				serviceEvents[machine.service] ||
					serviceEvents[msg.service] ||
					serviceEvents.default
			);
		} else {
			(machine[serviceFuncs[machine.service]] ||
				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;
			}
		}
		if ('data' === machine.states[machine.state] && 0 === machine.bodyLen) {
			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 = Buffer.from(JSON.stringify(data));
	}
	if (oldways && !data.byteLength) {
		data = Buffer.from(' ');
	}

	if (service && -1 === ['control', 'connection'].indexOf(service)) {
		//console.log('end?', service);
		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 if (service === 'connection') {
		header = Buffer.from(
			[
				meta.family,
				meta.address,
				meta.port,
				size,
				'connection',
				meta.serviceport || '',
				meta.name || '',
				meta.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) {
		if (
			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.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);
};