From ad5e4c205147e5654f33731c1de849b28c7e0d65 Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:21:51 +0800 Subject: [PATCH 01/13] move hole punch code to app --- neptune/mega/tunnel/api.js | 38 +++- neptune/mega/tunnel/main.js | 10 +- neptune/mega/tunnel/punch.js | 382 +++++++++++++++++++++++++++++++++++ 3 files changed, 420 insertions(+), 10 deletions(-) create mode 100644 neptune/mega/tunnel/punch.js diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index 46bc584e9..0b72cc16e 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -1,4 +1,4 @@ -export default function ({ app, mesh }) { +export default function ({ app, mesh, punch }) { var currentListens = [] var currentTargets = {} @@ -39,9 +39,12 @@ export default function ({ app, mesh }) { function getInbound(ep, protocol, name) { if (ep === app.endpoint.id) { return getLocalConfig().then( - config => config.inbound.find( - i => i.protocol === protocol && i.name === name - ) || null + config => { + var inbound = config.inbound.find( + i => i.protocol === protocol && i.name === name + ) || null + var hole = punch.findHole(ep) + } ) } else { return requestPeer(ep, new Message( @@ -201,19 +204,36 @@ export default function ({ app, mesh }) { var protocol = i.protocol var name = i.name var listens = i.listens - var $selectedEP + + var getHole = () => { + var h = punch.findHole($selectedEP) + if(!h) { + h = punch.createInboundHole(ep) + } + if(h.ready) { + return h + } + return null + } + var connectPeer = pipeline($=>$ .connectHTTPTunnel( new Message({ method: 'CONNECT', path: `/api/outbound/${protocol}/${name}`, }) - ).to($=>$ - .muxHTTP().to($=>$ - .pipe(() => mesh.connect($selectedEP)) + ).to(() => { + var hole = getHole() + if(hole) { + return h.directSession() + } + return pipeline($=>$ + .muxHTTP().to($=>$ + .pipe(() => mesh.connect($selectedEP)) + ) ) - ) + }) .onEnd(() => app.log(`Disconnected from ep ${$selectedEP} for ${protocol}/${name}`)) ) diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 3c2e6b39d..415f4ca6a 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -1,8 +1,10 @@ import initAPI from './api.js' import initCLI from './cli.js' +import initHole from './punch.js' export default function ({ app, mesh, utils }) { - var api = initAPI({ app, mesh }) + var punch = initHole({ app }) + var api = initAPI({ app, mesh, punch }) var cli = initCLI({ app, mesh, utils, api }) var $ctx @@ -73,6 +75,12 @@ export default function ({ app, mesh, utils }) { }), }, + '/api/punch/{otbEp}/{inbEp}': { + 'GET': responder(({otbEp, inbEp}) => { + + }) + }, + '*': { 'GET': responder((_, req) => { return Promise.resolve(gui.serve(req) || response(404)) diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js new file mode 100644 index 000000000..f335fbcc7 --- /dev/null +++ b/neptune/mega/tunnel/punch.js @@ -0,0 +1,382 @@ +export default function({ app }) { + var holes = {} + + + // Only available for symmetric NAT + function Hole(ep) { + // FIXME: throw on init fail? + // TODO: Add detailed comment. + + // create hole when find ep. + // hub save ep pair that fail or success + var bound = '0.0.0.0:' + randomPort() // local port that the hole using + var destIP // dest ip out of NAT + var destPort // dest port out of NAT + var role = null // server or client + // var managedSvc = {} + + // closed forwarding connecting(ready punching) connected fail + var state = 'closed' // inner state + var ready = false // exposed state + var $hubConnection = null + var $connection = null + var $hub = null + var $pHub = new pipeline.Hub + var $session + var $response + + console.info("Bound: ", bound) + + // Check if ep is self. + if(ep === app.endpoint.id) { + throw 'Must not create a hole to self' + } + + // A temp tunnel to help hub gather NAT info. + var hubSession = pipeline($ => $ + .onStart(() => { + return selectHub(ep).then(res => { + if(res) { + $hub = res + } else { + state = 'fail' + } + return new Data + }) + }) + .muxHTTP(() => ep + "hub", { version: 2 }).to($ => $ + .connectTLS({ + ...tlsOptions, + onState: (session) => { + var err = session.error + if (err) state = 'fail' + } + }).to($ => $ + .connect(() => $hub, { + onState: function (conn) { + console.info('Tmp connection state ', conn, state) + $hubConnection = conn + if (conn.state === 'open') { + // SOL_SOCKET & SO_REUSEPORT + conn.socket.setRawOption(1, 15, new Data([1])) + } else if (conn.state === 'connected' && state === 'closed') { + console.info("Tmp Hub Connection created") + state = 'forwarding' + reverseServer.spawn() + } else if (conn.state === 'closed' && state === 'forwarding') { + // this means closed unexpectedly + state = 'fail' + leave() + } + }, + bind: bound, + }) + ) + ) + ) + + var reverseServer = pipeline($ => $ + .onStart(new Data) + .repeat(() => new Timeout(5).wait().then(() => { + if (state != 'forwarding') { + $hubConnection.close() + return false + } + return true + })).to($ => $ + .loop($ => $ + .connectHTTPTunnel( + new Message({ + method: 'CONNECT', + path: `/api/punch/${config.agent.id}/${ep}`, + }) + ) + .to(hubSession) + .pipe(servePunch) + ) + ) + ) + + var servePunch = pipeline($ => $ + .demuxHTTP().to($ => $.pipe(() => { + var routes = Object.entries({ + '/api/punch/{srcEp}/{destEp}/sync': { + // Hub sent synchronize message. Once receive, start punch. + // Agent <- Hub -> Remote Agent. + 'POST': function (params, req) { + // TODO use params to check + var body = JSON.decode(req.body) + destIP = body.destIP + destPort = body.port + state = 'ready' + + punch(destIP, destPort) + } + }, + }).map(function ([path, methods]) { + var match = new http.Match(path) + var handler = function (params, req) { + var f = methods[req.head.method] + if (f) return f(params, req) + return response(405) + } + return { match, handler } + }) + + return pipeline($ => $ + .replaceMessage( + function (req) { + var params + var path = req.head.path + var route = routes.find(r => Boolean(params = r.match(path))) + if (route) { + try { + var res = route.handler(params, req) + return res instanceof Promise ? res.catch(responseError) : res + } catch (e) { + return responseError(e) + } + } + meshError('Invalid api call from hub') + return response(404) + } + ) + ) + } + )) + ) + + function directSession() { + // TODO !!! state error would happen when network is slow + // must handle this + + if (!role) meshError('Hole not init correctly') + if ($session) return $session + + // TODO: support TLS connection + if (role === 'client') { + // make session to server side directly + $session = pipeline($ => $ + .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ + .connect(() => `${destIP}:${destPort}`, { + onState: function (conn) { + if (conn.state === 'open') { + conn.socket.setRawOption(1, 15, new Data([1])) + } else if (conn.state === 'connected') { + logInfo(`Connected to remote ${destIP}:${destPort}`) + $connection = conn + state = 'connected' + } else if (conn.state === 'closed') { + logInfo(`Disconnected from remote ${destIP}:${destPort}`) + $connection = null + state = 'closed' + } + }, + bind: bound + }) + .handleStreamEnd(evt => { + if (evt.error) { + state = 'fail' + leave() + } + }) + ) + ) + + // reverse server for receiving requests + pipeline($ => $ + .onStart(new Data) + .repeat(() => new Timeout(5).wait().then(() => { + return state != 'fail' || state != 'closed' + })).to($ => $ + .loop($ => $ + .connectHTTPTunnel( + new Message({ + method: 'CONNECT', + path: `/api/punch/${config.agent.id}/${ep}`, + }) + ) + .to($session) + .pipe(serveHub) + ) + ) + ).spawn() + + } else if (role === 'server') { + pipy.listen(bound, 'tcp', serveHub) + + $session = pipeline($ => $ + .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ + .swap(() => $pHub) + ) + ) + } + + return $session + } + + // Send a request. + var request = pipeline($ => $ + .onStart(msg => { + console.info(`Reqesting Remote: ${JSON.encode(msg.head)}`) + return msg + }) + .pipe(() => { + if (state === 'connected') { + return directSession() + } else if(state != 'fail') { + return hubSession + } + + console.log(`State incorrect: ${state}, reject`) + return pipeline($=>$.dummy()) + }) + .handleMessage(msg => $response = msg) + .replaceMessage(new StreamEnd) + .onEnd(() => $response) + ) + + // use THE port sending request to hub. + function requestPunch() { + // FIXME: add state check + // state = 'connecting' + role = 'client' + + request.spawn(new Message({ + method: 'GET', + path: `/api/punch/${config.agent.id}/${ep}/request`, + })) + } + + // TODO add cert info into response + function acceptPunch() { + // state = 'connecting' + role = 'server' + + request.spawn(new Message({ + method: 'POST', + path: `/api/punch/${config.agent.id}/${ep}/reqeust`, + })) + } + + function punch(destIP, destPort) { + // receive TLS options + // connectTLS + // connectLocal or connectRemote + + // TODO add retry logic here + state = 'punching' + makeFakeCall(destIP, destPort) + $session = directSession() + heartbeat() // activate the session pipeline + } + + function makeRespTunnel() { + // TODO add state check + state = 'connected' + + return pipeline($ => $ + .acceptHTTPTunnel(() => response200()).to($ => $ + .onStart(new Data) + .swap(() => $pHub) + .onEnd(() => console.info(`Direct Connection from ${ep} lost`)) + ) + ) + } + + // send a SYN to dest, expect no return. + // this will cheat the firewall to allow inbound connection from dest. + function makeFakeCall(destIP, destPort) { + pipy().task().onStart(new Data).connect(`${destIP}:${destPort}`, { + bind: bound, + onState: function (conn) { + // REUSEPORT + if (conn.state === 'open') conn.socket.setRawOption(1, 15, new Data([1])) + + // abort this connection. + if (conn.state === 'connecting') conn.close() + } + }) + } + + + function heartbeat() { + request.spawn( + new Message( + { method: 'POST', path: '/api/status' }, + JSON.encode({ name: config.agent.name }) + ) + ) + } + + function leave() { + $hubConnection?.close() + $connection?.close() + $hubConnection = null + $connection = null + if (state != 'fail') state = 'closed' + + // try to find the hub holding this hole + if($hub) { + var parent = hubs.find(h => $hub === h.address) + parent.updateHoles() + } + } + + return { + role, + ready, + destIP, + destPort, + requestPunch, + acceptPunch, + punch, + makeRespTunnel, + directSession, + heartbeat, + leave, + } + } // End of Hole + + function createInboundHole(ep) { + try { + var hole = Hole(ep) + } catch { + app.log(`Failed to create Inbound Hole, peer ${ep}`) + } + + hole.requestPunch() + } + + function createOutboundHole(proto, name) { + try { + var hole = Hole(ep) + } catch { + app.log(`Failed to create Inbound Hole, peer ${ep}`) + } + + hole.acceptPunch() + } + + function deleteHole(ep) { + + } + + function findHole(ep) { + + } + + function isHoleReady(hole) { + + } + + + return { + holes, + createInboundHole, + createOutboundHole, + deleteHole, + findHole, + isHoleReady, + } +} From 842cb543daef7d66709e705f5f28793f22dcff75 Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:24:05 +0800 Subject: [PATCH 02/13] adjust for new app api --- neptune/mega/tunnel/api.js | 42 ++++--- neptune/mega/tunnel/main.js | 21 +++- neptune/mega/tunnel/punch.js | 219 +++++++---------------------------- 3 files changed, 89 insertions(+), 193 deletions(-) diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index 0b72cc16e..ecf9dd619 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -1,6 +1,7 @@ export default function ({ app, mesh, punch }) { var currentListens = [] var currentTargets = {} + var bound = '0.0.0.0:' + punch.randomPort() function allEndpoints() { return mesh.discover() @@ -89,6 +90,7 @@ export default function ({ app, mesh, punch }) { } else { all.push(ent) } + punch.createInboundHole(ep, ent, bound, requestPeer) setLocalConfig(config) applyLocalConfig(config) }) @@ -119,6 +121,7 @@ export default function ({ app, mesh, punch }) { } else { all.push(ent) } + punch.createOutboundHole(ep, ent, bound, requestPeer) setLocalConfig(config) applyLocalConfig(config) }) @@ -173,6 +176,17 @@ export default function ({ app, mesh, punch }) { } } + function createHole(ep, ip, port) { + var h = punch.findHole(ep) + if(h) return h + + return punch.createInboundHole(ep, ) + } + + function deleteHole(ep) { + punch.deleteHole(ep) + } + function getLocalConfig() { return mesh.read('/local/config.json').then( data => data ? JSON.decode(data) : { inbound: [], outbound: [] } @@ -206,17 +220,6 @@ export default function ({ app, mesh, punch }) { var listens = i.listens var $selectedEP - var getHole = () => { - var h = punch.findHole($selectedEP) - if(!h) { - h = punch.createInboundHole(ep) - } - if(h.ready) { - return h - } - return null - } - var connectPeer = pipeline($=>$ .connectHTTPTunnel( new Message({ @@ -224,8 +227,8 @@ export default function ({ app, mesh, punch }) { path: `/api/outbound/${protocol}/${name}`, }) ).to(() => { - var hole = getHole() - if(hole) { + var hole = punch.findHole($selectedEP) + if(hole && hole.ready) { return h.directSession() } return pipeline($=>$ @@ -306,7 +309,13 @@ export default function ({ app, mesh, punch }) { return pipeline($=>$ .onStart(req) .muxHTTP().to($=>$ - .pipe(mesh.connect(ep)) + .pipe(mesh.connect(ep, { + bind: bound, + onState: conn => { + if(conn.state === 'open') + conn.socket.setRawOption(1, 15, new Data([1,0, 0, 0])) + } + })) ) .replaceMessage(res => { $response = res @@ -363,6 +372,10 @@ export default function ({ app, mesh, punch }) { ) ) + var makeRespTunnel = pipeline($=>$ + .pipe(punch.makeRespTunnel()) + ) + getLocalConfig().then(applyLocalConfig) return { @@ -376,6 +389,7 @@ export default function ({ app, mesh, punch }) { deleteInbound, deleteOutbound, servePeerInbound, + makeRespTunnel, } } diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 415f4ca6a..5506547f5 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -75,9 +75,14 @@ export default function ({ app, mesh, utils }) { }), }, - '/api/punch/{otbEp}/{inbEp}': { - 'GET': responder(({otbEp, inbEp}) => { + '/api/punch/{destEp}': { + 'GET': responder(({destEp}) => { + return api.createHole(destEp) + }), + 'DELETE': responder(({destEp}) => { + api.deleteHole(destEp) + return response(204) }) }, @@ -136,6 +141,18 @@ export default function ({ app, mesh, utils }) { 'CONNECT': api.servePeerInbound, }, + + '/api/punch/{role}': { + 'GET': responder(({role}) => { + var ep = $ctx.id + var ip = $ctx.ip + var port = $ctx.port + + return api.createHole(ep, ip, port, role) + }), + + 'CONNECT': api.makeRespTunnel + }, }) return pipeline($=>$ diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index f335fbcc7..47eb0ebca 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -1,156 +1,31 @@ export default function({ app }) { - var holes = {} - - // Only available for symmetric NAT - function Hole(ep) { + function Hole(ep, bound, ent, request) { // FIXME: throw on init fail? // TODO: Add detailed comment. - // create hole when find ep. - // hub save ep pair that fail or success - var bound = '0.0.0.0:' + randomPort() // local port that the hole using - var destIP // dest ip out of NAT - var destPort // dest port out of NAT - var role = null // server or client - // var managedSvc = {} + var destIP = ent.ip + var destPort = ent.port + var role = null // closed forwarding connecting(ready punching) connected fail var state = 'closed' // inner state var ready = false // exposed state - var $hubConnection = null var $connection = null - var $hub = null var $pHub = new pipeline.Hub var $session var $response - console.info("Bound: ", bound) - // Check if ep is self. if(ep === app.endpoint.id) { throw 'Must not create a hole to self' } - // A temp tunnel to help hub gather NAT info. - var hubSession = pipeline($ => $ - .onStart(() => { - return selectHub(ep).then(res => { - if(res) { - $hub = res - } else { - state = 'fail' - } - return new Data - }) - }) - .muxHTTP(() => ep + "hub", { version: 2 }).to($ => $ - .connectTLS({ - ...tlsOptions, - onState: (session) => { - var err = session.error - if (err) state = 'fail' - } - }).to($ => $ - .connect(() => $hub, { - onState: function (conn) { - console.info('Tmp connection state ', conn, state) - $hubConnection = conn - if (conn.state === 'open') { - // SOL_SOCKET & SO_REUSEPORT - conn.socket.setRawOption(1, 15, new Data([1])) - } else if (conn.state === 'connected' && state === 'closed') { - console.info("Tmp Hub Connection created") - state = 'forwarding' - reverseServer.spawn() - } else if (conn.state === 'closed' && state === 'forwarding') { - // this means closed unexpectedly - state = 'fail' - leave() - } - }, - bind: bound, - }) - ) - ) - ) - - var reverseServer = pipeline($ => $ - .onStart(new Data) - .repeat(() => new Timeout(5).wait().then(() => { - if (state != 'forwarding') { - $hubConnection.close() - return false - } - return true - })).to($ => $ - .loop($ => $ - .connectHTTPTunnel( - new Message({ - method: 'CONNECT', - path: `/api/punch/${config.agent.id}/${ep}`, - }) - ) - .to(hubSession) - .pipe(servePunch) - ) - ) - ) - - var servePunch = pipeline($ => $ - .demuxHTTP().to($ => $.pipe(() => { - var routes = Object.entries({ - '/api/punch/{srcEp}/{destEp}/sync': { - // Hub sent synchronize message. Once receive, start punch. - // Agent <- Hub -> Remote Agent. - 'POST': function (params, req) { - // TODO use params to check - var body = JSON.decode(req.body) - destIP = body.destIP - destPort = body.port - state = 'ready' - - punch(destIP, destPort) - } - }, - }).map(function ([path, methods]) { - var match = new http.Match(path) - var handler = function (params, req) { - var f = methods[req.head.method] - if (f) return f(params, req) - return response(405) - } - return { match, handler } - }) - - return pipeline($ => $ - .replaceMessage( - function (req) { - var params - var path = req.head.path - var route = routes.find(r => Boolean(params = r.match(path))) - if (route) { - try { - var res = route.handler(params, req) - return res instanceof Promise ? res.catch(responseError) : res - } catch (e) { - return responseError(e) - } - } - meshError('Invalid api call from hub') - return response(404) - } - ) - ) - } - )) - ) - function directSession() { // TODO !!! state error would happen when network is slow // must handle this - if (!role) meshError('Hole not init correctly') + if (!role) throw 'Hole not init correctly' if ($session) return $session // TODO: support TLS connection @@ -161,13 +36,13 @@ export default function({ app }) { .connect(() => `${destIP}:${destPort}`, { onState: function (conn) { if (conn.state === 'open') { - conn.socket.setRawOption(1, 15, new Data([1])) + conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) } else if (conn.state === 'connected') { - logInfo(`Connected to remote ${destIP}:${destPort}`) + app.log(`Connected to remote ${destIP}:${destPort}`) $connection = conn state = 'connected' } else if (conn.state === 'closed') { - logInfo(`Disconnected from remote ${destIP}:${destPort}`) + app.log(`Disconnected from remote ${destIP}:${destPort}`) $connection = null state = 'closed' } @@ -215,26 +90,6 @@ export default function({ app }) { return $session } - // Send a request. - var request = pipeline($ => $ - .onStart(msg => { - console.info(`Reqesting Remote: ${JSON.encode(msg.head)}`) - return msg - }) - .pipe(() => { - if (state === 'connected') { - return directSession() - } else if(state != 'fail') { - return hubSession - } - - console.log(`State incorrect: ${state}, reject`) - return pipeline($=>$.dummy()) - }) - .handleMessage(msg => $response = msg) - .replaceMessage(new StreamEnd) - .onEnd(() => $response) - ) // use THE port sending request to hub. function requestPunch() { @@ -244,7 +99,7 @@ export default function({ app }) { request.spawn(new Message({ method: 'GET', - path: `/api/punch/${config.agent.id}/${ep}/request`, + path: `/api/punch/server`, })) } @@ -254,8 +109,8 @@ export default function({ app }) { role = 'server' request.spawn(new Message({ - method: 'POST', - path: `/api/punch/${config.agent.id}/${ep}/reqeust`, + method: 'GET', + path: `/api/punch/client`, })) } @@ -310,24 +165,14 @@ export default function({ app }) { } function leave() { - $hubConnection?.close() $connection?.close() - $hubConnection = null $connection = null if (state != 'fail') state = 'closed' - - // try to find the hub holding this hole - if($hub) { - var parent = hubs.find(h => $hub === h.address) - parent.updateHoles() - } } return { role, ready, - destIP, - destPort, requestPunch, acceptPunch, punch, @@ -338,45 +183,65 @@ export default function({ app }) { } } // End of Hole - function createInboundHole(ep) { + var holes = new Map + + function updateHoles() { + holes.forEach((key, hole) => { + if (hole.state === 'fail' || hole.state === 'closed') { + hole.leave() + holes.delete(key) + } + }) + } + + function createInboundHole(ep, ent, bound, request) { + updateHoles() try { - var hole = Hole(ep) + var hole = Hole(ep, ent, bound, request) + hole.requestPunch() + holes.set(ep, hole) } catch { app.log(`Failed to create Inbound Hole, peer ${ep}`) } - hole.requestPunch() + return hole } - function createOutboundHole(proto, name) { + function createOutboundHole(ep, ent, bound, request) { + updateHoles() try { var hole = Hole(ep) + hole.acceptPunch() + holes.set(ep, hole) } catch { app.log(`Failed to create Inbound Hole, peer ${ep}`) } - hole.acceptPunch() + return hole } function deleteHole(ep) { - + var sel = findHole(ep) + if(!sel) return + sel.leave() + holes.delete(ep) } function findHole(ep) { - + updateHoles() + return holes.get(ep) } - function isHoleReady(hole) { - + function randomPort() { + return Number.parseInt(Math.random() * (65535 - 1024)) + 1024 } - return { holes, createInboundHole, createOutboundHole, deleteHole, findHole, - isHoleReady, + randomPort, } } From 5d29863af426ce562419ebbe6114250c26d370a0 Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:28:02 +0800 Subject: [PATCH 03/13] fix hub forwarded handshake --- neptune/mega/tunnel/api.js | 65 ++++++++++++++++------- neptune/mega/tunnel/main.js | 45 ++++++++++------ neptune/mega/tunnel/punch.js | 100 ++++++++++++++++++++++++++++------- 3 files changed, 156 insertions(+), 54 deletions(-) diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index ecf9dd619..3d8bd0d83 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -1,7 +1,6 @@ export default function ({ app, mesh, punch }) { var currentListens = [] var currentTargets = {} - var bound = '0.0.0.0:' + punch.randomPort() function allEndpoints() { return mesh.discover() @@ -90,7 +89,6 @@ export default function ({ app, mesh, punch }) { } else { all.push(ent) } - punch.createInboundHole(ep, ent, bound, requestPeer) setLocalConfig(config) applyLocalConfig(config) }) @@ -121,7 +119,6 @@ export default function ({ app, mesh, punch }) { } else { all.push(ent) } - punch.createOutboundHole(ep, ent, bound, requestPeer) setLocalConfig(config) applyLocalConfig(config) }) @@ -176,11 +173,35 @@ export default function ({ app, mesh, punch }) { } } - function createHole(ep, ip, port) { + function createHole(ep, ip, port, role) { var h = punch.findHole(ep) if(h) return h - return punch.createInboundHole(ep, ) + if(role === 'server') { + checkIP(ip) + checkPort(Number.parseInt(port)) + punch.createOutboundHole(ep, ip, port) + } else if(role === 'client') { + punch.createInboundHole(ep) + } + } + + function updateHoleInfo(ep, ip, port) { + checkIP(ip) + checkPort(Number.parseInt(port)) + console.info(`Updating nat info: peer ${ep}, ip ${ip}, port ${port}`) + punch.updateHoleInfo(ep, ip, port) + } + + function punchHole(ep) { + console.log(`Hole punching......`) + var hole = punch.findHole(ep) + // if(!hole) return null + hole.punch() + } + + function makeRespTunnel(ep) { + } function deleteHole(ep) { @@ -214,6 +235,8 @@ export default function ({ app, mesh, punch }) { currentListens = [] currentTargets = {} + console.info(`Applying config: ${JSON.encode(config)}`) + config.inbound.forEach(i => { var protocol = i.protocol var name = i.name @@ -226,17 +249,20 @@ export default function ({ app, mesh, punch }) { method: 'CONNECT', path: `/api/outbound/${protocol}/${name}`, }) - ).to(() => { + ).to($=>$.pipe(() => { + punch.createInboundHole($selectedEP) var hole = punch.findHole($selectedEP) if(hole && hole.ready) { + console.info("Using direct session") return h.directSession() } + console.info("Using hub forwarded session") return pipeline($=>$ .muxHTTP().to($=>$ .pipe(() => mesh.connect($selectedEP)) ) ) - }) + })) .onEnd(() => app.log(`Disconnected from ep ${$selectedEP} for ${protocol}/${name}`)) ) @@ -309,19 +335,16 @@ export default function ({ app, mesh, punch }) { return pipeline($=>$ .onStart(req) .muxHTTP().to($=>$ - .pipe(mesh.connect(ep, { - bind: bound, - onState: conn => { - if(conn.state === 'open') - conn.socket.setRawOption(1, 15, new Data([1,0, 0, 0])) - } - })) + .pipe(mesh.connect(ep)) ) .replaceMessage(res => { $response = res return new StreamEnd }) - .onEnd(() => $response) + .onEnd(() => { + console.info('Hub answers in api: ', $response) + return $response + }) ).spawn() } @@ -372,9 +395,9 @@ export default function ({ app, mesh, punch }) { ) ) - var makeRespTunnel = pipeline($=>$ - .pipe(punch.makeRespTunnel()) - ) + // var makeRespTunnel = pipeline($=>$ + // .pipe(punch.makeRespTunnel()) + // ) getLocalConfig().then(applyLocalConfig) @@ -388,8 +411,12 @@ export default function ({ app, mesh, punch }) { setOutbound, deleteInbound, deleteOutbound, - servePeerInbound, + createHole, + updateHoleInfo, makeRespTunnel, + punchHole, + deleteHole, + servePeerInbound, } } diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 5506547f5..9c0c6c8d2 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -3,7 +3,7 @@ import initCLI from './cli.js' import initHole from './punch.js' export default function ({ app, mesh, utils }) { - var punch = initHole({ app }) + var punch = initHole({ app, mesh }) var api = initAPI({ app, mesh, punch }) var cli = initCLI({ app, mesh, utils, api }) @@ -75,16 +75,16 @@ export default function ({ app, mesh, utils }) { }), }, - '/api/punch/{destEp}': { - 'GET': responder(({destEp}) => { - return api.createHole(destEp) - }), + // '/api/punch/{destEp}': { + // 'GET': responder(({destEp}) => { + // return api.createHole(destEp) + // }), - 'DELETE': responder(({destEp}) => { - api.deleteHole(destEp) - return response(204) - }) - }, + // 'DELETE': responder(({destEp}) => { + // api.deleteHole(destEp) + // return response(204) + // }) + // }, '*': { 'GET': responder((_, req) => { @@ -142,13 +142,24 @@ export default function ({ app, mesh, utils }) { 'CONNECT': api.servePeerInbound, }, - '/api/punch/{role}': { - 'GET': responder(({role}) => { - var ep = $ctx.id - var ip = $ctx.ip - var port = $ctx.port - - return api.createHole(ep, ip, port, role) + '/api/punch/{action}': { + 'GET': responder(({action}) => { + var ep = $ctx.peer.id + var ip = $ctx.peer.ip + var port = $ctx.peer.port + + console.log(`Punch Event: ${action} from ${ep} ${ip} ${port}`) + switch(action) { + case 'request': + api.createHole(ep, ip, port, 'server') + break + case 'accept': + api.updateHoleInfo(ep, ip, port) + break + case 'sync': + api.syncPunch(ep) + } + return Promise.resolve(response(200)) }), 'CONNECT': api.makeRespTunnel diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index 47eb0ebca..929d892e0 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -1,11 +1,12 @@ -export default function({ app }) { +export default function({ app, mesh }) { // Only available for symmetric NAT - function Hole(ep, bound, ent, request) { + function Hole(ep) { // FIXME: throw on init fail? // TODO: Add detailed comment. - var destIP = ent.ip - var destPort = ent.port + var bound = '0.0.0.0:' + randomPort() + var destIP = null + var destPort = null var role = null // closed forwarding connecting(ready punching) connected fail @@ -16,6 +17,8 @@ export default function({ app }) { var $session var $response + console.info(`Creating hole to peer ${ep}, bound ${bound}`) + // Check if ep is self. if(ep === app.endpoint.id) { throw 'Must not create a hole to self' @@ -25,7 +28,7 @@ export default function({ app }) { // TODO !!! state error would happen when network is slow // must handle this - if (!role) throw 'Hole not init correctly' + if (!role || !destIP || !destPort) throw 'Hole not init correctly' if ($session) return $session // TODO: support TLS connection @@ -90,6 +93,29 @@ export default function({ app }) { return $session } + function fwdRequest(req) { + return pipeline($=>$ + .onStart(req) + .muxHTTP().to($=>$ + .pipe(mesh.connect(ep, { + bind: bound, + onState: conn => { + if(conn.state === 'open') + conn.socket.setRawOption(1,15,new Data([1,0,0,0])) + } + })) + ) + .print() + .replaceMessage(res => { + $response = res + return new StreamEnd + }) + .onEnd(() => { + console.info('Hub Answers in hole: ', $response) + return $response + }) + ).spawn() + } // use THE port sending request to hub. function requestPunch() { @@ -97,10 +123,12 @@ export default function({ app }) { // state = 'connecting' role = 'client' - request.spawn(new Message({ + console.info("Requesting punch") + fwdRequest(new Message({ method: 'GET', - path: `/api/punch/server`, + path: '/api/punch/request', })) + new Timeout(15).wait().then(connectOrFail) } // TODO add cert info into response @@ -108,19 +136,28 @@ export default function({ app }) { // state = 'connecting' role = 'server' - request.spawn(new Message({ + console.info("Accepting punch") + fwdRequest(new Message({ method: 'GET', - path: `/api/punch/client`, + path: '/api/punch/accept', })) + new Timeout(15).wait().then(connectOrFail) } - function punch(destIP, destPort) { + function updateNatInfo(ip, port) { + destIP = ip + destPort = port + } + + function punch() { // receive TLS options // connectTLS // connectLocal or connectRemote // TODO add retry logic here state = 'punching' + + console.info("Punching...") makeFakeCall(destIP, destPort) $session = directSession() heartbeat() // activate the session pipeline @@ -139,9 +176,16 @@ export default function({ app }) { ) } + function connectOrFail() { + console.info(`Failed unpunchable hole: ${ep}`) + if(state != 'connected') state = fail + updateHoles() + } + // send a SYN to dest, expect no return. // this will cheat the firewall to allow inbound connection from dest. function makeFakeCall(destIP, destPort) { + console.info("Making fake call") pipy().task().onStart(new Data).connect(`${destIP}:${destPort}`, { bind: bound, onState: function (conn) { @@ -156,7 +200,7 @@ export default function({ app }) { function heartbeat() { - request.spawn( + fwdRequest.spawn( new Message( { method: 'POST', path: '/api/status' }, JSON.encode({ name: config.agent.name }) @@ -175,6 +219,7 @@ export default function({ app }) { ready, requestPunch, acceptPunch, + updateNatInfo, punch, makeRespTunnel, directSession, @@ -192,34 +237,52 @@ export default function({ app }) { holes.delete(key) } }) + console.info(`Holes after updating: `, holes) } - function createInboundHole(ep, ent, bound, request) { + function createInboundHole(ep, bound, request) { updateHoles() + if(findHole(ep)) return + console.info(`Creating Inbound Hole to ${ep}`) try { - var hole = Hole(ep, ent, bound, request) + var hole = Hole(ep, bound, request) hole.requestPunch() holes.set(ep, hole) - } catch { - app.log(`Failed to create Inbound Hole, peer ${ep}`) + } catch(err) { + hole = null + app.log(`Failed to create Inbound Hole, peer ${ep}, err ${err}`) } + updateHoles() return hole } - function createOutboundHole(ep, ent, bound, request) { + function createOutboundHole(ep, natIp, natPort) { updateHoles() + if(findHole(ep)) return + + console.info(`Creating Outbound Hole to ${ep}`) try { var hole = Hole(ep) + hole.updateNatInfo(natIp, natPort) hole.acceptPunch() holes.set(ep, hole) - } catch { - app.log(`Failed to create Inbound Hole, peer ${ep}`) + } catch(err) { + hole = null + app.log(`Failed to create Outbound Hole, peer ${ep}, err ${err}`) } + updateHoles() return hole } + function updateHoleInfo(ep, natIp, natPort) { + var hole = findHole(ep) + if(!hole) throw 'No hole to update' + + hole.updateNatInfo(natIp, natPort) + } + function deleteHole(ep) { var sel = findHole(ep) if(!sel) return @@ -240,6 +303,7 @@ export default function({ app }) { holes, createInboundHole, createOutboundHole, + updateHoleInfo, deleteHole, findHole, randomPort, From 9cda6c18fb4ca3ff39989e44d62b76f7a91b35dd Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:29:23 +0800 Subject: [PATCH 04/13] make hole punch process work --- neptune/mega/tunnel/api.js | 23 ++-- neptune/mega/tunnel/main.js | 20 +++- neptune/mega/tunnel/punch.js | 197 ++++++++++++++++++++++------------- 3 files changed, 158 insertions(+), 82 deletions(-) diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index 3d8bd0d83..bf1a4a6a7 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -189,21 +189,15 @@ export default function ({ app, mesh, punch }) { function updateHoleInfo(ep, ip, port) { checkIP(ip) checkPort(Number.parseInt(port)) - console.info(`Updating nat info: peer ${ep}, ip ${ip}, port ${port}`) punch.updateHoleInfo(ep, ip, port) } - function punchHole(ep) { - console.log(`Hole punching......`) + function syncPunch(ep) { var hole = punch.findHole(ep) - // if(!hole) return null + if(!hole) throw `Invalid Hole State for ${ep}` hole.punch() } - function makeRespTunnel(ep) { - - } - function deleteHole(ep) { punch.deleteHole(ep) } @@ -395,6 +389,17 @@ export default function ({ app, mesh, punch }) { ) ) + var makeRespTunnel = pipeline($=>$ + .onStart(ctx => { + console.info("Making resp tunnel: ", ctx) + var ep = ctx.peer.id + var hole = punch.findHole(ep) + if(!hole) throw `Invalid Hole State for ${ep}` + return hole + }) + .pipe(hole => hole.makeRespTunnel()) + ) + // var makeRespTunnel = pipeline($=>$ // .pipe(punch.makeRespTunnel()) // ) @@ -414,7 +419,7 @@ export default function ({ app, mesh, punch }) { createHole, updateHoleInfo, makeRespTunnel, - punchHole, + syncPunch, deleteHole, servePeerInbound, } diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 9c0c6c8d2..0f7884e5c 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -142,6 +142,10 @@ export default function ({ app, mesh, utils }) { 'CONNECT': api.servePeerInbound, }, + '/api/ping': { + 'GET': responder(() => Promise.resolve(response(200))) + }, + '/api/punch/{action}': { 'GET': responder(({action}) => { var ep = $ctx.peer.id @@ -155,17 +159,27 @@ export default function ({ app, mesh, utils }) { break case 'accept': api.updateHoleInfo(ep, ip, port) - break - case 'sync': api.syncPunch(ep) + break + default: + return Promise.resolve(response(500, "Unknown punch action")) } return Promise.resolve(response(200)) }), - 'CONNECT': api.makeRespTunnel + 'CONNECT': pipeline($=>$.onStart(() => $ctx).pipe(() => api.makeRespTunnel)) }, }) + punch.setService((ctx) => { + // Tricky callback to set ctx, + // expecting everything in hole works + // just like it's coming from hub. + console.info("Setting ctx manually: ", ctx) + $ctx = ctx + return servePeer + }) + return pipeline($=>$ .onStart(c => void ($ctx = c)) .pipe(() => { diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index 929d892e0..1365a7291 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -1,4 +1,4 @@ -export default function({ app, mesh }) { +export default function ({ app, mesh }) { // Only available for symmetric NAT function Hole(ep) { // FIXME: throw on init fail? @@ -20,7 +20,7 @@ export default function({ app, mesh }) { console.info(`Creating hole to peer ${ep}, bound ${bound}`) // Check if ep is self. - if(ep === app.endpoint.id) { + if (ep === app.endpoint.id) { throw 'Must not create a hole to self' } @@ -31,60 +31,82 @@ export default function({ app, mesh }) { if (!role || !destIP || !destPort) throw 'Hole not init correctly' if ($session) return $session + var retryTimes = 0 + + var buildCtx = () => { + return { + source: 'peer', + peer: { + id: ep, + ip: destIP, + port: destPort, + } + } + } + // TODO: support TLS connection if (role === 'client') { // make session to server side directly $session = pipeline($ => $ - .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ - .connect(() => `${destIP}:${destPort}`, { - onState: function (conn) { - if (conn.state === 'open') { - conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) - } else if (conn.state === 'connected') { - app.log(`Connected to remote ${destIP}:${destPort}`) - $connection = conn - state = 'connected' - } else if (conn.state === 'closed') { - app.log(`Disconnected from remote ${destIP}:${destPort}`) - $connection = null - state = 'closed' - } - }, - bind: bound - }) - .handleStreamEnd(evt => { - if (evt.error) { - state = 'fail' - leave() - } + .repeat(() => new Timeout(5).wait().then(() => { + console.info("Retrying...... ", state != 'fail' && retryTimes <= 5) + return state != 'fail' && retryTimes <= 5 + })).to($=>$ + .muxHTTP(() => ep + "direct", /* { version: 2 } */).to($ => $ + .connect(() => `${destIP}:${destPort}`, { + onState: function (conn) { + console.info('Conn Info: ', conn) + if (conn.state === 'open') { + conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) + } else if (conn.state === 'connected') { + app.log(`Connected to remote ${destIP}:${destPort}`) + $connection = conn + state = 'connected' + } else if (conn.state === 'closed') { + app.log(`Disconnected from remote ${destIP}:${destPort}`) + $connection = null + retryTimes += 1 + } + }, + bind: bound + }) + ) + .onEnd(evt => { + console.info('Direct session end: ', evt) }) ) + .onEnd(() => { + app.log(`Direct session to ${ep} failed, hole closing...`) + state = 'fail' + leave() + }) ) // reverse server for receiving requests pipeline($ => $ .onStart(new Data) .repeat(() => new Timeout(5).wait().then(() => { - return state != 'fail' || state != 'closed' + return state != 'fail' && state != 'closed' })).to($ => $ .loop($ => $ .connectHTTPTunnel( new Message({ method: 'CONNECT', - path: `/api/punch/${config.agent.id}/${ep}`, + path: `/api/punch/tunnel`, }) ) .to($session) - .pipe(serveHub) + .pipe(() => svc(buildCtx())) ) ) ).spawn() } else if (role === 'server') { - pipy.listen(bound, 'tcp', serveHub) + console.info("Direct Server Listening...") + pipy.listen(bound, 'tcp', () => svc(buildCtx())) $session = pipeline($ => $ - .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ + .muxHTTP(() => ep + "direct", /* { version: 2 } */).to($ => $ .swap(() => $pHub) ) ) @@ -93,15 +115,15 @@ export default function({ app, mesh }) { return $session } - function fwdRequest(req) { - return pipeline($=>$ + function fwdRequest(req, callback) { + return pipeline($ => $ .onStart(req) - .muxHTTP().to($=>$ + .muxHTTP().to($ => $ .pipe(mesh.connect(ep, { bind: bound, onState: conn => { - if(conn.state === 'open') - conn.socket.setRawOption(1,15,new Data([1,0,0,0])) + if (conn.state === 'open') + conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) } })) ) @@ -112,6 +134,8 @@ export default function({ app, mesh }) { }) .onEnd(() => { console.info('Hub Answers in hole: ', $response) + if(callback) + callback($response) return $response }) ).spawn() @@ -128,11 +152,11 @@ export default function({ app, mesh }) { method: 'GET', path: '/api/punch/request', })) - new Timeout(15).wait().then(connectOrFail) + new Timeout(60).wait().then(connectOrFail) } - // TODO add cert info into response function acceptPunch() { + // TODO add cert info into response // state = 'connecting' role = 'server' @@ -140,26 +164,40 @@ export default function({ app, mesh }) { fwdRequest(new Message({ method: 'GET', path: '/api/punch/accept', - })) - new Timeout(15).wait().then(connectOrFail) + }), (resp) => { + if(resp.head.status != 200) { + app.log(`Failed on accepting`) + state = 'fail' + leave() + updateHoles() + return + } + punch() + }) + new Timeout(60).wait().then(connectOrFail) } function updateNatInfo(ip, port) { + console.info(`Peer NAT Info: ${ip}:${port}`) destIP = ip destPort = port } + // Punch when: + // 1. Server accept got 200 OK + // 2. Client receive accept function punch() { - // receive TLS options - // connectTLS - // connectLocal or connectRemote - + // TODO receive TLS options // TODO add retry logic here + // TODO estimate RTT and use it to make peer synchronized. state = 'punching' - console.info("Punching...") - makeFakeCall(destIP, destPort) - $session = directSession() + console.info(`Punching to ${destIP}:${destPort} (${ep})`) + if (role === 'server') { + makeFakeCall(destIP, destPort) + } + + directSession() heartbeat() // activate the session pipeline } @@ -178,38 +216,47 @@ export default function({ app, mesh }) { function connectOrFail() { console.info(`Failed unpunchable hole: ${ep}`) - if(state != 'connected') state = fail - updateHoles() + if (state != 'connected') { + state = fail + console.info("Made the hole failed") + updateHoles() + } } // send a SYN to dest, expect no return. // this will cheat the firewall to allow inbound connection from dest. function makeFakeCall(destIP, destPort) { console.info("Making fake call") - pipy().task().onStart(new Data).connect(`${destIP}:${destPort}`, { - bind: bound, - onState: function (conn) { - // REUSEPORT - if (conn.state === 'open') conn.socket.setRawOption(1, 15, new Data([1])) - - // abort this connection. - if (conn.state === 'connecting') conn.close() - } - }) + pipeline($=>$ + .onStart(new Data).connect(`${destIP}:${destPort}`, { + bind: bound, + onState: function (conn) { + // REUSEPORT + if (conn.state === 'open') conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) + + // abort this connection. + if (conn.state === 'connecting') { + console.info('Performing early close') + conn.close() + } + } + }) + ).spawn() } function heartbeat() { - fwdRequest.spawn( - new Message( - { method: 'POST', path: '/api/status' }, - JSON.encode({ name: config.agent.name }) - ) - ) + // FIXME use direct session + fwdRequest(new Message({ + method: 'GET', + path: '/api/ping' + })) } function leave() { - $connection?.close() + if($connection) { + $connection?.close() + } $connection = null if (state != 'fail') state = 'closed' } @@ -229,6 +276,7 @@ export default function({ app, mesh }) { } // End of Hole var holes = new Map + var svc = null function updateHoles() { holes.forEach((key, hole) => { @@ -242,13 +290,13 @@ export default function({ app, mesh }) { function createInboundHole(ep, bound, request) { updateHoles() - if(findHole(ep)) return + if (findHole(ep)) return console.info(`Creating Inbound Hole to ${ep}`) try { var hole = Hole(ep, bound, request) hole.requestPunch() holes.set(ep, hole) - } catch(err) { + } catch (err) { hole = null app.log(`Failed to create Inbound Hole, peer ${ep}, err ${err}`) } @@ -259,7 +307,7 @@ export default function({ app, mesh }) { function createOutboundHole(ep, natIp, natPort) { updateHoles() - if(findHole(ep)) return + if (findHole(ep)) return console.info(`Creating Outbound Hole to ${ep}`) try { @@ -267,7 +315,7 @@ export default function({ app, mesh }) { hole.updateNatInfo(natIp, natPort) hole.acceptPunch() holes.set(ep, hole) - } catch(err) { + } catch (err) { hole = null app.log(`Failed to create Outbound Hole, peer ${ep}, err ${err}`) } @@ -278,14 +326,14 @@ export default function({ app, mesh }) { function updateHoleInfo(ep, natIp, natPort) { var hole = findHole(ep) - if(!hole) throw 'No hole to update' + if (!hole) throw 'No hole to update' hole.updateNatInfo(natIp, natPort) } function deleteHole(ep) { var sel = findHole(ep) - if(!sel) return + if (!sel) return sel.leave() holes.delete(ep) } @@ -295,6 +343,14 @@ export default function({ app, mesh }) { return holes.get(ep) } + function getCtx() { + + } + + function setService(srvPeer) { + svc = srvPeer + } + function randomPort() { return Number.parseInt(Math.random() * (65535 - 1024)) + 1024 } @@ -306,6 +362,7 @@ export default function({ app, mesh }) { updateHoleInfo, deleteHole, findHole, + setService, randomPort, } } From 9e661439eec748ee59106cb694838d1dd64f100c Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:30:39 +0800 Subject: [PATCH 05/13] fix client to server data transfer --- neptune/mega/tunnel/api.js | 40 ++++-- neptune/mega/tunnel/main.js | 7 +- neptune/mega/tunnel/punch.js | 264 +++++++++++++++++++---------------- 3 files changed, 176 insertions(+), 135 deletions(-) diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index bf1a4a6a7..de3ae47dd 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -198,8 +198,8 @@ export default function ({ app, mesh, punch }) { hole.punch() } - function deleteHole(ep) { - punch.deleteHole(ep) + function deleteHole(ep, remote) { + punch.deleteHole(ep, remote) } function getLocalConfig() { @@ -246,14 +246,14 @@ export default function ({ app, mesh, punch }) { ).to($=>$.pipe(() => { punch.createInboundHole($selectedEP) var hole = punch.findHole($selectedEP) - if(hole && hole.ready) { - console.info("Using direct session") - return h.directSession() + if(hole && hole.ready()) { + console.info("Using direct session: ", hole) + return hole.directSession() } console.info("Using hub forwarded session") return pipeline($=>$ .muxHTTP().to($=>$ - .pipe(() => mesh.connect($selectedEP)) + .pipe(mesh.connect($selectedEP)) ) ) })) @@ -328,15 +328,22 @@ export default function ({ app, mesh, punch }) { var $response return pipeline($=>$ .onStart(req) - .muxHTTP().to($=>$ - .pipe(mesh.connect(ep)) - ) + .pipe(() => { + var h = punch.findHole(ep) + if(h && h.ready()) { + return h.directSession() + } + return pipeline($=>$ + .muxHTTP().to($=>$ + .pipe(mesh.connect(ep)) + )) + }) .replaceMessage(res => { $response = res return new StreamEnd }) .onEnd(() => { - console.info('Hub answers in api: ', $response) + console.info('Answers in api: ', $response) return $response }) ).spawn() @@ -389,15 +396,20 @@ export default function ({ app, mesh, punch }) { ) ) + var $hole = null var makeRespTunnel = pipeline($=>$ .onStart(ctx => { console.info("Making resp tunnel: ", ctx) var ep = ctx.peer.id - var hole = punch.findHole(ep) - if(!hole) throw `Invalid Hole State for ${ep}` - return hole + $hole = punch.findHole(ep) + if(!$hole) throw `Invalid Hole State for ${ep}` + return new Data + }) + .pipe(() => { + var p = $hole.makeRespTunnel() + $hole = null + return p }) - .pipe(hole => hole.makeRespTunnel()) ) // var makeRespTunnel = pipeline($=>$ diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 0f7884e5c..01575f854 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -1,6 +1,6 @@ +import initHole from './punch.js' import initAPI from './api.js' import initCLI from './cli.js' -import initHole from './punch.js' export default function ({ app, mesh, utils }) { var punch = initHole({ app, mesh }) @@ -161,13 +161,16 @@ export default function ({ app, mesh, utils }) { api.updateHoleInfo(ep, ip, port) api.syncPunch(ep) break + case 'leave': + api.deleteHole(ep, true) + break default: return Promise.resolve(response(500, "Unknown punch action")) } return Promise.resolve(response(200)) }), - 'CONNECT': pipeline($=>$.onStart(() => $ctx).pipe(() => api.makeRespTunnel)) + 'CONNECT': pipeline($=>$.pipe(api.makeRespTunnel, () => $ctx)) }, }) diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index 1365a7291..76a997aff 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -9,12 +9,11 @@ export default function ({ app, mesh }) { var destPort = null var role = null - // closed forwarding connecting(ready punching) connected fail - var state = 'closed' // inner state - var ready = false // exposed state + // idle handshake punching connected closed fail + var state = 'idle' + var session + var pHub = new pipeline.Hub var $connection = null - var $pHub = new pipeline.Hub - var $session var $response console.info(`Creating hole to peer ${ep}, bound ${bound}`) @@ -25,11 +24,8 @@ export default function ({ app, mesh }) { } function directSession() { - // TODO !!! state error would happen when network is slow - // must handle this - if (!role || !destIP || !destPort) throw 'Hole not init correctly' - if ($session) return $session + if (session) return session var retryTimes = 0 @@ -47,94 +43,101 @@ export default function ({ app, mesh }) { // TODO: support TLS connection if (role === 'client') { // make session to server side directly - $session = pipeline($ => $ - .repeat(() => new Timeout(5).wait().then(() => { - console.info("Retrying...... ", state != 'fail' && retryTimes <= 5) - return state != 'fail' && retryTimes <= 5 - })).to($=>$ - .muxHTTP(() => ep + "direct", /* { version: 2 } */).to($ => $ - .connect(() => `${destIP}:${destPort}`, { - onState: function (conn) { - console.info('Conn Info: ', conn) - if (conn.state === 'open') { - conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) - } else if (conn.state === 'connected') { - app.log(`Connected to remote ${destIP}:${destPort}`) - $connection = conn - state = 'connected' - } else if (conn.state === 'closed') { - app.log(`Disconnected from remote ${destIP}:${destPort}`) - $connection = null - retryTimes += 1 - } - }, - bind: bound - }) - ) - .onEnd(evt => { - console.info('Direct session end: ', evt) - }) + var reverseServer = null + session = pipeline($ => $ + .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ + .connect(() => `${destIP}:${destPort}`, { + bind: bound, + onState: function (conn) { + console.info('Conn Info: ', conn) + if (conn.state === 'open') { + conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) + } else if (conn.state === 'connected') { + app.log(`Connected to peer ${destIP}:${destPort}`) + $connection = conn + state = 'connected' + retryTimes = 0 + // reverseServer.spawn() + } else if (conn.state === 'closed') { + app.log(`Disconnected from peer ${destIP}:${destPort}`) + $connection = null + state = 'closed' + retryTimes += 1 + } + + if (retryTimes > 5 || state === 'fail') { + console.info(`Retry limit exceeded, punch failed.`) + state = 'fail' + updateHoles() + } + }, + }).handleStreamEnd(evt => console.info('Connection End, event: ', evt)) ) - .onEnd(() => { - app.log(`Direct session to ${ep} failed, hole closing...`) - state = 'fail' - leave() - }) ) // reverse server for receiving requests - pipeline($ => $ - .onStart(new Data) - .repeat(() => new Timeout(5).wait().then(() => { - return state != 'fail' && state != 'closed' - })).to($ => $ - .loop($ => $ - .connectHTTPTunnel( - new Message({ - method: 'CONNECT', - path: `/api/punch/tunnel`, - }) - ) - .to($session) - .pipe(() => svc(buildCtx())) - ) - ) - ).spawn() - + // reverseServer = pipeline($ => $ + // .onStart(new Data) + // .loop($ => $ + // .connectHTTPTunnel( + // new Message({ + // method: 'CONNECT', + // path: `/api/punch/tunnel`, + // }) + // ) + // .to($session) + // .pipe(() => svc(buildCtx())) + // ) + // ) + + heartbeat() } else if (role === 'server') { console.info("Direct Server Listening...") - pipy.listen(bound, 'tcp', () => svc(buildCtx())) - - $session = pipeline($ => $ + var $msg = null + pipy.listen(bound, 'tcp', $ => $.handleMessage(msg => { + console.info('Server Received: ', msg) + $msg = msg + return new Data + }).pipe(() => svc(buildCtx())), () => $msg) + + session = pipeline($ => $ .muxHTTP(() => ep + "direct", /* { version: 2 } */).to($ => $ - .swap(() => $pHub) + .swap(() => pHub) ) ) } - - return $session + state = 'connected' + return session } - function fwdRequest(req, callback) { + function request(req, callback) { + var store = req return pipeline($ => $ .onStart(req) - .muxHTTP().to($ => $ - .pipe(mesh.connect(ep, { - bind: bound, - onState: conn => { - if (conn.state === 'open') - conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) - } - })) - ) + .pipe(() => { + if (state === 'connected' || state === 'closed' || state === 'punching') { + return session + } + return pipeline($ => $ + .muxHTTP().to($ => $.pipe( + mesh.connect(ep, { + bind: bound, + onState: conn => { + if (conn.state === 'open') + conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) + } + }) + )) + ) + }) .print() .replaceMessage(res => { $response = res return new StreamEnd }) .onEnd(() => { - console.info('Hub Answers in hole: ', $response) - if(callback) + console.info('Answers in hole: ', $response, state === 'connected', store) + if (callback) callback($response) return $response }) @@ -144,11 +147,11 @@ export default function ({ app, mesh }) { // use THE port sending request to hub. function requestPunch() { // FIXME: add state check - // state = 'connecting' role = 'client' + state = 'handshake' console.info("Requesting punch") - fwdRequest(new Message({ + request(new Message({ method: 'GET', path: '/api/punch/request', })) @@ -157,23 +160,21 @@ export default function ({ app, mesh }) { function acceptPunch() { // TODO add cert info into response - // state = 'connecting' role = 'server' + state = 'handshake' console.info("Accepting punch") - fwdRequest(new Message({ + request(new Message({ method: 'GET', path: '/api/punch/accept', }), (resp) => { - if(resp.head.status != 200) { + if (resp.head.status != 200) { app.log(`Failed on accepting`) state = 'fail' - leave() updateHoles() - return } - punch() }) + punch() new Timeout(60).wait().then(connectOrFail) } @@ -184,11 +185,10 @@ export default function ({ app, mesh }) { } // Punch when: - // 1. Server accept got 200 OK + // 1. Server accept message got 200 OK // 2. Client receive accept function punch() { // TODO receive TLS options - // TODO add retry logic here // TODO estimate RTT and use it to make peer synchronized. state = 'punching' @@ -196,9 +196,7 @@ export default function ({ app, mesh }) { if (role === 'server') { makeFakeCall(destIP, destPort) } - directSession() - heartbeat() // activate the session pipeline } function makeRespTunnel() { @@ -206,19 +204,18 @@ export default function ({ app, mesh }) { state = 'connected' return pipeline($ => $ - .acceptHTTPTunnel(() => response200()).to($ => $ + .acceptHTTPTunnel(() => new Message({ status: 200 })).to($ => $ .onStart(new Data) - .swap(() => $pHub) + .swap(() => pHub) .onEnd(() => console.info(`Direct Connection from ${ep} lost`)) ) ) } function connectOrFail() { - console.info(`Failed unpunchable hole: ${ep}`) if (state != 'connected') { - state = fail - console.info("Made the hole failed") + console.info(`Current state ${state}, made the hole failed`) + state = 'fail' updateHoles() } } @@ -227,7 +224,7 @@ export default function ({ app, mesh }) { // this will cheat the firewall to allow inbound connection from dest. function makeFakeCall(destIP, destPort) { console.info("Making fake call") - pipeline($=>$ + pipeline($ => $ .onStart(new Data).connect(`${destIP}:${destPort}`, { bind: bound, onState: function (conn) { @@ -247,23 +244,56 @@ export default function ({ app, mesh }) { function heartbeat() { // FIXME use direct session - fwdRequest(new Message({ - method: 'GET', - path: '/api/ping' - })) + if (state === 'fail') return + + var resp = null + console.info("Sending heartbeat...") + pipeline($ => $ + .onStart(new Message({ + method: 'GET', + path: '/api/ping' + })) + .pipe(session) + .replaceMessage(res => { + resp = res + return new StreamEnd + }) + .onEnd(() => { + console.info('Heartbeat: ', resp) + }) + ).spawn() + // request(new Message({ + // method: 'GET', + // path: '/api/ping' + // }), (resp) => { + // console.info('Heartbeat: ', resp) + // }) + new Timeout(10).wait().then(heartbeat) } - function leave() { - if($connection) { + function leave(remote) { + if (role === 'server') { + pipy.listen(bound, 'tcp', null) + } + + if ($connection) { $connection?.close() } $connection = null if (state != 'fail') state = 'closed' + if (!remote) { + app.log("Hole closed by peer ", ep) + request(new Message({ + method: 'GET', + path: '/api/punch/leave' + })) + } } return { - role, - ready, + role: () => role, + state: () => state, + ready: () => state === 'connected', requestPunch, acceptPunch, updateNatInfo, @@ -280,7 +310,7 @@ export default function ({ app, mesh }) { function updateHoles() { holes.forEach((key, hole) => { - if (hole.state === 'fail' || hole.state === 'closed') { + if (hole.state() === 'fail' || hole.state() === 'closed') { hole.leave() holes.delete(key) } @@ -288,20 +318,20 @@ export default function ({ app, mesh }) { console.info(`Holes after updating: `, holes) } - function createInboundHole(ep, bound, request) { + function createInboundHole(ep) { updateHoles() if (findHole(ep)) return console.info(`Creating Inbound Hole to ${ep}`) try { - var hole = Hole(ep, bound, request) + var hole = Hole(ep) hole.requestPunch() holes.set(ep, hole) } catch (err) { - hole = null + hole.leave() + updateHoles() app.log(`Failed to create Inbound Hole, peer ${ep}, err ${err}`) } - updateHoles() return hole } @@ -316,26 +346,26 @@ export default function ({ app, mesh }) { hole.acceptPunch() holes.set(ep, hole) } catch (err) { - hole = null + hole.leave() + updateHoles() app.log(`Failed to create Outbound Hole, peer ${ep}, err ${err}`) } - updateHoles() return hole } function updateHoleInfo(ep, natIp, natPort) { var hole = findHole(ep) - if (!hole) throw 'No hole to update' + if (!hole) throw `No hole to update, ep ${ep}` hole.updateNatInfo(natIp, natPort) } - function deleteHole(ep) { + function deleteHole(ep, remote) { var sel = findHole(ep) if (!sel) return - sel.leave() - holes.delete(ep) + sel.leave(remote) + updateHoles() } function findHole(ep) { @@ -343,10 +373,6 @@ export default function ({ app, mesh }) { return holes.get(ep) } - function getCtx() { - - } - function setService(srvPeer) { svc = srvPeer } @@ -356,7 +382,7 @@ export default function ({ app, mesh }) { } return { - holes, + getHoles: () => holes, createInboundHole, createOutboundHole, updateHoleInfo, From 802095638e95ce37bd85f36e72206a66d225de9d Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:32:12 +0800 Subject: [PATCH 06/13] fix server to client data transfer --- neptune/mega/tunnel/api.js | 18 ++++----- neptune/mega/tunnel/main.js | 1 - neptune/mega/tunnel/punch.js | 73 +++++++++++++++--------------------- 3 files changed, 38 insertions(+), 54 deletions(-) diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index de3ae47dd..a295654ff 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -247,7 +247,7 @@ export default function ({ app, mesh, punch }) { punch.createInboundHole($selectedEP) var hole = punch.findHole($selectedEP) if(hole && hole.ready()) { - console.info("Using direct session: ", hole) + console.info("Using direct session") return hole.directSession() } console.info("Using hub forwarded session") @@ -396,26 +396,22 @@ export default function ({ app, mesh, punch }) { ) ) - var $hole = null + var tunnelHole = null var makeRespTunnel = pipeline($=>$ .onStart(ctx => { console.info("Making resp tunnel: ", ctx) var ep = ctx.peer.id - $hole = punch.findHole(ep) - if(!$hole) throw `Invalid Hole State for ${ep}` + tunnelHole = punch.findHole(ep) + if(!tunnelHole) throw `Invalid Hole State for ${ep}` return new Data }) .pipe(() => { - var p = $hole.makeRespTunnel() - $hole = null + var p = tunnelHole.makeRespTunnel() + tunnelHole = null return p - }) + }, () => tunnelHole) ) - // var makeRespTunnel = pipeline($=>$ - // .pipe(punch.makeRespTunnel()) - // ) - getLocalConfig().then(applyLocalConfig) return { diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 01575f854..f4476f097 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -178,7 +178,6 @@ export default function ({ app, mesh, utils }) { // Tricky callback to set ctx, // expecting everything in hole works // just like it's coming from hub. - console.info("Setting ctx manually: ", ctx) $ctx = ctx return servePeer }) diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index 76a997aff..a5d77d970 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -31,7 +31,10 @@ export default function ({ app, mesh }) { var buildCtx = () => { return { - source: 'peer', + source: 'direct', + self: { + id: app.endpoint.id, + }, peer: { id: ep, ip: destIP, @@ -57,7 +60,7 @@ export default function ({ app, mesh }) { $connection = conn state = 'connected' retryTimes = 0 - // reverseServer.spawn() + reverseServer.spawn() } else if (conn.state === 'closed') { app.log(`Disconnected from peer ${destIP}:${destPort}`) $connection = null @@ -76,19 +79,19 @@ export default function ({ app, mesh }) { ) // reverse server for receiving requests - // reverseServer = pipeline($ => $ - // .onStart(new Data) - // .loop($ => $ - // .connectHTTPTunnel( - // new Message({ - // method: 'CONNECT', - // path: `/api/punch/tunnel`, - // }) - // ) - // .to($session) - // .pipe(() => svc(buildCtx())) - // ) - // ) + reverseServer = pipeline($ => $ + .onStart(new Data) + .loop($ => $ + .connectHTTPTunnel( + new Message({ + method: 'CONNECT', + path: `/api/punch/tunnel`, + }) + ) + .to(session) + .pipe(() => svc(buildCtx())) + ) + ) heartbeat() } else if (role === 'server') { @@ -101,12 +104,11 @@ export default function ({ app, mesh }) { }).pipe(() => svc(buildCtx())), () => $msg) session = pipeline($ => $ - .muxHTTP(() => ep + "direct", /* { version: 2 } */).to($ => $ + .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ .swap(() => pHub) ) ) } - state = 'connected' return session } @@ -114,22 +116,15 @@ export default function ({ app, mesh }) { var store = req return pipeline($ => $ .onStart(req) - .pipe(() => { - if (state === 'connected' || state === 'closed' || state === 'punching') { - return session - } - return pipeline($ => $ - .muxHTTP().to($ => $.pipe( - mesh.connect(ep, { - bind: bound, - onState: conn => { - if (conn.state === 'open') - conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) - } - }) - )) - ) - }) + .muxHTTP().to($ => $.pipe( + mesh.connect(ep, { + bind: bound, + onState: conn => { + if (conn.state === 'open') + conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) + } + }) + )) .print() .replaceMessage(res => { $response = res @@ -174,6 +169,8 @@ export default function ({ app, mesh }) { updateHoles() } }) + // Just do it, regardless if accept fail. + // Because it's faster. punch() new Timeout(60).wait().then(connectOrFail) } @@ -243,8 +240,8 @@ export default function ({ app, mesh }) { function heartbeat() { - // FIXME use direct session if (state === 'fail') return + if (role === 'server') return var resp = null console.info("Sending heartbeat...") @@ -262,12 +259,6 @@ export default function ({ app, mesh }) { console.info('Heartbeat: ', resp) }) ).spawn() - // request(new Message({ - // method: 'GET', - // path: '/api/ping' - // }), (resp) => { - // console.info('Heartbeat: ', resp) - // }) new Timeout(10).wait().then(heartbeat) } @@ -327,7 +318,6 @@ export default function ({ app, mesh }) { hole.requestPunch() holes.set(ep, hole) } catch (err) { - hole.leave() updateHoles() app.log(`Failed to create Inbound Hole, peer ${ep}, err ${err}`) } @@ -346,7 +336,6 @@ export default function ({ app, mesh }) { hole.acceptPunch() holes.set(ep, hole) } catch (err) { - hole.leave() updateHoles() app.log(`Failed to create Outbound Hole, peer ${ep}, err ${err}`) } From a94ae2fbd34b6676fa496e720cd40b011390ae49 Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:33:56 +0800 Subject: [PATCH 07/13] resp tunnel auto reconnect --- neptune/mega/tunnel/punch.js | 48 +++++++++++++++++++++--------------- 1 file changed, 28 insertions(+), 20 deletions(-) diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index a5d77d970..5256d8c7e 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -9,7 +9,7 @@ export default function ({ app, mesh }) { var destPort = null var role = null - // idle handshake punching connected closed fail + // (idle) (handshake) (punching connected closed) (left fail) var state = 'idle' var session var pHub = new pipeline.Hub @@ -45,8 +45,10 @@ export default function ({ app, mesh }) { // TODO: support TLS connection if (role === 'client') { + var reverseTunnel = null + var reverseTunnelStarted = false + // make session to server side directly - var reverseServer = null session = pipeline($ => $ .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ .connect(() => `${destIP}:${destPort}`, { @@ -60,7 +62,11 @@ export default function ({ app, mesh }) { $connection = conn state = 'connected' retryTimes = 0 - reverseServer.spawn() + + if(!reverseTunnelStarted) { + reverseTunnel.spawn() + reverseTunnelStarted = true + } } else if (conn.state === 'closed') { app.log(`Disconnected from peer ${destIP}:${destPort}`) $connection = null @@ -79,17 +85,22 @@ export default function ({ app, mesh }) { ) // reverse server for receiving requests - reverseServer = pipeline($ => $ + reverseTunnel = pipeline($ => $ .onStart(new Data) - .loop($ => $ - .connectHTTPTunnel( - new Message({ - method: 'CONNECT', - path: `/api/punch/tunnel`, - }) + .repeat(() => new Timeout(1).wait().then(() => { + console.info("Resp Tunnel should start: ", state != 'fail' && state != 'left') + return state != 'fail' && state != 'left' + })).to($ => $ + .loop($ => $ + .connectHTTPTunnel( + new Message({ + method: 'CONNECT', + path: `/api/punch/tunnel`, + }) + ) + .to(session) + .pipe(() => svc(buildCtx())) ) - .to(session) - .pipe(() => svc(buildCtx())) ) ) @@ -131,7 +142,7 @@ export default function ({ app, mesh }) { return new StreamEnd }) .onEnd(() => { - console.info('Answers in hole: ', $response, state === 'connected', store) + console.info('Answers in hole: ', $response, store) if (callback) callback($response) return $response @@ -198,6 +209,7 @@ export default function ({ app, mesh }) { function makeRespTunnel() { // TODO add state check + console.info("Created Resp Tunnel") state = 'connected' return pipeline($ => $ @@ -240,11 +252,10 @@ export default function ({ app, mesh }) { function heartbeat() { - if (state === 'fail') return + if (state === 'fail' || state === 'left') return if (role === 'server') return var resp = null - console.info("Sending heartbeat...") pipeline($ => $ .onStart(new Message({ method: 'GET', @@ -255,9 +266,6 @@ export default function ({ app, mesh }) { resp = res return new StreamEnd }) - .onEnd(() => { - console.info('Heartbeat: ', resp) - }) ).spawn() new Timeout(10).wait().then(heartbeat) } @@ -271,7 +279,7 @@ export default function ({ app, mesh }) { $connection?.close() } $connection = null - if (state != 'fail') state = 'closed' + if (state != 'fail') state = 'left' if (!remote) { app.log("Hole closed by peer ", ep) request(new Message({ @@ -301,7 +309,7 @@ export default function ({ app, mesh }) { function updateHoles() { holes.forEach((key, hole) => { - if (hole.state() === 'fail' || hole.state() === 'closed') { + if (hole.state() === 'fail' || hole.state() === 'left') { hole.leave() holes.delete(key) } From aab8f6806fe656e03d457889bc7169b89d7faa70 Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:35:13 +0800 Subject: [PATCH 08/13] add pacemaker to accelarate directly connect --- neptune/mega/tunnel/punch.js | 87 +++++++++++++++++++++++++++--------- 1 file changed, 65 insertions(+), 22 deletions(-) diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index 5256d8c7e..c2a0272c7 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -1,17 +1,17 @@ export default function ({ app, mesh }) { // Only available for symmetric NAT function Hole(ep) { - // FIXME: throw on init fail? // TODO: Add detailed comment. + // (idle) (handshake) (punching connected closed) (left fail) + var state = 'idle' var bound = '0.0.0.0:' + randomPort() var destIP = null var destPort = null var role = null + var session = null + var rtt = null - // (idle) (handshake) (punching connected closed) (left fail) - var state = 'idle' - var session var pHub = new pipeline.Hub var $connection = null var $response @@ -54,7 +54,6 @@ export default function ({ app, mesh }) { .connect(() => `${destIP}:${destPort}`, { bind: bound, onState: function (conn) { - console.info('Conn Info: ', conn) if (conn.state === 'open') { conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) } else if (conn.state === 'connected') { @@ -74,13 +73,14 @@ export default function ({ app, mesh }) { retryTimes += 1 } - if (retryTimes > 5 || state === 'fail') { + // Max Retry set to 10 + if (retryTimes > 10 || state === 'fail') { console.info(`Retry limit exceeded, punch failed.`) state = 'fail' updateHoles() } }, - }).handleStreamEnd(evt => console.info('Connection End, event: ', evt)) + }).handleStreamEnd(evt => console.info('Hole connection end, retry: ', retryTimes, ' reason: ', evt?.error)) ) ) @@ -88,7 +88,6 @@ export default function ({ app, mesh }) { reverseTunnel = pipeline($ => $ .onStart(new Data) .repeat(() => new Timeout(1).wait().then(() => { - console.info("Resp Tunnel should start: ", state != 'fail' && state != 'left') return state != 'fail' && state != 'left' })).to($ => $ .loop($ => $ @@ -104,7 +103,7 @@ export default function ({ app, mesh }) { ) ) - heartbeat() + pacemaker() } else if (role === 'server') { console.info("Direct Server Listening...") var $msg = null @@ -208,7 +207,6 @@ export default function ({ app, mesh }) { } function makeRespTunnel() { - // TODO add state check console.info("Created Resp Tunnel") state = 'connected' @@ -222,7 +220,11 @@ export default function ({ app, mesh }) { } function connectOrFail() { - if (state != 'connected') { + if (state === 'left') { + // Be quiet when left. + // The hole has been released. + return + } else if (state != 'connected') { console.info(`Current state ${state}, made the hole failed`) state = 'fail' updateHoles() @@ -230,14 +232,14 @@ export default function ({ app, mesh }) { } // send a SYN to dest, expect no return. - // this will cheat the firewall to allow inbound connection from dest. + // this will cheat the firewall to allow inbound connection from peer. function makeFakeCall(destIP, destPort) { console.info("Making fake call") pipeline($ => $ .onStart(new Data).connect(`${destIP}:${destPort}`, { bind: bound, onState: function (conn) { - // REUSEPORT + // Socket Option: REUSEPORT if (conn.state === 'open') conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) // abort this connection. @@ -250,24 +252,67 @@ export default function ({ app, mesh }) { ).spawn() } - - function heartbeat() { + // Send something to server from time to time + // So the firewall and NAT rule should be held. + // + // Params: + // - pacemaker: whether called from pacemaker function + // + function heartbeat(pacemaker) { if (state === 'fail' || state === 'left') return if (role === 'server') return - var resp = null - pipeline($ => $ + var heart = pipeline($ => $ .onStart(new Message({ method: 'GET', path: '/api/ping' })) .pipe(session) .replaceMessage(res => { - resp = res + console.info("Heartbeat OK: ", res.head.status == 200) + if(pacemaker) return res return new StreamEnd }) + ) + + if (pacemaker) + return heart + + // if not called from pacemaker + // the heart should beat automatically :) + console.info('Heartbeating...') + heart.spawn() + new Timeout(10).wait().then(() => heartbeat(false)) + } + + // Used on direct connection setup. + // To urge the connect filter try to call the peer + function pacemaker(rtt) { + if(!rtt) rtt = 0.02 + + var $resp = null + var timeout = [rtt, rtt, rtt, rtt, rtt, 2 * rtt, 3 * rtt, 5 * rtt, 8 *rtt, 13 * rtt] + var round = 0 + var cont = true + + console.info('Pacemaking......') + pipeline($=>$ + .onStart(new Data) + .repeat(() => new Timeout(timeout[round]).wait().then(() => cont && round < 10)) + .to($=>$ + .pipe(() => heartbeat(true)) + .replaceMessage(resp => { + $resp = resp + round += 1 + if (resp.head.status == 200) { + cont = false + heartbeat(false) + } + console.info('Pacemaker: ', resp) + return new StreamEnd + }) + ) ).spawn() - new Timeout(10).wait().then(heartbeat) } function leave(remote) { @@ -281,12 +326,11 @@ export default function ({ app, mesh }) { $connection = null if (state != 'fail') state = 'left' if (!remote) { - app.log("Hole closed by peer ", ep) request(new Message({ method: 'GET', path: '/api/punch/leave' })) - } + } else app.log("Hole closed by peer ", ep) } return { @@ -299,7 +343,6 @@ export default function ({ app, mesh }) { punch, makeRespTunnel, directSession, - heartbeat, leave, } } // End of Hole From f32e7680abead8af947643b4ef89fc9830d0344b Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:36:07 +0800 Subject: [PATCH 09/13] automatically estimate RTT during punching --- neptune/mega/tunnel/main.js | 20 ++++++++++++++++--- neptune/mega/tunnel/punch.js | 37 +++++++++++++++++++++++++++--------- 2 files changed, 45 insertions(+), 12 deletions(-) diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index f4476f097..bd3bd8dda 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -153,6 +153,23 @@ export default function ({ app, mesh, utils }) { var port = $ctx.peer.port console.log(`Punch Event: ${action} from ${ep} ${ip} ${port}`) + switch(action) { + case 'leave': + api.deleteHole(ep, true) + break + default: + return Promise.resolve(response(500, "Unknown punch action")) + } + return Promise.resolve(response(200)) + }), + + 'POST': responder(({action}, req) => { + var obj = JSON.decode(req.body) + var ep = $ctx.peer.id + var ip = $ctx.peer.ip + var port = $ctx.peer.port + + console.log(`Punch Event: ${action} from ${ep} ${ip} ${port}, time ${obj.timestamp}`) switch(action) { case 'request': api.createHole(ep, ip, port, 'server') @@ -161,9 +178,6 @@ export default function ({ app, mesh, utils }) { api.updateHoleInfo(ep, ip, port) api.syncPunch(ep) break - case 'leave': - api.deleteHole(ep, true) - break default: return Promise.resolve(response(500, "Unknown punch action")) } diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index c2a0272c7..e71774d06 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -154,12 +154,25 @@ export default function ({ app, mesh }) { // FIXME: add state check role = 'client' state = 'handshake' + var start = Date.now() console.info("Requesting punch") request(new Message({ - method: 'GET', + method: 'POST', path: '/api/punch/request', - })) + }, JSON.encode({ + timestamp: Date.now() + })), (resp) => { + var end = Date.now() + rtt = (end - start) / 2000 + console.info('Estimated RTT: ', rtt) + + if (resp.head.status != 200) { + app.log(`Failed on requesting`) + state = 'fail' + updateHoles() + } + }) new Timeout(60).wait().then(connectOrFail) } @@ -167,12 +180,19 @@ export default function ({ app, mesh }) { // TODO add cert info into response role = 'server' state = 'handshake' + var start = Date.now() console.info("Accepting punch") request(new Message({ - method: 'GET', + method: 'POST', path: '/api/punch/accept', - }), (resp) => { + }, JSON.encode({ + timestamp: Date.now() + })), (resp) => { + var end = Date.now() + rtt = (end - start) / 2000 + console.info('Estimated RTT: ', rtt) + if (resp.head.status != 200) { app.log(`Failed on accepting`) state = 'fail' @@ -196,7 +216,6 @@ export default function ({ app, mesh }) { // 2. Client receive accept function punch() { // TODO receive TLS options - // TODO estimate RTT and use it to make peer synchronized. state = 'punching' console.info(`Punching to ${destIP}:${destPort} (${ep})`) @@ -269,7 +288,8 @@ export default function ({ app, mesh }) { })) .pipe(session) .replaceMessage(res => { - console.info("Heartbeat OK: ", res.head.status == 200) + if (res.head.status != 200 && !pacemaker) + app.log("Cardiac Arrest happens, hole: ", ep) if(pacemaker) return res return new StreamEnd }) @@ -280,15 +300,14 @@ export default function ({ app, mesh }) { // if not called from pacemaker // the heart should beat automatically :) - console.info('Heartbeating...') heart.spawn() new Timeout(10).wait().then(() => heartbeat(false)) } // Used on direct connection setup. // To urge the connect filter try to call the peer - function pacemaker(rtt) { - if(!rtt) rtt = 0.02 + function pacemaker() { + rtt ??= 0.02 var $resp = null var timeout = [rtt, rtt, rtt, rtt, rtt, 2 * rtt, 3 * rtt, 5 * rtt, 8 *rtt, 13 * rtt] From de2005f3fc4d1345b3c00245425ba06b0334400a Mon Sep 17 00:00:00 2001 From: Neon Date: Sat, 20 Jul 2024 14:36:41 +0800 Subject: [PATCH 10/13] add tls support for direct connection --- neptune/mega/tunnel/api.js | 12 ++- neptune/mega/tunnel/main.js | 11 ++- neptune/mega/tunnel/punch.js | 162 +++++++++++++++++++++++------------ 3 files changed, 121 insertions(+), 64 deletions(-) diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index a295654ff..e077c0263 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -173,23 +173,21 @@ export default function ({ app, mesh, punch }) { } } - function createHole(ep, ip, port, role) { + function createHole(ep, role) { var h = punch.findHole(ep) if(h) return h if(role === 'server') { - checkIP(ip) - checkPort(Number.parseInt(port)) - punch.createOutboundHole(ep, ip, port) + return punch.createOutboundHole(ep) } else if(role === 'client') { - punch.createInboundHole(ep) + return punch.createInboundHole(ep) } } - function updateHoleInfo(ep, ip, port) { + function updateHoleInfo(ep, ip, port, cert) { checkIP(ip) checkPort(Number.parseInt(port)) - punch.updateHoleInfo(ep, ip, port) + punch.updateHoleInfo(ep, ip, port, cert) } function syncPunch(ep) { diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index bd3bd8dda..821b3a016 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -169,13 +169,18 @@ export default function ({ app, mesh, utils }) { var ip = $ctx.peer.ip var port = $ctx.peer.port - console.log(`Punch Event: ${action} from ${ep} ${ip} ${port}, time ${obj.timestamp}`) + console.log(`Punch Event: ${action} from ${ep} ${ip} ${port}`) + console.log("Punch req: ", obj) switch(action) { case 'request': - api.createHole(ep, ip, port, 'server') + api.createHole(ep, 'server') + api.updateHoleInfo(ep, ip, port, obj.cert) + api.syncPunch(ep) + // var certs = hole.signPeerCert(new crypto.PublicKey(obj.pkey)) + // return Promise.resolve(response(200, cert)) break case 'accept': - api.updateHoleInfo(ep, ip, port) + api.updateHoleInfo(ep, ip, port, obj.cert) api.syncPunch(ep) break default: diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index e71774d06..3f6d74687 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -1,8 +1,6 @@ export default function ({ app, mesh }) { // Only available for symmetric NAT function Hole(ep) { - // TODO: Add detailed comment. - // (idle) (handshake) (punching connected closed) (left fail) var state = 'idle' var bound = '0.0.0.0:' + randomPort() @@ -12,13 +10,18 @@ export default function ({ app, mesh }) { var session = null var rtt = null + var tlsOptions = { + certificate: null, + trusted: null + } + var pHub = new pipeline.Hub var $connection = null var $response - console.info(`Creating hole to peer ${ep}, bound ${bound}`) // Check if ep is self. + console.info(`Creating hole to peer ${ep}, bound ${bound}`) if (ep === app.endpoint.id) { throw 'Must not create a hole to self' } @@ -28,6 +31,7 @@ export default function ({ app, mesh }) { if (session) return session var retryTimes = 0 + var tlsState = null var buildCtx = () => { return { @@ -43,7 +47,6 @@ export default function ({ app, mesh }) { } } - // TODO: support TLS connection if (role === 'client') { var reverseTunnel = null var reverseTunnelStarted = false @@ -51,36 +54,48 @@ export default function ({ app, mesh }) { // make session to server side directly session = pipeline($ => $ .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ - .connect(() => `${destIP}:${destPort}`, { - bind: bound, - onState: function (conn) { - if (conn.state === 'open') { - conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) - } else if (conn.state === 'connected') { - app.log(`Connected to peer ${destIP}:${destPort}`) - $connection = conn + .connectTLS({ + ...tlsOptions, + onState: tls => { + console.info('TLS State: ', tlsState === 'connected') + if($connection.state === 'connected' && tls.state === 'connected') { + app.log(`Connected TLS to peer ${destIP}:${destPort}`) state = 'connected' retryTimes = 0 - if(!reverseTunnelStarted) { + if (!reverseTunnelStarted) { reverseTunnel.spawn() reverseTunnelStarted = true } - } else if (conn.state === 'closed') { - app.log(`Disconnected from peer ${destIP}:${destPort}`) - $connection = null - state = 'closed' - retryTimes += 1 } + } + }).to($ => $ + .connect(() => `${destIP}:${destPort}`, { + bind: bound, + onState: function (conn) { + console.info("Conn Info: ", conn, tlsState) + + if (conn.state === 'open') { + conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) + } else if (conn.state === 'connected') { + app.log(`Connected to peer ${destIP}:${destPort}`) + $connection = conn + } else if (conn.state === 'closed') { + app.log(`Disconnected from peer ${destIP}:${destPort}`) + $connection = null + state = 'closed' + retryTimes += 1 + } - // Max Retry set to 10 - if (retryTimes > 10 || state === 'fail') { - console.info(`Retry limit exceeded, punch failed.`) - state = 'fail' - updateHoles() - } - }, - }).handleStreamEnd(evt => console.info('Hole connection end, retry: ', retryTimes, ' reason: ', evt?.error)) + // Max Retry set to 10 + if (retryTimes > 10 || state === 'fail') { + console.info(`Retry limit exceeded, punch failed.`) + state = 'fail' + updateHoles() + } + }, + }).handleStreamEnd(evt => console.info('Hole connection end, retry: ', retryTimes + 1, ' reason: ', evt?.error)) + ) ) ) @@ -103,15 +118,27 @@ export default function ({ app, mesh }) { ) ) + // Forced Heartbeats + // Do a PCR to the hole. pacemaker() + } else if (role === 'server') { - console.info("Direct Server Listening...") var $msg = null - pipy.listen(bound, 'tcp', $ => $.handleMessage(msg => { - console.info('Server Received: ', msg) - $msg = msg - return new Data - }).pipe(() => svc(buildCtx())), () => $msg) + var listen = pipeline($ => $ + .acceptTLS({ + ...tlsOptions, + onState: tls => console.info('TLS State: ', tls) + }).to($ => $ + .handleMessage(msg => { + console.info('Server Received: ', msg) + $msg = msg + return new Data + }).pipe(() => svc(buildCtx())), () => $msg + ) + ) + + console.info("Direct Server Listening...") + pipy.listen(bound, 'tcp', listen) session = pipeline($ => $ .muxHTTP(() => ep + "direct", { version: 2 }).to($ => $ @@ -151,7 +178,6 @@ export default function ({ app, mesh }) { // use THE port sending request to hub. function requestPunch() { - // FIXME: add state check role = 'client' state = 'handshake' var start = Date.now() @@ -161,7 +187,8 @@ export default function ({ app, mesh }) { method: 'POST', path: '/api/punch/request', }, JSON.encode({ - timestamp: Date.now() + timestamp: Date.now(), + cert: genCert() })), (resp) => { var end = Date.now() rtt = (end - start) / 2000 @@ -177,7 +204,6 @@ export default function ({ app, mesh }) { } function acceptPunch() { - // TODO add cert info into response role = 'server' state = 'handshake' var start = Date.now() @@ -187,24 +213,52 @@ export default function ({ app, mesh }) { method: 'POST', path: '/api/punch/accept', }, JSON.encode({ - timestamp: Date.now() + timestamp: Date.now(), + cert: genCert() })), (resp) => { var end = Date.now() rtt = (end - start) / 2000 console.info('Estimated RTT: ', rtt) - if (resp.head.status != 200) { + if (!resp || resp.head.status != 200) { app.log(`Failed on accepting`) state = 'fail' updateHoles() } }) - // Just do it, regardless if accept fail. - // Because it's faster. - punch() + new Timeout(60).wait().then(connectOrFail) } + // Locally generate certificate. + // The handshake process and hub + // will ensure peer is trustworthy + function genCert() { + var key = new crypto.PrivateKey({ type: 'rsa', bits: 2048 }) + var pKey = new crypto.PublicKey(key) + var cert = new crypto.Certificate({ + subject: { CN: role }, + publicKey: pKey, + privateKey: key, + days: 365, + }) + + tlsOptions = { + certificate: { + cert: cert, + key: key, + } + } + + return cert.toPEM().toString() + } + + function addPeerCert(cert) { + var peerCert = new crypto.Certificate(cert) + console.info("TLS: ", tlsOptions) + tlsOptions['trusted'] = [peerCert] + } + function updateNatInfo(ip, port) { console.info(`Peer NAT Info: ${ip}:${port}`) destIP = ip @@ -215,7 +269,6 @@ export default function ({ app, mesh }) { // 1. Server accept message got 200 OK // 2. Client receive accept function punch() { - // TODO receive TLS options state = 'punching' console.info(`Punching to ${destIP}:${destPort} (${ep})`) @@ -290,7 +343,7 @@ export default function ({ app, mesh }) { .replaceMessage(res => { if (res.head.status != 200 && !pacemaker) app.log("Cardiac Arrest happens, hole: ", ep) - if(pacemaker) return res + if (pacemaker) return res return new StreamEnd }) ) @@ -309,19 +362,21 @@ export default function ({ app, mesh }) { function pacemaker() { rtt ??= 0.02 - var $resp = null - var timeout = [rtt, rtt, rtt, rtt, rtt, 2 * rtt, 3 * rtt, 5 * rtt, 8 *rtt, 13 * rtt] + var timeout = [rtt, rtt, rtt, rtt, rtt, 2 * rtt, 3 * rtt, 5 * rtt, 8 * rtt, 13 * rtt] var round = 0 var cont = true console.info('Pacemaking......') - pipeline($=>$ + pipeline($ => $ .onStart(new Data) - .repeat(() => new Timeout(timeout[round]).wait().then(() => cont && round < 10)) - .to($=>$ + .repeat(() => { + if(round < 10) + return new Timeout(timeout[round]).wait().then(() => cont) + return false + }) + .to($ => $ .pipe(() => heartbeat(true)) .replaceMessage(resp => { - $resp = resp round += 1 if (resp.head.status == 200) { cont = false @@ -359,6 +414,7 @@ export default function ({ app, mesh }) { requestPunch, acceptPunch, updateNatInfo, + addPeerCert, punch, makeRespTunnel, directSession, @@ -389,7 +445,7 @@ export default function ({ app, mesh }) { holes.set(ep, hole) } catch (err) { updateHoles() - app.log(`Failed to create Inbound Hole, peer ${ep}, err ${err}`) + app.log('Failed to create Inbound Hole, Error: ', err) } return hole @@ -398,26 +454,25 @@ export default function ({ app, mesh }) { function createOutboundHole(ep, natIp, natPort) { updateHoles() if (findHole(ep)) return - console.info(`Creating Outbound Hole to ${ep}`) try { var hole = Hole(ep) - hole.updateNatInfo(natIp, natPort) hole.acceptPunch() holes.set(ep, hole) } catch (err) { updateHoles() - app.log(`Failed to create Outbound Hole, peer ${ep}, err ${err}`) + app.log('Failed to create Outbound Hole, Error: ', err) } return hole } - function updateHoleInfo(ep, natIp, natPort) { + function updateHoleInfo(ep, natIp, natPort, cert) { var hole = findHole(ep) if (!hole) throw `No hole to update, ep ${ep}` hole.updateNatInfo(natIp, natPort) + hole.addPeerCert(cert) } function deleteHole(ep, remote) { @@ -428,7 +483,6 @@ export default function ({ app, mesh }) { } function findHole(ep) { - updateHoles() return holes.get(ep) } From bac52255cfa0ebce2cc84159b33edc0d4a660305 Mon Sep 17 00:00:00 2001 From: Neon Date: Sun, 21 Jul 2024 10:25:00 +0800 Subject: [PATCH 11/13] add reuse-port option to pipy --- neptune/src/lib.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/neptune/src/lib.rs b/neptune/src/lib.rs index 18b1a2554..29b62a3e3 100644 --- a/neptune/src/lib.rs +++ b/neptune/src/lib.rs @@ -18,6 +18,7 @@ pub fn start_agent(database: &str, listen_port: u16) { let args = [ CString::new("ztm-pipy").unwrap(), CString::new("repo://ztm/agent").unwrap(), + CString::new("--reuse-port").unwrap(), CString::new("--args").unwrap(), CString::new(format!("--database={}", database)).unwrap(), CString::new(format!("--listen=0.0.0.0:{}", listen_port)).unwrap(), From 3e8304afdf0774329edc5853b15953000623b36c Mon Sep 17 00:00:00 2001 From: Neon Date: Sun, 21 Jul 2024 15:36:18 +0800 Subject: [PATCH 12/13] add fake CA certificate --- neptune/mega/tunnel/punch.js | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index 3f6d74687..3d312be1b 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -73,7 +73,7 @@ export default function ({ app, mesh }) { .connect(() => `${destIP}:${destPort}`, { bind: bound, onState: function (conn) { - console.info("Conn Info: ", conn, tlsState) + console.info("Conn Info: ", conn) if (conn.state === 'open') { conn.socket.setRawOption(1, 15, new Data([1, 0, 0, 0])) @@ -243,6 +243,14 @@ export default function ({ app, mesh }) { days: 365, }) + cert = new crypto.Certificate({ + subject: { CN: role }, + publicKey: pKey, + privateKey: key, + issuer: cert, + days: 365, + }) + tlsOptions = { certificate: { cert: cert, From e5772dca3bad4ae392fb3e09fc1411b8be4e16c3 Mon Sep 17 00:00:00 2001 From: Neon Date: Sun, 21 Jul 2024 15:36:51 +0800 Subject: [PATCH 13/13] fail count for holes --- neptune/mega/tunnel/punch.js | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel/punch.js index 3d312be1b..029a0b247 100644 --- a/neptune/mega/tunnel/punch.js +++ b/neptune/mega/tunnel/punch.js @@ -31,7 +31,6 @@ export default function ({ app, mesh }) { if (session) return session var retryTimes = 0 - var tlsState = null var buildCtx = () => { return { @@ -57,7 +56,7 @@ export default function ({ app, mesh }) { .connectTLS({ ...tlsOptions, onState: tls => { - console.info('TLS State: ', tlsState === 'connected') + console.info('TLS State: ', tls) if($connection.state === 'connected' && tls.state === 'connected') { app.log(`Connected TLS to peer ${destIP}:${destPort}`) state = 'connected' @@ -94,7 +93,8 @@ export default function ({ app, mesh }) { updateHoles() } }, - }).handleStreamEnd(evt => console.info('Hole connection end, retry: ', retryTimes + 1, ' reason: ', evt?.error)) + }) + .handleStreamEnd(evt => console.info('Hole connection end, retry: ', retryTimes + 1, ' reason: ', evt?.error)) ) ) ) @@ -237,7 +237,7 @@ export default function ({ app, mesh }) { var key = new crypto.PrivateKey({ type: 'rsa', bits: 2048 }) var pKey = new crypto.PublicKey(key) var cert = new crypto.Certificate({ - subject: { CN: role }, + subject: { CN: 'Fake CA' }, publicKey: pKey, privateKey: key, days: 365, @@ -431,13 +431,16 @@ export default function ({ app, mesh }) { } // End of Hole var holes = new Map + var fails = {} var svc = null function updateHoles() { holes.forEach((key, hole) => { + fails[key] ??= 0 if (hole.state() === 'fail' || hole.state() === 'left') { hole.leave() holes.delete(key) + fails[key] += 1 } }) console.info(`Holes after updating: `, holes) @@ -446,6 +449,10 @@ export default function ({ app, mesh }) { function createInboundHole(ep) { updateHoles() if (findHole(ep)) return + if (fails[ep] && fails[ep] >= 3) { + console.info(`Won't create hole to ${ep}, too many fails!`) + return + } console.info(`Creating Inbound Hole to ${ep}`) try { var hole = Hole(ep)