From 079bbc2060678541ee98b267c13e4e89697242f6 Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Thu, 22 Aug 2024 15:41:42 +0800 Subject: [PATCH 1/2] update relay node online check --- aries/src/service/relay_server.rs | 3 +++ jupiter/src/storage/ztm_storage.rs | 9 ++++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/aries/src/service/relay_server.rs b/aries/src/service/relay_server.rs index b92356065..59f4dac92 100644 --- a/aries/src/service/relay_server.rs +++ b/aries/src/service/relay_server.rs @@ -252,6 +252,9 @@ async fn check_nodes_online(context: Context) { let nodelist: Vec = storage.get_all_node().await.unwrap().into_iter().collect(); for mut node in nodelist { + if !node.online { + continue; + } //check online let from_timestamp = Duration::from_millis(node.last_online_time as u64); let now = SystemTime::now(); diff --git a/jupiter/src/storage/ztm_storage.rs b/jupiter/src/storage/ztm_storage.rs index ed73eb1e6..a1abe7f7e 100644 --- a/jupiter/src/storage/ztm_storage.rs +++ b/jupiter/src/storage/ztm_storage.rs @@ -54,7 +54,14 @@ impl ZTMStorage { } pub async fn update_node(&self, node: ztm_node::Model) -> Result { - Ok(ztm_node::Entity::update(node.into_active_model()) + let mut active_model: ztm_node::ActiveModel = node.clone().into_active_model(); + active_model.hub = Set(node.hub); + active_model.agent_name = Set(node.agent_name); + active_model.service_name = Set(node.service_name); + active_model.r#type = Set(node.r#type); + active_model.online = Set(node.online); + active_model.last_online_time = Set(node.last_online_time); + Ok(ztm_node::Entity::update(active_model) .exec(self.get_connection()) .await .unwrap()) From 18bae8b2afc5de304fd4ffc90c510ce392bcd764 Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Fri, 23 Aug 2024 15:17:26 +0800 Subject: [PATCH 2/2] Temporarily disable ztm zat punch --- neptune/mega/tunnel/api.js | 98 +--- neptune/mega/tunnel/main.js | 73 +-- neptune/mega/tunnel_punch/api.js | 511 ++++++++++++++++++ neptune/mega/tunnel_punch/cli.js | 303 +++++++++++ neptune/mega/tunnel_punch/main.js | 222 ++++++++ .../mega/{tunnel => tunnel_punch}/punch.js | 0 6 files changed, 1050 insertions(+), 157 deletions(-) create mode 100644 neptune/mega/tunnel_punch/api.js create mode 100644 neptune/mega/tunnel_punch/cli.js create mode 100644 neptune/mega/tunnel_punch/main.js rename neptune/mega/{tunnel => tunnel_punch}/punch.js (100%) diff --git a/neptune/mega/tunnel/api.js b/neptune/mega/tunnel/api.js index e077c0263..46bc584e9 100644 --- a/neptune/mega/tunnel/api.js +++ b/neptune/mega/tunnel/api.js @@ -1,4 +1,4 @@ -export default function ({ app, mesh, punch }) { +export default function ({ app, mesh }) { var currentListens = [] var currentTargets = {} @@ -39,12 +39,9 @@ export default function ({ app, mesh, punch }) { function getInbound(ep, protocol, name) { if (ep === app.endpoint.id) { return getLocalConfig().then( - config => { - var inbound = config.inbound.find( - i => i.protocol === protocol && i.name === name - ) || null - var hole = punch.findHole(ep) - } + config => config.inbound.find( + i => i.protocol === protocol && i.name === name + ) || null ) } else { return requestPeer(ep, new Message( @@ -173,33 +170,6 @@ export default function ({ app, mesh, punch }) { } } - function createHole(ep, role) { - var h = punch.findHole(ep) - if(h) return h - - if(role === 'server') { - return punch.createOutboundHole(ep) - } else if(role === 'client') { - return punch.createInboundHole(ep) - } - } - - function updateHoleInfo(ep, ip, port, cert) { - checkIP(ip) - checkPort(Number.parseInt(port)) - punch.updateHoleInfo(ep, ip, port, cert) - } - - function syncPunch(ep) { - var hole = punch.findHole(ep) - if(!hole) throw `Invalid Hole State for ${ep}` - hole.punch() - } - - function deleteHole(ep, remote) { - punch.deleteHole(ep, remote) - } - function getLocalConfig() { return mesh.read('/local/config.json').then( data => data ? JSON.decode(data) : { inbound: [], outbound: [] } @@ -227,34 +197,23 @@ 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 var listens = i.listens - var $selectedEP + var $selectedEP var connectPeer = pipeline($=>$ .connectHTTPTunnel( new Message({ method: 'CONNECT', path: `/api/outbound/${protocol}/${name}`, }) - ).to($=>$.pipe(() => { - punch.createInboundHole($selectedEP) - var hole = punch.findHole($selectedEP) - if(hole && hole.ready()) { - console.info("Using direct session") - return hole.directSession() - } - console.info("Using hub forwarded session") - return pipeline($=>$ - .muxHTTP().to($=>$ - .pipe(mesh.connect($selectedEP)) - ) + ).to($=>$ + .muxHTTP().to($=>$ + .pipe(() => mesh.connect($selectedEP)) ) - })) + ) .onEnd(() => app.log(`Disconnected from ep ${$selectedEP} for ${protocol}/${name}`)) ) @@ -326,24 +285,14 @@ export default function ({ app, mesh, punch }) { var $response return pipeline($=>$ .onStart(req) - .pipe(() => { - var h = punch.findHole(ep) - if(h && h.ready()) { - return h.directSession() - } - return pipeline($=>$ - .muxHTTP().to($=>$ - .pipe(mesh.connect(ep)) - )) - }) + .muxHTTP().to($=>$ + .pipe(mesh.connect(ep)) + ) .replaceMessage(res => { $response = res return new StreamEnd }) - .onEnd(() => { - console.info('Answers in api: ', $response) - return $response - }) + .onEnd(() => $response) ).spawn() } @@ -394,22 +343,6 @@ export default function ({ app, mesh, punch }) { ) ) - var tunnelHole = null - var makeRespTunnel = pipeline($=>$ - .onStart(ctx => { - console.info("Making resp tunnel: ", ctx) - var ep = ctx.peer.id - tunnelHole = punch.findHole(ep) - if(!tunnelHole) throw `Invalid Hole State for ${ep}` - return new Data - }) - .pipe(() => { - var p = tunnelHole.makeRespTunnel() - tunnelHole = null - return p - }, () => tunnelHole) - ) - getLocalConfig().then(applyLocalConfig) return { @@ -422,11 +355,6 @@ export default function ({ app, mesh, punch }) { setOutbound, deleteInbound, deleteOutbound, - createHole, - updateHoleInfo, - makeRespTunnel, - syncPunch, - deleteHole, servePeerInbound, } } diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 4089ea68a..c899ed385 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -1,10 +1,8 @@ -import initHole from './punch.js' import initAPI from './api.js' import initCLI from './cli.js' export default function ({ app, mesh, utils }) { - var punch = initHole({ app, mesh }) - var api = initAPI({ app, mesh, punch }) + var api = initAPI({ app, mesh }) var cli = initCLI({ app, mesh, utils, api }) var $ctx @@ -84,17 +82,6 @@ export default function ({ app, mesh, utils }) { }), }, - // '/api/punch/{destEp}': { - // 'GET': responder(({destEp}) => { - // return api.createHole(destEp) - // }), - - // 'DELETE': responder(({destEp}) => { - // api.deleteHole(destEp) - // return response(204) - // }) - // }, - '*': { 'GET': responder((_, req) => { return Promise.resolve(gui.serve(req) || response(404)) @@ -150,64 +137,6 @@ 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 - var ip = $ctx.peer.ip - 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}`) - console.log("Punch req: ", obj) - switch(action) { - case 'request': - 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, obj.cert) - api.syncPunch(ep) - break - default: - return Promise.resolve(response(500, "Unknown punch action")) - } - return Promise.resolve(response(200)) - }), - - 'CONNECT': pipeline($=>$.pipe(api.makeRespTunnel, () => $ctx)) - }, - }) - - punch.setService((ctx) => { - // Tricky callback to set ctx, - // expecting everything in hole works - // just like it's coming from hub. - $ctx = ctx - return servePeer }) return pipeline($=>$ diff --git a/neptune/mega/tunnel_punch/api.js b/neptune/mega/tunnel_punch/api.js new file mode 100644 index 000000000..e077c0263 --- /dev/null +++ b/neptune/mega/tunnel_punch/api.js @@ -0,0 +1,511 @@ +export default function ({ app, mesh, punch }) { + var currentListens = [] + var currentTargets = {} + + function allEndpoints() { + return mesh.discover() + } + + function allInbound(ep) { + if (ep === app.endpoint.id) { + return getLocalConfig().then( + config => config.inbound + ) + } else { + return requestPeer(ep, new Message( + { + method: 'GET', + path: `/api/inbound`, + } + )).then(res => res ? JSON.decode(res.body) : null) + } + } + + function allOutbound(ep) { + if (ep === app.endpoint.id) { + return getLocalConfig().then( + config => config.outbound + ) + } else { + return requestPeer(ep, new Message( + { + method: 'GET', + path: `/api/outbound`, + } + )).then(res => res ? JSON.decode(res.body) : null) + } + } + + function getInbound(ep, protocol, name) { + if (ep === app.endpoint.id) { + return getLocalConfig().then( + 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( + { + method: 'GET', + path: `/api/inbound/${protocol}/${name}`, + } + )).then(res => res?.head?.status === 200 ? JSON.decode(res.body) : null) + } + } + + function getOutbound(ep, protocol, name) { + if (ep === app.endpoint.id) { + return getLocalConfig().then( + config => config.outbound.find( + o => o.protocol === protocol && o.name === name + ) || null + ) + } else { + return requestPeer(ep, new Message( + { + method: 'GET', + path: `/api/outbound/${protocol}/${name}`, + } + )).then(res => res?.head?.status === 200 ? JSON.decode(res.body) : null) + } + } + + function setInbound(ep, protocol, name, listens, exits) { + if (ep === app.endpoint.id) { + exits = exits || [] + checkProtocol(protocol) + checkName(name) + checkListens(listens) + checkExits(exits) + return getLocalConfig().then(config => { + var all = config.inbound + var ent = { protocol, name, listens, exits } + var i = all.findIndex(i => i.protocol === protocol && i.name === name) + if (i >= 0) { + all[i] = ent + } else { + all.push(ent) + } + setLocalConfig(config) + applyLocalConfig(config) + }) + } else { + return requestPeer(ep, new Message( + { + method: 'POST', + path: `/api/inbound/${protocol}/${name}`, + }, + JSON.encode({ listens, exits }) + )) + } + } + + function setOutbound(ep, protocol, name, targets, entrances) { + if (ep === app.endpoint.id) { + entrances = entrances || [] + checkProtocol(protocol) + checkName(name) + checkTargets(targets) + checkEntrances(entrances) + return getLocalConfig().then(config => { + var all = config.outbound + var ent = { protocol, name, targets, entrances } + var i = all.findIndex(o => o.protocol === protocol && o.name === name) + if (i >= 0) { + all[i] = ent + } else { + all.push(ent) + } + setLocalConfig(config) + applyLocalConfig(config) + }) + } else { + return requestPeer(ep, new Message( + { + method: 'POST', + path: `/api/outbound/${protocol}/${name}`, + }, + JSON.encode({ targets, entrances }) + )) + } + } + + function deleteInbound(ep, protocol, name) { + if (ep === app.endpoint.id) { + return getLocalConfig().then(config => { + var all = config.inbound + var i = all.findIndex(i => i.protocol === protocol && i.name === name) + if (i >= 0) { + all.splice(i, 1) + setLocalConfig(config) + } + }) + } else { + return requestPeer(ep, new Message( + { + method: 'DELETE', + path: `/api/inbound/${protocol}/${name}`, + } + )) + } + } + + function deleteOutbound(ep, protocol, name) { + if (ep === app.endpoint.id) { + return getLocalConfig().then(config => { + var all = config.outbound + var i = all.findIndex(o => o.protocol === protocol && o.name === name) + if (i >= 0) { + all.splice(i, 1) + setLocalConfig(config) + } + }) + } else { + return requestPeer(ep, new Message( + { + method: 'DELETE', + path: `/api/outbound/${protocol}/${name}`, + } + )) + } + } + + function createHole(ep, role) { + var h = punch.findHole(ep) + if(h) return h + + if(role === 'server') { + return punch.createOutboundHole(ep) + } else if(role === 'client') { + return punch.createInboundHole(ep) + } + } + + function updateHoleInfo(ep, ip, port, cert) { + checkIP(ip) + checkPort(Number.parseInt(port)) + punch.updateHoleInfo(ep, ip, port, cert) + } + + function syncPunch(ep) { + var hole = punch.findHole(ep) + if(!hole) throw `Invalid Hole State for ${ep}` + hole.punch() + } + + function deleteHole(ep, remote) { + punch.deleteHole(ep, remote) + } + + function getLocalConfig() { + return mesh.read('/local/config.json').then( + data => data ? JSON.decode(data) : { inbound: [], outbound: [] } + ) + } + + function setLocalConfig(config) { + mesh.write('/local/config.json', JSON.encode(config)) + } + + function applyLocalConfig(config) { + currentListens.forEach(l => { + var protocol = l.protocol + var ip = l.ip + var port = l.port + if (!config.inbound.some(i => ( + i.protocol === protocol && + i.listens.some(l => l.ip === ip && l.port === port) + ))) { + pipy.listen(`${ip}:${port}`, protocol, null) + app.log(`Stopped ${protocol} listening ${ip}:${port}`) + } + }) + + currentListens = [] + currentTargets = {} + + console.info(`Applying config: ${JSON.encode(config)}`) + + config.inbound.forEach(i => { + var protocol = i.protocol + var name = i.name + var listens = i.listens + var $selectedEP + + var connectPeer = pipeline($=>$ + .connectHTTPTunnel( + new Message({ + method: 'CONNECT', + path: `/api/outbound/${protocol}/${name}`, + }) + ).to($=>$.pipe(() => { + punch.createInboundHole($selectedEP) + var hole = punch.findHole($selectedEP) + if(hole && hole.ready()) { + console.info("Using direct session") + return hole.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}`)) + ) + + var pass = null + var deny = pipeline($=>$.replaceStreamStart(new StreamEnd)) + + switch (protocol) { + case 'tcp': + pass = connectPeer + break + case 'udp': + pass = pipeline($=>$ + .replaceData(data => new Message(data)) + .encodeWebSocket() + .pipe(connectPeer) + .decodeWebSocket() + .replaceMessage(msg => msg.body) + ) + break + } + + var p = pipeline($=>$ + .onStart(() => + ((i.exits && i.exits.length > 0) + ? Promise.resolve(i.exits) + : mesh.discover().then(list => list.map(ep => ep.id)) + ).then(exits => Promise.all( + exits.map( + id => getOutbound(id, protocol, name).then( + o => o ? { ep: id, ...o } : null + ) + ) + )).then(list => { + list = list.filter(o => (o && ( + !o.entrances || + o.entrances.length === 0 || + o.entrances.includes(app.endpoint.id) + ))) + if (list.length > 0) { + $selectedEP = list[Math.floor(Math.random() * list.length)].ep + app.log(`Connect to ep ${$selectedEP} for ${protocol}/${name}`) + } else { + app.log(`No exit found for ${protocol}/${name}`) + } + return new Data + }) + ) + .pipe(() => $selectedEP ? pass : deny) + ) + + listens.forEach(l => { + try { + pipy.listen(`${l.ip}:${l.port}`, protocol, p) + currentListens.push({ protocol, ip: l.ip, port: l.port }) + app.log(`Started ${protocol} listening ${l.ip}:${l.port}`) + } catch (err) { + app.log(`Cannot open port ${l.ip}:${l.port}: ${err}`) + } + }) + }) + + config.outbound.forEach(o => { + var key = `${o.protocol}/${o.name}` + currentTargets[key] = new algo.LoadBalancer(o.targets) + }) + } + + function requestPeer(ep, req) { + var $response + return pipeline($=>$ + .onStart(req) + .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('Answers in api: ', $response) + return $response + }) + ).spawn() + } + + var matchApiOutbound = new http.Match('/api/outbound/{proto}/{name}') + var response200 = new Message({ status: 200 }) + var response404 = new Message({ status: 404 }) + + var $resource + var $target + var $protocol + + var servePeerInbound = pipeline($=>$ + .acceptHTTPTunnel(req => { + var params = matchApiOutbound(req.head.path) + var proto = params?.proto + var name = params?.name + var key = `${proto}/${name}` + var lb = currentTargets[key] + if (lb) { + $resource = lb.allocate() + var target = $resource.target + var host = target.host + var port = target.port + $target = `${host}:${port}` + $protocol = proto + app.log(`Connect to ${$target} for ${key}`) + return response200 + } + app.log(`No target found for ${key}`) + return response404 + }).to($=>$ + .pipe(() => $protocol, { + 'tcp': ($=>$ + .connect(() => $target) + ), + 'udp': ($=>$ + .decodeWebSocket() + .replaceMessage(msg => msg.body) + .connect(() => $target, { protocol: 'udp' }) + .replaceData(data => new Message(data)) + .encodeWebSocket() + ) + }) + .onEnd(() => { + $resource.free() + app.log(`Disconnected from ${$target}`) + }) + ) + ) + + var tunnelHole = null + var makeRespTunnel = pipeline($=>$ + .onStart(ctx => { + console.info("Making resp tunnel: ", ctx) + var ep = ctx.peer.id + tunnelHole = punch.findHole(ep) + if(!tunnelHole) throw `Invalid Hole State for ${ep}` + return new Data + }) + .pipe(() => { + var p = tunnelHole.makeRespTunnel() + tunnelHole = null + return p + }, () => tunnelHole) + ) + + getLocalConfig().then(applyLocalConfig) + + return { + allEndpoints, + allInbound, + allOutbound, + getInbound, + getOutbound, + setInbound, + setOutbound, + deleteInbound, + deleteOutbound, + createHole, + updateHoleInfo, + makeRespTunnel, + syncPunch, + deleteHole, + servePeerInbound, + } +} + +function checkProtocol(protocol) { + switch (protocol) { + case 'tcp': + case 'udp': + return + default: throw `invalid protocol '${protocol}'` + } +} + +function checkName(name) { + if ( + typeof name !== 'string' || + name.indexOf('/') >= 0 + ) throw `invalid name '${name}'` +} + +function checkIP(ip) { + try { + new IP(ip) + } catch { + throw `malformed IP address '${ip}'` + } +} + +function checkHost(host) { + if ( + typeof host !== 'string' || + host.indexOf(':') >= 0 || + host.indexOf('[') >= 0 || + host.indexOf(']') >= 0 + ) throw `invalid host '${host}'` +} + +function checkPort(port) { + if ( + typeof port !== 'number' || + port < 1 || port > 65535 + ) throw `invalid port number: ${port}` +} + +function checkUUID(uuid) { + if ( + typeof uuid !== 'string' || + uuid.length !== 36 || + uuid.charAt(8) != '-' || + uuid.charAt(13) != '-' || + uuid.charAt(18) != '-' || + uuid.charAt(23) != '-' + ) throw `malformed UUID '${uuid}'` +} + +function checkListens(listens) { + if (!(listens instanceof Array)) throw 'invalid listen array' + listens.forEach(l => { + if (typeof l !== 'object' || l === null) throw 'invalid listen' + checkIP(l.ip) + checkPort(l.port) + }) +} + +function checkTargets(targets) { + if (!(targets instanceof Array)) throw 'invalid target array' + targets.forEach(t => { + if (typeof t !== 'object' || t === null) throw 'invalid target' + checkHost(t.host) + checkPort(t.port) + }) +} + +function checkExits(exits) { + if (!(exits instanceof Array)) throw 'invalid exit array' + exits.forEach(e => checkUUID(e)) +} + +function checkEntrances(entrances) { + if (!(entrances instanceof Array)) throw 'invalid entrance array' + entrances.forEach(e => checkUUID(e)) +} diff --git a/neptune/mega/tunnel_punch/cli.js b/neptune/mega/tunnel_punch/cli.js new file mode 100644 index 000000000..6f2416dfb --- /dev/null +++ b/neptune/mega/tunnel_punch/cli.js @@ -0,0 +1,303 @@ +export default function ({ api, utils }) { + return pipeline($=>$ + .onStart(ctx => main(ctx)) + ) + + function main({ argv, endpoint }) { + var buffer = new Data + + function output(str) { + buffer.push(str) + } + + function error(err) { + output('ztm: ') + output(err.message || err.toString()) + output('\n') + } + + function flush() { + return [buffer, new StreamEnd] + } + + var endpoints = null + + function allEndpoints() { + if (endpoints) return Promise.resolve(endpoints) + return api.allEndpoints().then(list => (endpoints = list)) + } + + function lookupEndpointNames(list) { + return allEndpoints().then(endpoints => ( + list.map(id => { + var ep = endpoints.find(ep => ep.id === id) + return ep ? ep.name : id + }) + )) + } + + function lookupEndpointIDs(list) { + return allEndpoints().then(endpoints => ( + list.flatMap(name => { + if (endpoints.some(ep => ep.id === name)) return name + var list = endpoints.filter(ep => ep.name === name) + if (list.length === 1) return list[0].id + if (list.length === 0) throw `Endpoint '${name}' not found` + return list.map(ep => ep.id) + }) + )) + } + + try { + return utils.parseArgv(['ztm tunnel', ...argv], { + help: text => Promise.resolve(output(text + '\n')), + notes: objectTypeNotes + objectNameNotes, + commands: [ + + { + title: 'List objects of the specified type', + usage: 'get ', + notes: objectTypeNotes, + action: (args) => { + switch (validateObjectType(args, 'get')) { + case 'inbound': return getInbound() + case 'outbound': return getOutbound() + } + } + }, + + { + title: 'Show detailed info of the specified object', + usage: 'describe ', + notes: objectTypeNotes + objectNameNotes, + action: (args) => { + var name = args[''] + switch (validateObjectType(args, 'describe')) { + case 'inbound': return describeInbound(name) + case 'outbound': return describeOutbound(name) + } + } + }, + + { + title: 'Create an object of the specified type', + usage: 'open ', + options: ` + For inbound end: + + --listen <[ip:]port ...> Set local ports to listen on + --exit Select endpoints as the outbound end + + For outbound end: + + --target Set targets to connect to + --entrance Select endpoints as the inbound end + `, + notes: objectTypeNotes + objectNameNotes, + action: (args) => { + var name = args[''] + switch (validateObjectType(args, 'open')) { + case 'inbound': return openInbound(name, args['--listen'], args['--exit']) + case 'outbound': return openOutbound(name, args['--target'], args['--entrance']) + } + } + }, + + { + title: 'Delete the specified object', + usage: 'close ', + notes: objectTypeNotes + objectNameNotes, + action: (args) => { + var name = args[''] + switch (validateObjectType(args, 'close')) { + case 'inbound': return closeInbound(name) + case 'outbound': return closeOutbound(name) + } + } + }, + ] + + }).then(flush).catch(err => { + error(err) + return flush() + }) + + } catch (err) { + error(err) + return Promise.resolve(flush()) + } + + function getInbound() { + return api.allInbound(endpoint.id).then(list => ( + Promise.all(list.map(i => + lookupEndpointNames(i.exits || []).then(exits => ({ + ...i, + exits, + })) + )).then(list => + printTable(list, { + 'NAME': i => `${i.protocol}/${i.name}`, + 'LISTENS': i => i.listens.map(l => `${l.ip}:${l.port}`).join(', '), + 'EXITS': i => i.exits.join(', '), + }) + ) + )) + } + + function getOutbound() { + return api.allOutbound(endpoint.id).then(list => ( + Promise.all(list.map(o => + lookupEndpointNames(o.entrances || []).then(entrances => ({ + ...o, + entrances, + })) + )).then(list => + printTable(list, { + 'NAME': o => `${o.protocol}/${o.name}`, + 'TARGETS': o => o.targets.map(t => `${t.host}:${t.port}`).join(', '), + 'ENTRANCES': o => o.entrances.join(', '), + }) + ) + )) + } + + function describeInbound(tunnelName) { + var obj = validateObjectName(tunnelName) + return api.getInbound(endpoint.id, obj.protocol, obj.name).then(obj => { + if (!obj) return + return lookupEndpointNames(obj.exits || []).then(exits => { + output(`Inbound ${obj.protocol}/${obj.name}\n`) + output(`Endpoint: ${endpoint.name} (${endpoint.id})\n`) + output(`Listens:\n`) + obj.listens.forEach(l => output(` ${l.ip}:${l.port}\n`)) + output(`Exits:\n`) + exits.forEach(e => output(` ${e}\n`)) + if (exits.length === 0) output(` (all endpoints)\n`) + }) + }) + } + + function describeOutbound(tunnelName) { + var obj = validateObjectName(tunnelName) + return api.getOutbound(endpoint.id, obj.protocol, obj.name).then(obj => { + if (!obj) return + return lookupEndpointNames(obj.entrances || []).then(entrances => { + output(`Outbound ${obj.protocol}/${obj.name}\n`) + output(`Endpoint: ${endpoint.name} (${endpoint.id})\n`) + output(`Targets:\n`) + obj.targets.forEach(t => output(` ${t.host}:${t.port}\n`)) + output(`Entrances:\n`) + entrances.forEach(e => output(` ${e}\n`)) + if (entrances.length === 0) output(` (all endpoints)\n`) + }) + }) + } + + function openInbound(tunnelName, listens, exits) { + var obj = validateObjectName(tunnelName) + if (!listens || listens.length === 0) throw `Option '--listen' is required` + listens = listens.map(l => validateHostPort(l)).map(({ host, port }) => ({ ip: host, port })) + return lookupEndpointIDs(exits || []).then( + exits => api.setInbound(endpoint.id, obj.protocol, obj.name, listens, exits) + ) + } + + function openOutbound(tunnelName, targets, entrances) { + var obj = validateObjectName(tunnelName) + if (!targets || targets.length === 0) throw `Option '--target' is required` + targets = targets.map(t => validateHostPort(t)) + return lookupEndpointIDs(entrances || []).then( + entrances => api.setOutbound(endpoint.id, obj.protocol, obj.name, targets, entrances) + ) + } + + function closeInbound(tunnelName) { + var obj = validateObjectName(tunnelName) + return api.deleteInbound(endpoint.id, obj.protocol, obj.name) + } + + function closeOutbound(tunnelName) { + var obj = validateObjectName(tunnelName) + return api.deleteOutbound(endpoint.id, obj.protocol, obj.name) + } + + function validateObjectType(args, command) { + var ot = args[''] + switch (ot) { + case 'inbound': + case 'in': + return 'inbound' + case 'outbound': + case 'out': + return 'outbound' + default: throw `Invalid object type '${ot}'. Type 'ztm tunnel ${command} for help.'` + } + } + + function validateObjectName(name) { + if (!name) return + var segs = name.split('/') + if (segs.length === 2) { + var protocol = segs[0] + if (protocol === 'tcp' || protocol === 'udp') { + if (segs[1]) { + return { protocol, name: segs[1] } + } + } + } + throw `Invalid inbound/outbound name '${name}'` + } + + function validateHostPort(addr) { + var i = addr.lastIndexOf(':') + if (i >= 0) { + var host = addr.substring(0,i) + var port = addr.substring(i+1) + } else { + var host = '' + var port = addr + } + port = Number.parseInt(port) + if (Number.isNaN(port)) throw `Invalid port number in '${addr}'` + if (!host) host = '127.0.0.1' + return { host, port } + } + + function printTable(data, columns, indent) { + var head = ' '.repeat(indent || 0) + var cols = Object.entries(columns) + var colHeaders = cols.map(i => i[0]) + var colFormats = cols.map(i => i[1]) + var colSizes = colHeaders.map(name => name.length) + var rows = data.map(row => colFormats.map( + (format, i) => { + var v = format(row).toString() + colSizes[i] = Math.max(colSizes[i], v.length) + return v + } + )) + output(head) + colHeaders.forEach((name, i) => output(name.padEnd(colSizes[i]) + ' ')) + output('\n') + rows.forEach(row => { + output(head) + row.forEach((v, i) => output(v.padEnd(colSizes[i]) + ' ')) + output('\n') + }) + } + } +} + +var objectTypeNotes = ` + Object Types: + + inbound in Inbound end of a tunnel + outbound out Outbound end of a tunnel +` + +var objectNameNotes = ` + Object Names: + + tcp/ Name for a TCP tunnel + udp/ Name for a UDP tunnel +` diff --git a/neptune/mega/tunnel_punch/main.js b/neptune/mega/tunnel_punch/main.js new file mode 100644 index 000000000..4089ea68a --- /dev/null +++ b/neptune/mega/tunnel_punch/main.js @@ -0,0 +1,222 @@ +import initHole from './punch.js' +import initAPI from './api.js' +import initCLI from './cli.js' + +export default function ({ app, mesh, utils }) { + var punch = initHole({ app, mesh }) + var api = initAPI({ app, mesh, punch }) + var cli = initCLI({ app, mesh, utils, api }) + + var $ctx + + var gui = new http.Directory(os.path.join(app.root, 'gui')) + var response = utils.createResponse + var responder = utils.createResponder + + var serveUser = utils.createServer({ + '/cli': { + 'CONNECT': utils.createCLIResponder(cli), + }, + + '/api/appinfo': { + 'GET': responder(() => Promise.resolve(response(200, { + name: app.name, + provider: app.provider, + username: app.username, + endpoint: app.endpoint, + }))) + }, + + '/api/endpoints': { + 'GET': responder(() => api.allEndpoints().then( + ret => ret ? response(200, ret) : response(404) + )) + }, + + '/api/endpoints/{ep}/inbound': { + 'GET': responder(({ ep }) => api.allInbound(ep).then( + ret => ret ? response(200, ret) : response(404) + )) + }, + + '/api/endpoints/{ep}/outbound': { + 'GET': responder(({ ep }) => api.allOutbound(ep).then( + ret => ret ? response(200, ret) : response(404) + )) + }, + + '/api/endpoints/{ep}/inbound/{proto}/{name}': { + 'GET': responder(({ ep, proto, name }) => { + return api.getInbound(ep, proto, name).then( + ret => ret ? response(200, ret) : response(404) + ) + }), + + 'POST': responder(({ ep, proto, name }, req) => { + var obj = JSON.decode(req.body) + var listens = obj.listens + var exits = obj.exits || null + return api.setInbound(ep, proto, name, listens, exits).then(response(201)) + }), + + 'DELETE': responder(({ ep, proto, name }) => { + return api.deleteInbound(ep, proto, name).then(response(204)) + }), + + }, + + '/api/endpoints/{ep}/outbound/{proto}/{name}': { + 'GET': responder(({ ep, proto, name }) => { + return api.getOutbound(ep, proto, name).then( + ret => ret ? response(200, ret) : response(404) + ) + }), + + 'POST': responder(({ ep, proto, name }, req) => { + var obj = JSON.decode(req.body) + var targets = obj.targets + var entrances = obj.entrances || null + return api.setOutbound(ep, proto, name, targets, entrances).then(response(201)) + }), + + 'DELETE': responder(({ ep, proto, name }) => { + return api.deleteOutbound(ep, proto, name).then(response(204)) + }), + }, + + // '/api/punch/{destEp}': { + // 'GET': responder(({destEp}) => { + // return api.createHole(destEp) + // }), + + // 'DELETE': responder(({destEp}) => { + // api.deleteHole(destEp) + // return response(204) + // }) + // }, + + '*': { + 'GET': responder((_, req) => { + return Promise.resolve(gui.serve(req) || response(404)) + }) + }, + }) + + var servePeer = utils.createServer({ + '/api/inbound': { + 'GET': responder(() => api.allInbound(app.endpoint.id).then( + ret => ret ? response(200, ret) : response(404) + )) + }, + + '/api/outbound': { + 'GET': responder(() => api.allOutbound(app.endpoint.id).then( + ret => ret ? response(200, ret) : response(404) + )) + }, + + '/api/inbound/{proto}/{name}': { + 'GET': responder(({ proto, name }) => api.getInbound(app.endpoint.id, proto, name).then( + ret => ret ? response(200, ret) : response(404) + )), + + 'POST': responder(({ proto, name }, req) => { + var obj = JSON.decode(req.body) + var listens = obj.listens + var exits = obj.exits || null + return api.setInbound(app.endpoint.id, proto, name, listens, exits).then(response(201)) + }), + + 'DELETE': responder(({ proto, name }) => { + return api.deleteInbound(app.endpoint.id, proto, name).then(response(204)) + }), + }, + + '/api/outbound/{proto}/{name}': { + 'GET': responder(({ proto, name }) => api.getOutbound(app.endpoint.id, proto, name).then( + ret => ret ? response(200, ret) : response(404) + )), + + 'POST': responder(({ proto, name }, req) => { + var obj = JSON.decode(req.body) + var targets = obj.targets + var entrances = obj.entrances || null + return api.setOutbound(app.endpoint.id, proto, name, targets, entrances).then(response(201)) + }), + + 'DELETE': responder(({ proto, name }) => { + return api.deleteOutbound(app.endpoint.id, proto, name).then(response(204)) + }), + + 'CONNECT': api.servePeerInbound, + }, + + '/api/ping': { + 'GET': responder(() => Promise.resolve(response(200))) + }, + + '/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 '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}`) + console.log("Punch req: ", obj) + switch(action) { + case 'request': + 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, obj.cert) + api.syncPunch(ep) + break + default: + return Promise.resolve(response(500, "Unknown punch action")) + } + return Promise.resolve(response(200)) + }), + + 'CONNECT': pipeline($=>$.pipe(api.makeRespTunnel, () => $ctx)) + }, + }) + + punch.setService((ctx) => { + // Tricky callback to set ctx, + // expecting everything in hole works + // just like it's coming from hub. + $ctx = ctx + return servePeer + }) + + return pipeline($=>$ + .onStart(c => void ($ctx = c)) + .pipe(() => { + switch ($ctx.source) { + case 'user': return serveUser + case 'peer': return servePeer + } + }) + ) +} diff --git a/neptune/mega/tunnel/punch.js b/neptune/mega/tunnel_punch/punch.js similarity index 100% rename from neptune/mega/tunnel/punch.js rename to neptune/mega/tunnel_punch/punch.js