WIP reconnect on network change

This commit is contained in:
AJ ONeal 2018-09-05 01:18:12 -06:00
parent d8aedb39c2
commit 9b0d758a8b
2 changed files with 281 additions and 380 deletions

View File

@ -14,6 +14,8 @@ var YAML = require('js-yaml');
var recase = require('recase').create({}); var recase = require('recase').create({});
var camelCopy = recase.camelCopy.bind(recase); var camelCopy = recase.camelCopy.bind(recase);
var snakeCopy = recase.snakeCopy.bind(recase); var snakeCopy = recase.snakeCopy.bind(recase);
var TelebitRemote = require('../').TelebitRemote;
var state = { homedir: os.homedir(), servernames: {}, ports: {} }; var state = { homedir: os.homedir(), servernames: {}, ports: {} };
var argv = process.argv.slice(2); var argv = process.argv.slice(2);
@ -680,28 +682,26 @@ function serveControlsHelper() {
} }
function serveControls() { function serveControls() {
serveControlsHelper();
if (state.config.disable) { if (state.config.disable) {
console.info("[info] starting disabled"); console.info("[info] starting disabled");
serveControlsHelper();
return; return;
} }
if (!(state.config.relay && (state.config.token || state.config.pretoken))) { if (!(state.config.relay && (state.config.token || state.config.pretoken))) {
console.info("[info] waiting for init/authentication (missing relay and/or token)"); console.info("[info] waiting for init/authentication (missing relay and/or token)");
serveControlsHelper();
return; return;
} }
console.info("[info] connecting with stored token"); console.info("[info] connecting with stored token");
startTelebitRemote(function (err/*, _tun*/) { function tryAgain() {
if (err) { throw err; } startTelebitRemote(function (err) {
//if (_tun) { myRemote = _tun; } if (err) { console.error('error starting (probably internet)', err); setTimeout(tryAgain, 5 * 1000); }
setTimeout(function () {
// TODO attach handler to tunnel
serveControlsHelper();
}, 150);
}); });
} }
tryAgain();
}
function parseConfig(err, text) { function parseConfig(err, text) {
@ -762,62 +762,7 @@ function parseConfig(err, text) {
} }
} }
function startTelebitRemote(rawCb) { function approveDomains(opts, certs, cb) {
if (state.config.disable || !state.config.relay || !(state.config.token || state.config.agreeTos)) {
rawCb(null, null);
return;
}
state.relay = state.config.relay;
if (!state.relay) {
rawCb(new Error("'" + state._confpath + "' is missing 'relay'"));
return;
}
if (!(state.token || state.pretoken)) {
rawCb(null, null);
return;
}
if (myRemote) {
rawCb(null, myRemote);
return;
}
// get the wss url
common.api.wss(state, function (err, wss) {
if (err) { rawCb(err); return; }
state.wss = wss;
// Saves the token
// state.handlers.access_token({ jwt: token });
// Adds the token to the connection
// tun.append(token);
state.greenlockConf = state.config.greenlock || {};
state.sortingHat = state.config.sortingHat;
// TODO sortingHat.print(); ?
// TODO Check undefined vs false for greenlock config
var TelebitRemote = require('../').TelebitRemote;
state.greenlockConfig = {
version: state.greenlockConf.version || 'draft-11'
, server: state.greenlockConf.server || 'https://acme-v02.api.letsencrypt.org/directory'
, communityMember: state.greenlockConf.communityMember || state.config.communityMember
, telemetry: state.greenlockConf.telemetry || state.config.telemetry
, configDir: state.greenlockConf.configDir
|| (state.config.root && path.join(state.config.root, 'etc/acme'))
|| path.join(os.homedir(), '.config/telebit/acme')
// TODO, store: require(state.greenlockConf.store.name || 'le-store-certbot').create(state.greenlockConf.store.options || {})
, approveDomains: function (opts, certs, cb) {
// Certs being renewed are listed in certs.altnames
if (certs) {
opts.domains = certs.altnames;
cb(null, { options: opts, certs: certs });
return;
}
// Even though it's being tunneled by a trusted source // Even though it's being tunneled by a trusted source
// we need to make sure we don't get rate-limit spammed // we need to make sure we don't get rate-limit spammed
// with wildcard domains // with wildcard domains
@ -829,13 +774,40 @@ function startTelebitRemote(rawCb) {
return; return;
} }
//cb(new Error("servername not found in allowed list")); cb(new Error("servername not found in allowed list"));
} }
function greenlockHelper() {
// TODO Check undefined vs false for greenlock config
state.greenlockConf = state.config.greenlock || {};
state.greenlockConfig = {
version: state.greenlockConf.version || 'draft-11'
, server: state.greenlockConf.server || 'https://acme-v02.api.letsencrypt.org/directory'
, communityMember: state.greenlockConf.communityMember || state.config.communityMember
, telemetry: state.greenlockConf.telemetry || state.config.telemetry
, configDir: state.greenlockConf.configDir
|| (state.config.root && path.join(state.config.root, 'etc/acme'))
|| path.join(os.homedir(), '.config/telebit/acme')
// TODO, store: require(state.greenlockConf.store.name || 'le-store-certbot').create(state.greenlockConf.store.options || {})
, approveDomains: approveDomains
}; };
state.insecure = state.config.relay_ignore_invalid_certificates; state.insecure = state.config.relay_ignore_invalid_certificates;
// { relay, config, servernames, ports, sortingHat, net, insecure, token, handlers, greenlockConfig } }
function startTelebitRemote(rawCb) {
console.log('DEBUG startTelebitRemote');
function startHelper() {
console.log('DEBUG startHelper');
greenlockHelper();
// Saves the token
// state.handlers.access_token({ jwt: token });
// Adds the token to the connection
// tun.append(token);
function onError(err) { function onError(err) {
myRemote = null;
console.log('DEBUG err', err);
// Likely causes: // Likely causes:
// * DNS lookup failed (no Internet) // * DNS lookup failed (no Internet)
// * Rejected (bad authn) // * Rejected (bad authn)
@ -843,7 +815,7 @@ function startTelebitRemote(rawCb) {
// DNS issue, probably network is disconnected // DNS issue, probably network is disconnected
setTimeout(function () { setTimeout(function () {
startTelebitRemote(rawCb); startTelebitRemote(rawCb);
}, 90 * 1000); }, 10 * 1000);
return; return;
} }
if ('function' === typeof rawCb) { if ('function' === typeof rawCb) {
@ -854,12 +826,14 @@ function startTelebitRemote(rawCb) {
} }
} }
console.log("[DEBUG] token", typeof token, token); console.log("[DEBUG] token", typeof token, token);
//state.sortingHat = state.config.sortingHat;
// { relay, config, servernames, ports, sortingHat, net, insecure, token, handlers, greenlockConfig }
myRemote = TelebitRemote.createConnection({ myRemote = TelebitRemote.createConnection({
relay: state.relay relay: state.relay
, wss: state.wss , wss: state.wss
, config: state.config , config: state.config
, otp: state.otp , otp: state.otp
, sortingHat: state.sortingHat , sortingHat: state.config.sortingHat
, net: state.net , net: state.net
, insecure: state.insecure , insecure: state.insecure
, token: state.token || state.pretoken // instance , token: state.token || state.pretoken // instance
@ -868,10 +842,83 @@ function startTelebitRemote(rawCb) {
, handlers: state.handlers , handlers: state.handlers
, greenlockConfig: state.greenlockConfig , greenlockConfig: state.greenlockConfig
}, function () { }, function () {
console.log('DEBUG on connect');
myRemote.removeListener('error', onError); myRemote.removeListener('error', onError);
myRemote.once('error', retryLoop);
rawCb(null, myRemote); rawCb(null, myRemote);
}); });
function retryLoop() {
// disconnected somehow
myRemote.destroy();
myRemote = null;
setTimeout(function () {
startTelebitRemote(function () {});
}, 10 * 1000);
}
myRemote.once('error', onError); myRemote.once('error', onError);
myRemote.once('close', retryLoop);
myRemote.on('grant', state.handlers.grant);
myRemote.on('access_token', state.handlers.access_token);
}
if (state.config.disable || !state.config.relay || !(state.config.token || state.config.agreeTos)) {
console.log('DEBUG disabled or incapable');
rawCb(null, null);
return;
}
state.relay = state.config.relay;
if (!state.relay) {
console.log('DEBUG no relay');
rawCb(new Error("'" + state._confpath + "' is missing 'relay'"));
return;
}
if (!(state.token || state.pretoken)) {
console.log('DEBUG no token');
rawCb(null, null);
return;
}
if (myRemote) {
console.log('DEBUG has remote');
rawCb(null, myRemote);
return;
}
if (state.wss) {
startHelper();
return;
}
// get the wss url
function retryWssLoop(err) {
myRemote = null;
if (!err) {
startHelper();
return;
}
if ('ENOTFOUND' === err.code) {
// The internet is disconnected
// try again, and again, and again
setTimeout(function () {
startTelebitRemote(rawCb);
}, 2 * 1000);
return;
}
rawCb(err);
return;
}
common.api.wss(state, function onWss(err, wss) {
if (err) {
retryWssLoop(err);
return;
}
state.wss = wss;
startHelper();
}); });
} }

View File

@ -27,14 +27,16 @@ function TelebitRemote(state) {
} }
EventEmitter.call(this); EventEmitter.call(this);
var me = this; var me = this;
var priv = {};
var defaultHttpTimeout = (2 * 60); //var defaultHttpTimeout = (2 * 60);
var activityTimeout = state.activityTimeout || (defaultHttpTimeout - 5) * 1000; //var activityTimeout = state.activityTimeout || (defaultHttpTimeout - 5) * 1000;
var activityTimeout = 6 * 1000;
var pongTimeout = state.pongTimeout || 10*1000; var pongTimeout = state.pongTimeout || 10*1000;
// Allow the tunnel client to be created with no token. This will prevent the connection from // Allow the tunnel client to be created with no token. This will prevent the connection from
// being established initialy and allows the caller to use `.append` for the first token so // being established initialy and allows the caller to use `.append` for the first token so
// they can get a promise that will provide feedback about invalid tokens. // they can get a promise that will provide feedback about invalid tokens.
var tokens = []; priv.tokens = [];
var auth; var auth;
if(!state.sortingHat) { if(!state.sortingHat) {
state.sortingHat = "./sorting-hat.js"; state.sortingHat = "./sorting-hat.js";
@ -43,7 +45,7 @@ function TelebitRemote(state) {
if ('undefined' === state.token) { if ('undefined' === state.token) {
throw new Error("passed string 'undefined' as token"); throw new Error("passed string 'undefined' as token");
} }
tokens.push(state.token); priv.tokens.push(state.token);
} }
var wstunneler; var wstunneler;
@ -51,11 +53,11 @@ function TelebitRemote(state) {
var authsent = false; var authsent = false;
var initialConnect = true; var initialConnect = true;
var localclients = {}; priv.localclients = {};
var pausedClients = []; var pausedClients = [];
var clientHandlers = { var clientHandlers = {
add: function (conn, cid, tun) { add: function (conn, cid, tun) {
localclients[cid] = conn; priv.localclients[cid] = conn;
console.info("[connect] new client '" + cid + "' for '" + tun.name + ":" + tun.serviceport + "' " console.info("[connect] new client '" + cid + "' for '" + tun.name + ":" + tun.serviceport + "' "
+ "(" + clientHandlers.count() + " clients)"); + "(" + clientHandlers.count() + " clients)");
@ -80,9 +82,9 @@ function TelebitRemote(state) {
// down the data we are getting to send over. We also want to pause all active connections // down the data we are getting to send over. We also want to pause all active connections
// if any connections are paused to make things more fair so one connection doesn't get // if any connections are paused to make things more fair so one connection doesn't get
// stuff waiting for all other connections to finish because it tried writing near the border. // stuff waiting for all other connections to finish because it tried writing near the border.
var bufSize = wsHandlers.sendMessage(Packer.packHeader(tun, chunk)); var bufSize = sendMessage(Packer.packHeader(tun, chunk));
// Sending 2 messages instead of copying the buffer // Sending 2 messages instead of copying the buffer
var bufSize2 = wsHandlers.sendMessage(chunk); var bufSize2 = sendMessage(chunk);
if (pausedClients.length || (bufSize + bufSize2) > 1024*1024) { if (pausedClients.length || (bufSize + bufSize2) > 1024*1024) {
// console.log('[onLocalData] paused connection', cid, 'to allow websocket to catch up'); // console.log('[onLocalData] paused connection', cid, 'to allow websocket to catch up');
conn.pause(); conn.pause();
@ -95,7 +97,7 @@ function TelebitRemote(state) {
console.info("[onLocalEnd] connection '" + cid + "' ended, will probably close soon"); console.info("[onLocalEnd] connection '" + cid + "' ended, will probably close soon");
conn.tunnelClosing = true; conn.tunnelClosing = true;
if (!sentEnd) { if (!sentEnd) {
wsHandlers.sendMessage(Packer.packHeader(tun, null, 'end')); sendMessage(Packer.packHeader(tun, null, 'end'));
sentEnd = true; sentEnd = true;
} }
}); });
@ -103,22 +105,22 @@ function TelebitRemote(state) {
console.info("[onLocalError] connection '" + cid + "' errored:", err); console.info("[onLocalError] connection '" + cid + "' errored:", err);
if (!sentEnd) { if (!sentEnd) {
var packBody = true; var packBody = true;
wsHandlers.sendMessage(Packer.packHeader(tun, {message: err.message, code: err.code}, 'error', packBody)); sendMessage(Packer.packHeader(tun, {message: err.message, code: err.code}, 'error', packBody));
sentEnd = true; sentEnd = true;
} }
}); });
conn.on('close', function onLocalClose(hadErr) { conn.on('close', function onLocalClose(hadErr) {
delete localclients[cid]; delete priv.localclients[cid];
console.log('[onLocalClose] closed "' + cid + '" read:'+conn.tunnelRead+', wrote:'+conn.tunnelWritten+' (' + clientHandlers.count() + ' clients)'); console.log('[onLocalClose] closed "' + cid + '" read:'+conn.tunnelRead+', wrote:'+conn.tunnelWritten+' (' + clientHandlers.count() + ' clients)');
if (!sentEnd) { if (!sentEnd) {
wsHandlers.sendMessage(Packer.packHeader(tun, null, hadErr && 'error' || 'end')); sendMessage(Packer.packHeader(tun, null, hadErr && 'error' || 'end'));
sentEnd = true; sentEnd = true;
} }
}); });
} }
, write: function (cid, opts) { , write: function (cid, opts) {
var conn = localclients[cid]; var conn = priv.localclients[cid];
if (!conn) { if (!conn) {
return false; return false;
} }
@ -136,12 +138,12 @@ function TelebitRemote(state) {
if (!conn.remotePaused && conn.bufferSize > 1024*1024) { if (!conn.remotePaused && conn.bufferSize > 1024*1024) {
var packBody = true; var packBody = true;
wsHandlers.sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'pause', packBody)); sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'pause', packBody));
conn.remotePaused = true; conn.remotePaused = true;
conn.once('drain', function () { conn.once('drain', function () {
var packBody = true; var packBody = true;
wsHandlers.sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'resume', packBody)); sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'resume', packBody));
conn.remotePaused = false; conn.remotePaused = false;
}); });
} }
@ -149,13 +151,13 @@ function TelebitRemote(state) {
} }
, closeSingle: function (cid) { , closeSingle: function (cid) {
if (!localclients[cid]) { if (!priv.localclients[cid]) {
return; return;
} }
console.log('[closeSingle]', cid); console.log('[closeSingle]', cid);
PromiseA.resolve().then(function () { PromiseA.resolve().then(function () {
var conn = localclients[cid]; var conn = priv.localclients[cid];
conn.tunnelClosing = true; conn.tunnelClosing = true;
conn.end(); conn.end();
@ -173,41 +175,49 @@ function TelebitRemote(state) {
}); });
}); });
}).then(function () { }).then(function () {
if (localclients[cid]) { if (priv.localclients[cid]) {
console.warn('[closeSingle]', cid, 'connection still present after calling `end`'); console.warn('[closeSingle]', cid, 'connection still present after calling `end`');
localclients[cid].destroy(); priv.localclients[cid].destroy();
return timeoutPromise(500); return timeoutPromise(500);
} }
}).then(function () { }).then(function () {
if (localclients[cid]) { if (priv.localclients[cid]) {
console.error('[closeSingle]', cid, 'connection still present after calling `destroy`'); console.error('[closeSingle]', cid, 'connection still present after calling `destroy`');
delete localclients[cid]; delete priv.localclients[cid];
} }
}).catch(function (err) { }).catch(function (err) {
console.error('[closeSingle] failed to close connection', cid, err.toString()); console.error('[closeSingle] failed to close connection', cid, err.toString());
delete localclients[cid]; delete priv.localclients[cid];
}); });
} }
, closeAll: function () { , closeAll: function () {
console.log('[closeAll]'); console.log('[closeAll]');
Object.keys(localclients).forEach(function (cid) { Object.keys(priv.localclients).forEach(function (cid) {
clientHandlers.closeSingle(cid); clientHandlers.closeSingle(cid);
}); });
} }
, count: function () { , count: function () {
return Object.keys(localclients).length; return Object.keys(priv.localclients).length;
} }
}; };
var pendingCommands = {}; var pendingCommands = {};
function sendMessage(msg) {
// There is a chance that this occurred after the websocket was told to close
// and before it finished, in which case we don't need to log the error.
if (wstunneler.readyState !== wstunneler.CLOSING) {
wstunneler.send(msg, {binary: true});
return wstunneler.bufferedAmount;
}
}
function sendCommand(name) { function sendCommand(name) {
var id = Math.ceil(1e9 * Math.random()); var id = Math.ceil(1e9 * Math.random());
var cmd = [id, name].concat(Array.prototype.slice.call(arguments, 1)); var cmd = [id, name].concat(Array.prototype.slice.call(arguments, 1));
if (state.debug) { console.log('[DEBUG] command sending', cmd); } if (state.debug) { console.log('[DEBUG] command sending', cmd); }
var packBody = true; var packBody = true;
wsHandlers.sendMessage(Packer.packHeader(null, cmd, 'control', packBody)); sendMessage(Packer.packHeader(null, cmd, 'control', packBody));
setTimeout(function () { setTimeout(function () {
if (pendingCommands[id]) { if (pendingCommands[id]) {
console.warn('command', name, id, 'timed out'); console.warn('command', name, id, 'timed out');
@ -230,24 +240,6 @@ function TelebitRemote(state) {
}); });
} }
function sendAllTokens() {
if (auth) {
authsent = true;
sendCommand('auth', auth).catch(function (err) { console.error('1', err); });
}
tokens.forEach(function (jwtoken) {
if (state.debug) { console.log('[DEBUG] send token'); }
authsent = true;
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.
});
});
}
function noHandler(cmd) { function noHandler(cmd) {
console.warn("[telebit] state.handlers['" + cmd[1] + "'] not set"); console.warn("[telebit] state.handlers['" + cmd[1] + "'] not set");
console.warn(cmd[2]); console.warn(cmd[2]);
@ -303,7 +295,21 @@ function TelebitRemote(state) {
if (cmd[1] === 'hello') { if (cmd[1] === 'hello') {
if (state.debug) { console.log('[DEBUG] hello received'); } if (state.debug) { console.log('[DEBUG] hello received'); }
sendAllTokens(); if (auth) {
authsent = true;
sendCommand('auth', auth).catch(function (err) { console.error('1', err); });
}
priv.tokens.forEach(function (jwtoken) {
if (state.debug) { console.log('[DEBUG] send token'); }
authsent = true;
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.
});
});
if (connCallback) { if (connCallback) {
connCallback(); connCallback();
} }
@ -331,7 +337,7 @@ function TelebitRemote(state) {
} }
var packBody = true; var packBody = true;
wsHandlers.sendMessage(Packer.packHeader(null, [-cmd[0], err], 'control', packBody)); sendMessage(Packer.packHeader(null, [-cmd[0], err], 'control', packBody));
} }
, onconnection: function (tun) { , onconnection: function (tun) {
@ -369,28 +375,28 @@ function TelebitRemote(state) {
, onpause: function (opts) { , onpause: function (opts) {
var cid = Packer.addrToId(opts); var cid = Packer.addrToId(opts);
if (localclients[cid]) { if (priv.localclients[cid]) {
console.log("[TunnelPause] pausing '"+cid+"', remote received", opts.data.toString(), 'of', localclients[cid].tunnelWritten, 'sent'); console.log("[TunnelPause] pausing '"+cid+"', remote received", opts.data.toString(), 'of', priv.localclients[cid].tunnelWritten, 'sent');
localclients[cid].manualPause = true; priv.localclients[cid].manualPause = true;
localclients[cid].pause(); priv.localclients[cid].pause();
} else { } else {
console.log('[TunnelPause] remote tried pausing finished connection', cid); 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 // 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. // don't worry about sending back errors, since we won't be sending data over anyway.
// var packBody = true; // var packBody = true;
// wsHandlers.sendMessage(Packer.packHeader(opts, {message: 'no matching connection', code: 'E_NO_CONN'}, 'error', packBody)); // sendMessage(Packer.packHeader(opts, {message: 'no matching connection', code: 'E_NO_CONN'}, 'error', packBody));
} }
} }
, onresume: function (opts) { , onresume: function (opts) {
var cid = Packer.addrToId(opts); var cid = Packer.addrToId(opts);
if (localclients[cid]) { if (priv.localclients[cid]) {
console.log("[TunnelResume] resuming '"+cid+"', remote received", opts.data.toString(), 'of', localclients[cid].tunnelWritten, 'sent'); console.log("[TunnelResume] resuming '"+cid+"', remote received", opts.data.toString(), 'of', priv.localclients[cid].tunnelWritten, 'sent');
localclients[cid].manualPause = false; priv.localclients[cid].manualPause = false;
localclients[cid].resume(); priv.localclients[cid].resume();
} else { } else {
console.log('[TunnelResume] remote tried resuming finished connection', cid); console.log('[TunnelResume] remote tried resuming finished connection', cid);
// var packBody = true; // var packBody = true;
// wsHandlers.sendMessage(Packer.packHeader(opts, {message: 'no matching connection', code: 'E_NO_CONN'}, 'error', packBody)); // sendMessage(Packer.packHeader(opts, {message: 'no matching connection', code: 'E_NO_CONN'}, 'error', packBody));
} }
} }
@ -407,28 +413,27 @@ function TelebitRemote(state) {
, _onConnectError: function (cid, opts, err) { , _onConnectError: function (cid, opts, err) {
console.info("[_onConnectError] opening '" + cid + "' failed because " + err.message); console.info("[_onConnectError] opening '" + cid + "' failed because " + err.message);
wsHandlers.sendMessage(Packer.packHeader(opts, null, 'error')); sendMessage(Packer.packHeader(opts, null, 'error'));
} }
}; };
var lastActivity; priv.timeoutId = null;
var timeoutId; priv.lastActivity = Date.now();
var wsHandlers = { priv.refreshTimeout = function refreshTimeout() {
refreshTimeout: function () { priv.lastActivity = Date.now();
lastActivity = Date.now(); };
} priv.checkTimeout = function checkTimeout() {
, checkTimeout: function () {
if (!wstunneler) { if (!wstunneler) {
console.warn('checkTimeout called when websocket already closed'); console.warn('checkTimeout called when websocket already closed');
return; return;
} }
// Determine how long the connection has been "silent", ie no activity. // Determine how long the connection has been "silent", ie no activity.
var silent = Date.now() - lastActivity; var silent = Date.now() - priv.lastActivity;
// If we have had activity within the last activityTimeout then all we need to do is // If we have had activity within the last activityTimeout then all we need to do is
// call this function again at the soonest time when the connection could be timed out. // call this function again at the soonest time when the connection could be timed out.
if (silent < activityTimeout) { if (silent < activityTimeout) {
timeoutId = setTimeout(wsHandlers.checkTimeout, activityTimeout-silent); priv.timeoutId = setTimeout(priv.checkTimeout, activityTimeout-silent);
} }
// Otherwise we check to see if the pong has also timed out, and if not we send a ping // Otherwise we check to see if the pong has also timed out, and if not we send a ping
@ -440,7 +445,7 @@ function TelebitRemote(state) {
} catch (err) { } catch (err) {
console.warn('failed to ping tunnel server', err); console.warn('failed to ping tunnel server', err);
} }
timeoutId = setTimeout(wsHandlers.checkTimeout, pongTimeout); priv.timeoutId = setTimeout(priv.checkTimeout, pongTimeout);
} }
// Last case means the ping we sent before didn't get a response soon enough, so we // Last case means the ping we sent before didn't get a response soon enough, so we
@ -449,14 +454,32 @@ function TelebitRemote(state) {
console.log('connection timed out'); console.log('connection timed out');
wstunneler.close(1000, 'connection timeout'); wstunneler.close(1000, 'connection timeout');
} }
};
me.destroy = function destroy() {
try {
//wstunneler.close(1000, 're-connect');
wstunneler._socket.destroy();
} catch(e) {
// ignore
} }
};
me.connect = function connect() {
if (!priv.tokens.length && state.config.email) {
auth = TelebitRemote._tokenFromState(state);
}
priv.timeoutId = null;
var machine = Packer.create(packerHandlers);
, onOpen: function () { console.info("[connect] '" + (state.wss || state.relay) + "'");
var tunnelUrl = (state.wss || state.relay).replace(/\/$/, '') + '/'; // + auth;
wstunneler = new WebSocket(tunnelUrl, { rejectUnauthorized: !state.insecure });
// XXXXXX
wstunneler.on('open', function () {
console.info("[open] connected to '" + (state.wss || state.relay) + "'"); console.info("[open] connected to '" + (state.wss || state.relay) + "'");
wsHandlers.refreshTimeout(); me.emit('connect');
priv.refreshTimeout();
timeoutId = setTimeout(wsHandlers.checkTimeout, activityTimeout); priv.timeoutId = setTimeout(priv.checkTimeout, activityTimeout);
wstunneler._socket.on('drain', function () { wstunneler._socket.on('drain', function () {
// the websocket library has it's own buffer apart from node's socket buffer, but that one // the websocket library has it's own buffer apart from node's socket buffer, but that one
// is much more difficult to watch, so we watch for the lower level buffer to drain and // is much more difficult to watch, so we watch for the lower level buffer to drain and
@ -473,24 +496,13 @@ function TelebitRemote(state) {
conn.resume(); conn.resume();
} }
}); });
pausedClients.length = 0; pausedClients.length = 0;
}); });
/*
//Call either Open or Reconnect handlers.
if(state.handlers.onOpen && initialConnect) {
state.handlers.onOpen();
} else if (state.handlers.onReconnect && !initialConnect) {
state.handlers.onReconnect();
}
*/
initialConnect = false; initialConnect = false;
} });
wstunneler.on('close', function () {
, onClose: function () { console.log("DEBUG closing");
clearTimeout(timeoutId); clearTimeout(priv.timeoutId);
wstunneler = null;
clientHandlers.closeAll(); clientHandlers.closeAll();
var error = new Error('websocket connection closed before response'); var error = new Error('websocket connection closed before response');
@ -503,75 +515,42 @@ function TelebitRemote(state) {
} }
me.emit('close'); me.emit('close');
/* });
return; wstunneler.on('error', function (err) {
if (!authenticated) {
if(state.handlers.onError) {
var err = new Error('Failed to connect on first attempt... check authentication');
state.handlers.onError(err);
}
if(state.handlers.onClose) {
state.handlers.onClose();
}
console.info('[close] failed on first attempt... check authentication.');
timeoutId = null;
}
else if (tokens.length) {
if(state.handlers.onDisconnect) {
state.handlers.onDisconnect();
}
console.info('[retry] disconnected and waiting...');
timeoutId = setTimeout(connect, 5000);
} else {
if(state.handlers.onClose) {
state.handlers.onClose();
}
}
*/
}
, onError: function (err) {
me.emit('error', err); me.emit('error', err);
/* });
return;
if ('ENOTFOUND' === err.code) {
// DNS issue, probably network is disconnected
timeoutId = setTimeout(connect, 90 * 1000);
return;
}
console.error("[tunnel error] " + err.message);
console.error(err);
if (connCallback) {
connCallback(err);
}
*/
}
, sendMessage: function (msg) { // Our library will automatically handle sending the pong respose to ping requests.
if (wstunneler) { wstunneler.on('ping', priv.refreshTimeout);
try { wstunneler.on('pong', function () {
wstunneler.send(msg, {binary: true}); console.log('DEBUG got pong');
return wstunneler.bufferedAmount; priv.refreshTimeout();
} catch (err) { });
// There is a chance that this occurred after the websocket was told to close wstunneler.on('message', function (data, flags) {
// and before it finished, in which case we don't need to log the error. priv.refreshTimeout();
if (wstunneler.readyState !== wstunneler.CLOSING) { if (data.error || '{' === data[0]) {
console.warn('[sendMessage] error sending websocket message', err); console.log(data);
} return;
}
}
} }
machine.fns.addChunk(data, flags);
});
}; };
me.end = function() {
var connPromise; priv.tokens.length = 0;
me.connect = function connect() { if (priv.timeoutId) {
if (wstunneler) { clearTimeout(priv.timeoutId);
console.warn('attempted to connect with connection already active'); priv.timeoutId = null;
return;
} }
if (!tokens.length) { wstunneler.close(1000, 're-connect');
if (state.config.email) { wstunneler.on('close', function () {
auth = { me.emit('end');
});
};
}
TelebitRemote.prototype = EventEmitter.prototype;
TelebitRemote._tokenFromState = function (state) {
return {
subject: state.config.email subject: state.config.email
, subject_scheme: 'mailto' , subject_scheme: 'mailto'
// TODO create domains list earlier // TODO create domains list earlier
@ -584,132 +563,7 @@ function TelebitRemote(state) {
, os_release: os.release() , os_release: os.release()
, os_arch: os.arch() , os_arch: os.arch()
}; };
}
}
timeoutId = null;
var machine = Packer.create(packerHandlers);
console.info("[connect] '" + (state.wss || state.relay) + "'");
var tunnelUrl = (state.wss || state.relay).replace(/\/$/, '') + '/'; // + auth;
wstunneler = new WebSocket(tunnelUrl, { rejectUnauthorized: !state.insecure });
// XXXXXX
wstunneler.on('open', function () {
me.emit('connect');
wsHandlers.onOpen();
});
wstunneler.on('close', wsHandlers.onClose);
wstunneler.on('error', wsHandlers.onError);
// Our library will automatically handle sending the pong respose to ping requests.
wstunneler.on('ping', wsHandlers.refreshTimeout);
wstunneler.on('pong', wsHandlers.refreshTimeout);
wstunneler.on('message', function (data, flags) {
wsHandlers.refreshTimeout();
if (data.error || '{' === data[0]) {
console.log(data);
return;
}
machine.fns.addChunk(data, flags);
});
}; };
me.end = function() {
tokens.length = 0;
if (timeoutId) {
clearTimeout(timeoutId);
timeoutId = null;
}
if (wstunneler) {
try {
wstunneler.close(1000, 're-connect');
wstunneler.on('close', function () {
me.emit('end');
});
} catch(e) {
console.error("[error] wstunneler.close()");
console.error(e);
}
}
};
me.authz = me.append = function (token) {
if (!token) {
throw new Error("attempted to append empty token");
}
if ('undefined' === token) {
throw new Error("attempted to append token as the string 'undefined'");
}
if (tokens.indexOf(token) >= 0) {
return PromiseA.resolve();
}
tokens.push(token);
var prom;
if (tokens.length === 1 && !wstunneler) {
// We just added the only token in the list, and the websocket connection isn't up
// so we need to restart the connection.
if (timeoutId) {
// Handle the case were the last token was removed and this token added between
// reconnect attempts to make sure we don't try openning multiple connections.
clearTimeout(timeoutId);
timeoutId = null;
}
// We want this case to behave as much like the other case as we can, but we don't have
// the same kind of reponses when we open brand new connections, so we have to rely on
// the 'hello' and the 'un-associated' error commands to determine if the token is good.
prom = connPromise = new PromiseA(function (resolve, reject) {
connCallback = function (err) {
connCallback = null;
connPromise = null;
if (err) {
reject(err);
} else {
resolve();
}
};
});
connect();
}
else if (connPromise) {
prom = connPromise.then(function () {
return sendCommand('add_token', token);
});
}
else {
prom = sendCommand('add_token', token);
}
prom.catch(function (err) {
console.error('adding token', token, 'failed:', err);
// Most probably an invalid token of some kind, so we don't really want to keep it.
tokens.splice(tokens.indexOf(token), 1);
});
return prom;
};
me.clear = function (token) {
if (typeof token === 'undefined') {
token = '*';
}
if (token === '*') {
tokens.length = 0;
} else {
var index = tokens.indexOf(token);
if (index < 0) {
return PromiseA.resolve();
}
tokens.splice(index);
}
var prom = sendCommand('delete_token', token);
prom.catch(function (err) {
console.error('clearing token', token, 'failed:', err);
});
return prom;
};
}
TelebitRemote.prototype = EventEmitter.prototype;
TelebitRemote.create = function (opts) { TelebitRemote.create = function (opts) {
return new TelebitRemote(opts); return new TelebitRemote(opts);