Hole punch support for neptune

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(
@@ -170,6 +173,33 @@ export default function ({ app, mesh }) {
     }
   }
 
+  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: [] }
@@ -197,23 +227,34 @@ export default function ({ app, mesh }) {
     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($=>$
-          .muxHTTP().to($=>$
-            .pipe(() => mesh.connect($selectedEP))
+        ).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}`))
       )
 
@@ -285,14 +326,24 @@ export default function ({ app, mesh }) {
     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(() => $response)
+      .onEnd(() => {
+        console.info('Answers in api: ', $response)
+        return $response
+      })
     ).spawn()
   }
 
@@ -343,6 +394,22 @@ export default function ({ app, mesh }) {
     )
   )
 
+  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 {
@@ -355,6 +422,11 @@ export default function ({ app, mesh }) {
     setOutbound,
     deleteInbound,
     deleteOutbound,
+    createHole,
+    updateHoleInfo,
+    makeRespTunnel,
+    syncPunch,
+    deleteHole,
     servePeerInbound,
   }
 }

neptune/mega/tunnel/main.js

@@ -1,8 +1,10 @@
+import initHole from './punch.js'
 import initAPI from './api.js'
 import initCLI from './cli.js'
 
 export default function ({ app, mesh, utils }) {
-  var api = initAPI({ app, mesh })
+  var punch = initHole({ app, mesh })
+  var api = initAPI({ app, mesh, punch })
   var cli = initCLI({ app, mesh, utils, api })
 
   var $ctx
@@ -82,6 +84,17 @@ 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))
@@ -137,6 +150,64 @@ 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($=>$