From 9b0d758a8bf8c7b4d4bb23e4e0f844631bd2709d Mon Sep 17 00:00:00 2001 From: AJ ONeal Date: Wed, 5 Sep 2018 01:18:12 -0600 Subject: [PATCH] WIP reconnect on network change --- bin/telebitd.js | 203 ++++++++++++--------- lib/remote.js | 458 +++++++++++++++++------------------------------- 2 files changed, 281 insertions(+), 380 deletions(-) diff --git a/bin/telebitd.js b/bin/telebitd.js index c545246..9901a26 100755 --- a/bin/telebitd.js +++ b/bin/telebitd.js @@ -14,6 +14,8 @@ var YAML = require('js-yaml'); var recase = require('recase').create({}); var camelCopy = recase.camelCopy.bind(recase); var snakeCopy = recase.snakeCopy.bind(recase); +var TelebitRemote = require('../').TelebitRemote; + var state = { homedir: os.homedir(), servernames: {}, ports: {} }; var argv = process.argv.slice(2); @@ -680,27 +682,25 @@ function serveControlsHelper() { } function serveControls() { + serveControlsHelper(); + if (state.config.disable) { console.info("[info] starting disabled"); - serveControlsHelper(); return; } if (!(state.config.relay && (state.config.token || state.config.pretoken))) { console.info("[info] waiting for init/authentication (missing relay and/or token)"); - serveControlsHelper(); return; } console.info("[info] connecting with stored token"); - startTelebitRemote(function (err/*, _tun*/) { - if (err) { throw err; } - //if (_tun) { myRemote = _tun; } - setTimeout(function () { - // TODO attach handler to tunnel - serveControlsHelper(); - }, 150); - }); + function tryAgain() { + startTelebitRemote(function (err) { + if (err) { console.error('error starting (probably internet)', err); setTimeout(tryAgain, 5 * 1000); } + }); + } + tryAgain(); } function parseConfig(err, text) { @@ -762,80 +762,52 @@ function parseConfig(err, text) { } } +function approveDomains(opts, certs, cb) { + // Even though it's being tunneled by a trusted source + // we need to make sure we don't get rate-limit spammed + // with wildcard domains + // TODO: finish implementing dynamic dns for wildcard certs + if (getServername(state.servernames, opts.domains[0])) { + opts.email = state.greenlockConf.email || state.config.email; + opts.agreeTos = state.greenlockConf.agree || state.greenlockConf.agreeTos || state.config.agreeTos; + cb(null, { options: opts, certs: certs }); + return; + } + + 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; +} + function startTelebitRemote(rawCb) { - 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; + 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); - 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 - // we need to make sure we don't get rate-limit spammed - // with wildcard domains - // TODO: finish implementing dynamic dns for wildcard certs - if (getServername(state.servernames, opts.domains[0])) { - opts.email = state.greenlockConf.email || state.config.email; - opts.agreeTos = state.greenlockConf.agree || state.greenlockConf.agreeTos || state.config.agreeTos; - cb(null, { options: opts, certs: certs }); - return; - } - - //cb(new Error("servername not found in allowed list")); - } - }; - state.insecure = state.config.relay_ignore_invalid_certificates; - // { relay, config, servernames, ports, sortingHat, net, insecure, token, handlers, greenlockConfig } - function onError(err) { + myRemote = null; + console.log('DEBUG err', err); // Likely causes: // * DNS lookup failed (no Internet) // * Rejected (bad authn) @@ -843,7 +815,7 @@ function startTelebitRemote(rawCb) { // DNS issue, probably network is disconnected setTimeout(function () { startTelebitRemote(rawCb); - }, 90 * 1000); + }, 10 * 1000); return; } if ('function' === typeof rawCb) { @@ -854,12 +826,14 @@ function startTelebitRemote(rawCb) { } } 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({ relay: state.relay , wss: state.wss , config: state.config , otp: state.otp - , sortingHat: state.sortingHat + , sortingHat: state.config.sortingHat , net: state.net , insecure: state.insecure , token: state.token || state.pretoken // instance @@ -868,10 +842,83 @@ function startTelebitRemote(rawCb) { , handlers: state.handlers , greenlockConfig: state.greenlockConfig }, function () { + console.log('DEBUG on connect'); myRemote.removeListener('error', onError); + myRemote.once('error', retryLoop); rawCb(null, myRemote); }); + function retryLoop() { + // disconnected somehow + myRemote.destroy(); + myRemote = null; + setTimeout(function () { + startTelebitRemote(function () {}); + }, 10 * 1000); + } 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(); }); } diff --git a/lib/remote.js b/lib/remote.js index bb42cbc..b677724 100644 --- a/lib/remote.js +++ b/lib/remote.js @@ -27,14 +27,16 @@ function TelebitRemote(state) { } EventEmitter.call(this); var me = this; + var priv = {}; - var defaultHttpTimeout = (2 * 60); - var activityTimeout = state.activityTimeout || (defaultHttpTimeout - 5) * 1000; + //var defaultHttpTimeout = (2 * 60); + //var activityTimeout = state.activityTimeout || (defaultHttpTimeout - 5) * 1000; + var activityTimeout = 6 * 1000; var pongTimeout = state.pongTimeout || 10*1000; // 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 // they can get a promise that will provide feedback about invalid tokens. - var tokens = []; + priv.tokens = []; var auth; if(!state.sortingHat) { state.sortingHat = "./sorting-hat.js"; @@ -43,7 +45,7 @@ function TelebitRemote(state) { if ('undefined' === state.token) { throw new Error("passed string 'undefined' as token"); } - tokens.push(state.token); + priv.tokens.push(state.token); } var wstunneler; @@ -51,11 +53,11 @@ function TelebitRemote(state) { var authsent = false; var initialConnect = true; - var localclients = {}; + priv.localclients = {}; var pausedClients = []; var clientHandlers = { add: function (conn, cid, tun) { - localclients[cid] = conn; + priv.localclients[cid] = conn; console.info("[connect] new client '" + cid + "' for '" + tun.name + ":" + tun.serviceport + "' " + "(" + 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 // 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. - var bufSize = wsHandlers.sendMessage(Packer.packHeader(tun, chunk)); + var bufSize = sendMessage(Packer.packHeader(tun, chunk)); // Sending 2 messages instead of copying the buffer - var bufSize2 = wsHandlers.sendMessage(chunk); + var bufSize2 = sendMessage(chunk); if (pausedClients.length || (bufSize + bufSize2) > 1024*1024) { // console.log('[onLocalData] paused connection', cid, 'to allow websocket to catch up'); conn.pause(); @@ -95,7 +97,7 @@ function TelebitRemote(state) { console.info("[onLocalEnd] connection '" + cid + "' ended, will probably close soon"); conn.tunnelClosing = true; if (!sentEnd) { - wsHandlers.sendMessage(Packer.packHeader(tun, null, 'end')); + sendMessage(Packer.packHeader(tun, null, 'end')); sentEnd = true; } }); @@ -103,22 +105,22 @@ function TelebitRemote(state) { console.info("[onLocalError] connection '" + cid + "' errored:", err); if (!sentEnd) { 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; } }); 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)'); if (!sentEnd) { - wsHandlers.sendMessage(Packer.packHeader(tun, null, hadErr && 'error' || 'end')); + sendMessage(Packer.packHeader(tun, null, hadErr && 'error' || 'end')); sentEnd = true; } }); } , write: function (cid, opts) { - var conn = localclients[cid]; + var conn = priv.localclients[cid]; if (!conn) { return false; } @@ -136,12 +138,12 @@ function TelebitRemote(state) { if (!conn.remotePaused && conn.bufferSize > 1024*1024) { var packBody = true; - wsHandlers.sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'pause', packBody)); + sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'pause', packBody)); conn.remotePaused = true; conn.once('drain', function () { var packBody = true; - wsHandlers.sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'resume', packBody)); + sendMessage(Packer.packHeader(opts, conn.tunnelRead, 'resume', packBody)); conn.remotePaused = false; }); } @@ -149,13 +151,13 @@ function TelebitRemote(state) { } , closeSingle: function (cid) { - if (!localclients[cid]) { + if (!priv.localclients[cid]) { return; } console.log('[closeSingle]', cid); PromiseA.resolve().then(function () { - var conn = localclients[cid]; + var conn = priv.localclients[cid]; conn.tunnelClosing = true; conn.end(); @@ -173,41 +175,49 @@ function TelebitRemote(state) { }); }); }).then(function () { - if (localclients[cid]) { + if (priv.localclients[cid]) { console.warn('[closeSingle]', cid, 'connection still present after calling `end`'); - localclients[cid].destroy(); + priv.localclients[cid].destroy(); return timeoutPromise(500); } }).then(function () { - if (localclients[cid]) { + if (priv.localclients[cid]) { console.error('[closeSingle]', cid, 'connection still present after calling `destroy`'); - delete localclients[cid]; + delete priv.localclients[cid]; } }).catch(function (err) { console.error('[closeSingle] failed to close connection', cid, err.toString()); - delete localclients[cid]; + delete priv.localclients[cid]; }); } , closeAll: function () { console.log('[closeAll]'); - Object.keys(localclients).forEach(function (cid) { + Object.keys(priv.localclients).forEach(function (cid) { clientHandlers.closeSingle(cid); }); } , count: function () { - return Object.keys(localclients).length; + return Object.keys(priv.localclients).length; } }; 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) { var id = Math.ceil(1e9 * Math.random()); var cmd = [id, name].concat(Array.prototype.slice.call(arguments, 1)); if (state.debug) { console.log('[DEBUG] command sending', cmd); } var packBody = true; - wsHandlers.sendMessage(Packer.packHeader(null, cmd, 'control', packBody)); + sendMessage(Packer.packHeader(null, cmd, 'control', packBody)); setTimeout(function () { if (pendingCommands[id]) { 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) { console.warn("[telebit] state.handlers['" + cmd[1] + "'] not set"); console.warn(cmd[2]); @@ -303,7 +295,21 @@ function TelebitRemote(state) { if (cmd[1] === 'hello') { 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) { connCallback(); } @@ -331,7 +337,7 @@ function TelebitRemote(state) { } 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) { @@ -369,28 +375,28 @@ function TelebitRemote(state) { , onpause: function (opts) { var cid = Packer.addrToId(opts); - if (localclients[cid]) { - console.log("[TunnelPause] pausing '"+cid+"', remote received", opts.data.toString(), 'of', localclients[cid].tunnelWritten, 'sent'); - localclients[cid].manualPause = true; - localclients[cid].pause(); + if (priv.localclients[cid]) { + console.log("[TunnelPause] pausing '"+cid+"', remote received", opts.data.toString(), 'of', priv.localclients[cid].tunnelWritten, 'sent'); + priv.localclients[cid].manualPause = true; + priv.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; - // 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) { var cid = Packer.addrToId(opts); - if (localclients[cid]) { - console.log("[TunnelResume] resuming '"+cid+"', remote received", opts.data.toString(), 'of', localclients[cid].tunnelWritten, 'sent'); - localclients[cid].manualPause = false; - localclients[cid].resume(); + if (priv.localclients[cid]) { + console.log("[TunnelResume] resuming '"+cid+"', remote received", opts.data.toString(), 'of', priv.localclients[cid].tunnelWritten, 'sent'); + priv.localclients[cid].manualPause = false; + priv.localclients[cid].resume(); } else { console.log('[TunnelResume] remote tried resuming finished connection', cid); // 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,56 +413,73 @@ function TelebitRemote(state) { , _onConnectError: function (cid, opts, err) { console.info("[_onConnectError] opening '" + cid + "' failed because " + err.message); - wsHandlers.sendMessage(Packer.packHeader(opts, null, 'error')); + sendMessage(Packer.packHeader(opts, null, 'error')); } }; - var lastActivity; - var timeoutId; - var wsHandlers = { - refreshTimeout: function () { - lastActivity = Date.now(); + priv.timeoutId = null; + priv.lastActivity = Date.now(); + priv.refreshTimeout = function refreshTimeout() { + priv.lastActivity = Date.now(); + }; + priv.checkTimeout = function checkTimeout() { + if (!wstunneler) { + console.warn('checkTimeout called when websocket already closed'); + return; } - , checkTimeout: function () { - if (!wstunneler) { - console.warn('checkTimeout called when websocket already closed'); - return; - } - // Determine how long the connection has been "silent", ie no activity. - var silent = Date.now() - lastActivity; + // Determine how long the connection has been "silent", ie no activity. + var silent = Date.now() - priv.lastActivity; - // 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. - if (silent < activityTimeout) { - timeoutId = setTimeout(wsHandlers.checkTimeout, activityTimeout-silent); - } - - // Otherwise we check to see if the pong has also timed out, and if not we send a ping - // and call this function again when the pong will have timed out. - else if (silent < activityTimeout + pongTimeout) { - console.log('pinging tunnel server'); - try { - wstunneler.ping(); - } catch (err) { - console.warn('failed to ping tunnel server', err); - } - timeoutId = setTimeout(wsHandlers.checkTimeout, pongTimeout); - } - - // Last case means the ping we sent before didn't get a response soon enough, so we - // need to close the websocket connection. - else { - console.log('connection timed out'); - wstunneler.close(1000, 'connection timeout'); - } + // 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. + if (silent < activityTimeout) { + priv.timeoutId = setTimeout(priv.checkTimeout, activityTimeout-silent); } - , onOpen: function () { + // Otherwise we check to see if the pong has also timed out, and if not we send a ping + // and call this function again when the pong will have timed out. + else if (silent < activityTimeout + pongTimeout) { + console.log('pinging tunnel server'); + try { + wstunneler.ping(); + } catch (err) { + console.warn('failed to ping tunnel server', err); + } + priv.timeoutId = setTimeout(priv.checkTimeout, pongTimeout); + } + + // Last case means the ping we sent before didn't get a response soon enough, so we + // need to close the websocket connection. + else { + console.log('connection timed out'); + 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); + + 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) + "'"); - wsHandlers.refreshTimeout(); - - timeoutId = setTimeout(wsHandlers.checkTimeout, activityTimeout); - + me.emit('connect'); + priv.refreshTimeout(); + priv.timeoutId = setTimeout(priv.checkTimeout, activityTimeout); wstunneler._socket.on('drain', function () { // 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 @@ -473,24 +496,13 @@ function TelebitRemote(state) { conn.resume(); } }); - 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; - } - - , onClose: function () { - clearTimeout(timeoutId); - wstunneler = null; + }); + wstunneler.on('close', function () { + console.log("DEBUG closing"); + clearTimeout(priv.timeoutId); clientHandlers.closeAll(); var error = new Error('websocket connection closed before response'); @@ -503,108 +515,19 @@ function TelebitRemote(state) { } me.emit('close'); - /* - return; - 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); - /* - 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) { - if (wstunneler) { - try { - wstunneler.send(msg, {binary: true}); - return wstunneler.bufferedAmount; - } catch (err) { - // 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) { - console.warn('[sendMessage] error sending websocket message', err); - } - } - } - } - }; - - var connPromise; - me.connect = function connect() { - if (wstunneler) { - console.warn('attempted to connect with connection already active'); - return; - } - if (!tokens.length) { - if (state.config.email) { - auth = { - subject: state.config.email - , subject_scheme: 'mailto' - // TODO create domains list earlier - , scope: Object.keys(state.config.servernames || {}).join(',') - , otp: state.otp - , hostname: os.hostname() - // Used for User-Agent - , os_type: os.type() - , os_platform: os.platform() - , os_release: os.release() - , 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); + wstunneler.on('error', function (err) { + me.emit('error', err); + }); // Our library will automatically handle sending the pong respose to ping requests. - wstunneler.on('ping', wsHandlers.refreshTimeout); - wstunneler.on('pong', wsHandlers.refreshTimeout); + wstunneler.on('ping', priv.refreshTimeout); + wstunneler.on('pong', function () { + console.log('DEBUG got pong'); + priv.refreshTimeout(); + }); wstunneler.on('message', function (data, flags) { - wsHandlers.refreshTimeout(); + priv.refreshTimeout(); if (data.error || '{' === data[0]) { console.log(data); return; @@ -613,103 +536,34 @@ function TelebitRemote(state) { }); }; me.end = function() { - tokens.length = 0; - if (timeoutId) { - clearTimeout(timeoutId); - timeoutId = null; + priv.tokens.length = 0; + if (priv.timeoutId) { + clearTimeout(priv.timeoutId); + priv.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); + wstunneler.close(1000, 're-connect'); + wstunneler.on('close', function () { + me.emit('end'); }); - - 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._tokenFromState = function (state) { + return { + subject: state.config.email + , subject_scheme: 'mailto' + // TODO create domains list earlier + , scope: Object.keys(state.config.servernames || {}).join(',') + , otp: state.otp + , hostname: os.hostname() + // Used for User-Agent + , os_type: os.type() + , os_platform: os.platform() + , os_release: os.release() + , os_arch: os.arch() + }; +}; TelebitRemote.create = function (opts) { return new TelebitRemote(opts);