proxy-packer.js/index.js

476 lines
14 KiB
JavaScript

'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);
};