proxy-packer.js/index.js

394 lines
11 KiB
JavaScript
Raw Normal View History

2016-09-22 19:11:54 +00:00
'use strict';
2016-10-01 02:07:01 +00:00
var Packer = module.exports;
Packer.create = function (opts) {
2016-09-30 04:25:54 +00:00
var machine;
if (!opts.onMessage && !opts.onmessage) {
machine = new (require('events').EventEmitter)();
2016-09-30 04:51:40 +00:00
} else {
machine = {};
2016-09-30 04:25:54 +00:00
}
machine.onmessage = opts.onmessage || opts.onMessage;
2018-04-29 01:24:13 +00:00
machine.oncontrol = opts.oncontrol || opts.onControl;
2016-09-30 04:25:54 +00:00
machine.onerror = opts.onerror || opts.onError;
machine.onend = opts.onend || opts.onEnd;
2016-09-22 19:11:54 +00:00
machine._version = 1;
machine.fns = {};
2018-04-29 01:24:13 +00:00
machine.chunkIndex = 0;
machine.buf = null;
machine.bufIndex = 0;
machine.fns.collectData = function (chunk, size) {
var chunkLeft = chunk.length - machine.chunkIndex;
if (size <= 0) {
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;
};
2016-09-22 19:11:54 +00:00
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]');
2018-04-29 01:24:13 +00:00
var header = machine.fns.collectData(chunk, machine.headerLen);
2016-09-22 19:11:54 +00:00
2018-04-29 01:24:13 +00:00
// We don't have the entire header yet so return false.
if (!header) {
2016-09-22 19:11:54 +00:00
return false;
}
2018-04-29 01:24:13 +00:00
machine._headers = header.toString().split(/,/g);
2016-09-22 19:11:54 +00:00
2018-04-29 01:24:13 +00:00
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];
//console.log('machine.service', machine.service);
2016-09-22 19:11:54 +00:00
2018-04-29 01:24:13 +00:00
return true;
2016-09-22 19:11:54 +00:00
};
machine.fns.data = function (chunk) {
//console.log('');
//console.log('[data]');
2018-04-29 01:24:13 +00:00
var data = machine.fns.collectData(chunk, machine.bodyLen);
2016-09-22 19:11:54 +00:00
2018-04-29 01:24:13 +00:00
// We don't have the entire body yet so return false.
if (!data) {
2016-09-22 19:11:54 +00:00
return false;
}
2016-09-30 04:25:54 +00:00
//
// data, end, error
//
2018-04-29 01:24:13 +00:00
var msg = {};
if ('error' === machine.service) {
try {
msg = JSON.parse(data.toString());
} catch(e) {
msg.message = data.toString();
msg.code = 'E_UNKNOWN_ERR';
}
}
2016-09-30 04:25:54 +00:00
2018-04-29 01:24:13 +00:00
msg.family = machine.family;
msg.address = machine.address;
msg.port = machine.port;
msg.service = machine.service;
msg.data = data;
2016-09-30 04:25:54 +00:00
2018-04-29 01:24:13 +00:00
if ('end' === machine.service) {
2016-09-30 04:25:54 +00:00
if (machine.emit) {
machine.emit('tunnelEnd', msg);
}
else {
(machine.onend||machine.onmessage)(msg);
}
}
2016-09-30 07:03:26 +00:00
else if ('error' === machine.service) {
2016-09-30 04:25:54 +00:00
if (machine.emit) {
machine.emit('tunnelError', msg);
}
else {
2016-09-30 05:03:06 +00:00
(machine.onerror||machine.onmessage)(msg);
2016-09-30 04:25:54 +00:00
}
}
2018-04-29 01:24:13 +00:00
else if ('control' === machine.service) {
if (machine.emit) {
machine.emit('tunnelControl', msg);
}
else {
(machine.oncontrol||machine.onmessage)(msg);
}
}
2016-09-30 04:25:54 +00:00
else {
if (machine.emit) {
machine.emit('tunnelData', msg);
}
else {
machine.onmessage(msg);
}
}
2016-09-22 19:11:54 +00:00
return true;
};
2018-04-29 01:24:13 +00:00
machine.state = 0;
machine.states = ['version', 'headerLength', 'header', 'data'];
2016-09-22 19:11:54 +00:00
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;
2018-04-29 01:24:13 +00:00
machine.state %= machine.states.length;
2016-09-22 19:11:54 +00:00
}
}
};
return machine;
};
2016-10-01 02:07:01 +00:00
Packer.pack = function (address, data, service) {
2016-09-30 06:44:36 +00:00
data = data || Buffer.alloc(1);
2018-04-29 01:24:13 +00:00
if (!Buffer.isBuffer(data)) {
data = new Buffer(JSON.stringify(data));
}
2016-09-30 06:44:36 +00:00
if (!data.byteLength) {
data = Buffer.alloc(1);
}
2016-09-30 07:03:26 +00:00
if ('error' === service) {
address.service = 'error';
}
else if ('end' === service) {
address.service = 'end';
2016-09-30 03:41:42 +00:00
}
2016-09-22 19:11:54 +00:00
var version = 1;
2018-04-29 01:24:13 +00:00
var header;
if (service === 'control') {
header = Buffer.from(['', '', '', data.byteLength, service].join(','));
}
else {
header = Buffer.from([
address.family, address.address, address.port, data.byteLength, (address.service || '')
].join(','));
}
2016-09-22 19:31:31 +00:00
var meta = Buffer.from([ 255 - version, header.length ]);
var buf = Buffer.alloc(meta.byteLength + header.byteLength + data.byteLength);
2018-04-29 01:24:13 +00:00
meta.copy(buf, 0);
header.copy(buf, 2);
data.copy(buf, 2 + header.byteLength);
2016-09-22 19:11:54 +00:00
return buf;
};
2016-10-01 02:07:01 +00:00
2018-04-29 01:25:46 +00:00
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 || '';
}
2016-10-01 02:07:01 +00:00
Packer.socketToAddr = function (socket) {
// TODO BUG XXX
// https://github.com/nodejs/node/issues/8854
// tlsSocket.remoteAddress = remoteAddress; // causes core dump
// console.log(tlsSocket.remoteAddress);
return {
2018-04-29 01:25:46 +00:00
family: extractSocketProp(socket, 'remoteFamily')
, address: extractSocketProp(socket, 'remoteAddress')
, port: extractSocketProp(socket, 'remotePort')
2016-10-01 02:07:01 +00:00
};
};
Packer.addrToId = function (address) {
return address.family + ',' + address.address + ',' + address.port;
};
Packer.socketToId = function (socket) {
return Packer.addrToId(Packer.socketToAddr(socket));
};
/*
*
* Tunnel Packer
*
*/
2018-04-29 01:25:46 +00:00
var addressNames = [
'remoteAddress'
, 'remotePort'
, 'remoteFamily'
, 'localAddress'
, 'localPort'
];
// Imporoved workaround for https://github.com/nodejs/node/issues/8854
// Unlike Duplex this should handle all of the events needed to make everything work.
Packer.wrapSocket = function (socket) {
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');
});
myDuplex.destroy = socket.destroy.bind(socket);
return myDuplex;
};
2016-10-01 02:07:01 +00:00
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;
Transform.call(this, options);
}
util.inherits(MyTransform, Transform);
MyTransform.prototype._transform = function (data, encoding, callback) {
2018-04-29 01:25:46 +00:00
var address = this.__my_addr;
address.service = address.service || this.__my_service;
this.push(Packer.pack(address, data));
callback();
2016-10-01 02:07:01 +00:00
};
Packer.Stream = {};
var Dup = {
write: function (chunk, encoding, cb) {
//console.log('_write', chunk.byteLength);
2018-04-29 01:25:46 +00:00
this.__my_socket.write(chunk, encoding, cb);
2016-10-01 02:07:01 +00:00
}
, 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) {
2018-04-29 01:25:46 +00:00
if (!Packer.Stream.warned) {
console.warn('`Stream.create` deprecated, use `wrapSocket` instead');
Packer.Stream.warned = true;
}
2016-10-01 02:07:01 +00:00
// 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);
};