Compare commits
1 Commits
| Author | SHA1 | Date |
|---|---|---|
 AJ ONeal
|
49835c1a6a | 6 years ago |
4 changed files with 478 additions and 330 deletions
@ -0,0 +1,245 @@ |
|||
'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; |
|||
}; |
|||
@ -0,0 +1,75 @@ |
|||
// https://github.com/websockets/ws/issues/596
|
|||
var WSStream = module.exports = (function(){ |
|||
var util = require('util'); |
|||
var stream = require('stream'); |
|||
var Duplex = stream.Duplex || require('readable-stream').Duplex; |
|||
var WSStream = function(ws, wsOptions, streamingOptions) { // Only the websocket (ws) is mandatory
|
|||
// Proper duplex stream with automatic flow control (backpressure) management
|
|||
if(!(this instanceof WSStream)) return new WSStream(ws, wsOptions, streamingOptions); |
|||
if(!(wsOptions instanceof Object)) wsOptions = {binary: false}; |
|||
Duplex.call(this, streamingOptions); |
|||
this.waitingForData = true; |
|||
this.writeModBufferEmpty = true; |
|||
this.webSocket = ws; |
|||
this.webSocketOptions = wsOptions; |
|||
this.on('finish', finishEventHandler(this)); |
|||
ws._socket.write = writeMod(ws._socket.write, this); |
|||
ws._socket.on('drain', drainEventHandler(this)); |
|||
ws._socket.on('error', errorRouter(this)); |
|||
ws.on('close', closeEventHandler(this)); |
|||
ws.on('message', messageHandler(this)); |
|||
}, |
|||
finishEventHandler = function(self) { |
|||
return function() { |
|||
if(self.webSocket !== null) { |
|||
self.webSocket.close(); |
|||
self.webSocket = null; |
|||
}; |
|||
self.emit('close'); |
|||
}; |
|||
}, |
|||
writeMod = function(nativeWriteFunction, self) { |
|||
return function() { |
|||
return self.writeModBufferEmpty = nativeWriteFunction.apply(this, arguments); |
|||
}; |
|||
}, |
|||
drainEventHandler = function(self) { |
|||
return function() { |
|||
self.writeModBufferEmpty = true; |
|||
self.emit('drain'); |
|||
}; |
|||
}, |
|||
closeEventHandler = function(self) { |
|||
return function() { |
|||
self.push(null); |
|||
} |
|||
}, |
|||
errorRouter = function(self) { |
|||
return function() { |
|||
self.emit.apply(self, ['error'].concat(arguments)); |
|||
}; |
|||
}, |
|||
messageHandler = function(self) { |
|||
return function(data) { |
|||
if(!self.push(data) && self.waitingForData) { |
|||
self.webSocket._socket.pause(); |
|||
self.waitingForData = false; |
|||
} |
|||
}; |
|||
}; |
|||
util.inherits(WSStream, Duplex); |
|||
WSStream.prototype._read = function(size) { |
|||
if(!this.waitingForData) { |
|||
this.waitingForData = true; |
|||
this.webSocket._socket.resume(); |
|||
} |
|||
}; |
|||
WSStream.prototype._write = function(chunk, encoding, callback) { |
|||
this.webSocket.send(chunk, this.webSocketOptions, callback); |
|||
return this.writeModBufferEmpty; |
|||
}; |
|||
return WSStream; |
|||
}()); |
|||
module.exports.create = function (ws, wsOpts, streamOpts) { |
|||
return new WSStream(ws, wsOpts, streamOpts); |
|||
}; |
|||
Loading…
Reference in new issue