246 lines
8.2 KiB
JavaScript
246 lines
8.2 KiB
JavaScript
'use strict';
|
|
|
|
var sni = require('sni');
|
|
var Packer = require('proxy-packer');
|
|
|
|
module.exports.create = function (handlers) {
|
|
var client = module.exports;
|
|
client.pendingCommands = {};
|
|
client.auth = null;
|
|
client.wstunneler = null;
|
|
client.localclients = {};
|
|
client.authenticated = false;
|
|
|
|
var multiplexed = {};
|
|
var stream = require('stream');
|
|
var Duplex = stream.Duplex;
|
|
|
|
function SingleConn(tun, streamOpts) {
|
|
// Proper duplex stream with automatic flow control (backpressure) management
|
|
if(!(this instanceof SingleConn)) { return new SingleConn(tun, streamOpts); }
|
|
Duplex.call(this, streamOpts);
|
|
}
|
|
SingleConn.create = function (opts) {
|
|
return new SingleConn(opts);
|
|
};
|
|
|
|
if (!handlers) { handlers = {}; }
|
|
// XXX TODO
|
|
handlers.onAuthRequest = function (authRequest) {
|
|
// XXX out of scope
|
|
client._wsTunnelRemote._sendCommand('auth', authRequest).catch(function (err) { console.error('1', err); });
|
|
};
|
|
handlers.onAddToken = function (jwtoken) {
|
|
// XXX out of scope
|
|
client._wsTunnelRemote._sendCommand('add_token', jwtoken)
|
|
.catch(function (err) {
|
|
console.error('failed re-adding token', jwtoken, 'after reconnect', err);
|
|
// Not sure if we should do something like remove the token here. It worked
|
|
// once or it shouldn't have stayed in the list, so it's less certain why
|
|
// it would have failed here.
|
|
});
|
|
};
|
|
handlers.onAck = function (body) {
|
|
var packBody = true;
|
|
client._wsTunnelRemote.sendMessage(Packer.packHeader(null, body, 'control', packBody));
|
|
};
|
|
handlers.onNoHandler = function (cmd) {
|
|
console.warn("[telebit] state.handlers['" + cmd[1] + "'] not set");
|
|
console.warn(cmd[2]);
|
|
};
|
|
// TODO
|
|
// make proxy-packer a readable stream?
|
|
// create per-connection buffer?
|
|
handlers.onNonReadable = function (/*fn*/) {
|
|
client.wstunneler.pause();
|
|
//packerHandlers.onReadable = fn;
|
|
};
|
|
handlers.onReadable = function () {
|
|
//packerHandlers.onReadable();
|
|
client.wstunneler.resume();
|
|
};
|
|
|
|
var authsent = false;
|
|
function sendAllTokens() {
|
|
if (client.auth) {
|
|
authsent = true;
|
|
handlers.onAuthRequest(client.auth);
|
|
}
|
|
client.sharedTokens.forEach(function (jwtoken) {
|
|
// XXX out of scope
|
|
if (client._state.debug) { console.log('[DEBUG] send token'); }
|
|
authsent = true;
|
|
handlers.onAddToken(jwtoken);
|
|
});
|
|
}
|
|
|
|
function hyperPeek(tun) {
|
|
var m;
|
|
var str;
|
|
if (tun.data) {
|
|
if ('http' === tun.service) {
|
|
str = tun.data.toString();
|
|
m = str.match(/(?:^|[\r\n])Host: ([^\r\n]+)[\r\n]*/im);
|
|
tun._name = tun._hostname = (m && m[1].toLowerCase() || '').split(':')[0];
|
|
}
|
|
else if ('https' === tun.service || 'tls' === tun.service) {
|
|
tun._name = tun._servername = sni(tun.data);
|
|
} else {
|
|
tun._name = '';
|
|
}
|
|
}
|
|
}
|
|
|
|
var packerHandlers = {
|
|
oncontrol: function (opts) {
|
|
var cmd, err;
|
|
try {
|
|
cmd = JSON.parse(opts.data.toString());
|
|
} catch (err) {
|
|
// ignore
|
|
}
|
|
|
|
if (!Array.isArray(cmd) || typeof cmd[0] !== 'number') {
|
|
console.warn('received bad command "' + opts.data.toString() + '"');
|
|
return;
|
|
}
|
|
|
|
if (cmd[0] < 0) {
|
|
var cb = client.pendingCommands[-cmd[0]];
|
|
if (!cb) {
|
|
console.warn('received response for unknown request:', cmd);
|
|
} else {
|
|
cb.apply(null, cmd.slice(1));
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (cmd[0] === 0) {
|
|
console.warn('received dis-associated error from server', cmd[1]);
|
|
if (client.connCallback) {
|
|
client.connCallback(cmd[1]);
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (cmd[1] === 'hello') {
|
|
if (client._state.debug) { console.log('[DEBUG] hello received'); }
|
|
sendAllTokens();
|
|
if (client.connCallback) {
|
|
client.connCallback();
|
|
}
|
|
// TODO: handle the versions and commands provided by 'hello' - isn't super important
|
|
// yet since there is only one version and set of commands.
|
|
err = null;
|
|
} else if (cmd[1] === 'grant') {
|
|
client.authenticated = true;
|
|
if (client._state.handlers[cmd[1]]) {
|
|
client._state.handlers[cmd[1]](cmd[2]);
|
|
} else {
|
|
handlers.onNoHandler(cmd);
|
|
}
|
|
return;
|
|
} else if (cmd[1] === 'access_token') {
|
|
client.authenticated = true;
|
|
if (client._state.handlers[cmd[1]]) {
|
|
client._state.handlers[cmd[1]](cmd[2]);
|
|
} else {
|
|
handlers.onNoHandler(cmd);
|
|
}
|
|
return;
|
|
} else {
|
|
err = { message: 'unknown command "'+cmd[1]+'"', code: 'E_UNKNOWN_COMMAND' };
|
|
}
|
|
|
|
handlers.onAck([-cmd[0], err]);
|
|
}
|
|
|
|
, onconnection: function (tun, handledCb) {
|
|
var cid = tun._id = Packer.addrToId(tun);
|
|
if (multiplexed[cid]) {
|
|
throw new Error("[Sanity Error] a new connection can't already exist in the connection pool");
|
|
}
|
|
|
|
// this data should have been gathered already as part of the proxy protocol
|
|
// but if it's available again here we can double check
|
|
hyperPeek(tun);
|
|
|
|
// the next data that comes in may be the next packet of data for this connection
|
|
// and that may happen before the detection and assignment is complete
|
|
handlers.onNonReadable(); // pause()
|
|
// TODO use readable streams instead
|
|
require(client._state.sortingHat).assign(client._state, tun, function (err, conn) {
|
|
if (err) {
|
|
err.message = err.message.replace(/:tun_id/, tun._id);
|
|
console.info("[_onConnectError] opening '" + cid + "' failed because " + err.message);
|
|
client._wsTunnelRemote.sendMessage(Packer.packHeader(tun, null, 'error'));
|
|
return;
|
|
}
|
|
//handlers.on
|
|
client.clientHandlers.add(conn, cid, tun);
|
|
if (tun.data) { conn.write(tun.data); }
|
|
handlers.onReadable(); // resume
|
|
if ('function' === typeof handledCb) { handledCb(); }
|
|
});
|
|
}
|
|
|
|
, onmessage: function (tun) {
|
|
var cid = tun._id = Packer.addrToId(tun);
|
|
var handled;
|
|
|
|
hyperPeek(tun);
|
|
|
|
handled = client.clientHandlers.write(cid, tun);
|
|
|
|
if (!handled) {
|
|
throw new Error("No 'message' event came before 'connection' event."
|
|
+ " You're probably using a different version of proxy-packer on the server than the client");
|
|
}
|
|
}
|
|
|
|
, onpause: function (opts) {
|
|
var cid = Packer.addrToId(opts);
|
|
if (client.localclients[cid]) {
|
|
console.log("[TunnelPause] pausing '"+cid+"', remote received", opts.data.toString(), 'of', client.localclients[cid].tunnelWritten, 'sent');
|
|
client.localclients[cid].manualPause = true;
|
|
client.localclients[cid].pause();
|
|
} else {
|
|
console.log('[TunnelPause] remote tried pausing finished connection', cid);
|
|
// Often we have enough latency that we've finished sending before we're told to pause, so
|
|
// don't worry about sending back errors, since we won't be sending data over anyway.
|
|
// var packBody = true;
|
|
// wsTunnelRemote.sendMessage(Packer.packHeader(opts, {message: 'no matching connection', code: 'E_NO_CONN'}, 'error', packBody));
|
|
}
|
|
}
|
|
, onresume: function (opts) {
|
|
var cid = Packer.addrToId(opts);
|
|
if (client.localclients[cid]) {
|
|
console.log("[TunnelResume] resuming '"+cid+"', remote received", opts.data.toString(), 'of', client.localclients[cid].tunnelWritten, 'sent');
|
|
client.localclients[cid].manualPause = false;
|
|
client.localclients[cid].resume();
|
|
} else {
|
|
console.log('[TunnelResume] remote tried resuming finished connection', cid);
|
|
// var packBody = true;
|
|
// wsTunnelRemote.sendMessage(Packer.packHeader(opts, {message: 'no matching connection', code: 'E_NO_CONN'}, 'error', packBody));
|
|
}
|
|
}
|
|
|
|
, onend: function (opts) {
|
|
var cid = Packer.addrToId(opts);
|
|
//console.log("[end] '" + cid + "'");
|
|
client.clientHandlers.closeSingle(cid);
|
|
}
|
|
, onerror: function (opts) {
|
|
var cid = Packer.addrToId(opts);
|
|
//console.log("[error] '" + cid + "'", opts.code || '', opts.message);
|
|
client.clientHandlers.closeSingle(cid);
|
|
}
|
|
};
|
|
|
|
|
|
client.machine = Packer.create(packerHandlers);
|
|
client.sharedTokens = [];
|
|
|
|
return client;
|
|
};
|