Implement blind relay support

examples/connection-relaying/client.js

@@ -0,0 +1,14 @@
+const DHT = require('../..')
+
+const [relay, server] = process.argv.slice(2)
+
+const dht = new DHT()
+
+const socket = dht.connect(Buffer.from(server, 'hex'), {
+  localConnection: false,
+  relayThrough: Buffer.from(relay, 'hex')
+})
+
+console.log('Client connecting from', socket.publicKey.toString('hex'))
+
+socket.end('Hello!')

examples/connection-relaying/relay.js

@@ -0,0 +1,23 @@
+const RelayServer = require('protomux-bridging-relay').Server
+const DHT = require('../..')
+
+const dht = new DHT()
+
+const relay = new RelayServer({
+  createStream (opts) {
+    return dht.createRawStream({ ...opts, framed: true })
+  }
+})
+
+const server = dht.createServer({ shareLocalAddress: false }, (socket) => {
+  console.log('Connection from', socket.remotePublicKey.toString('hex'))
+  const session = relay.accept(socket, { id: socket.remotePublicKey })
+  session
+    .on('pair', (isInitiator, token, stream, remoteId) => {
+      console.log('Pair isInitiator =', isInitiator, 'token =', token.toString('hex'))
+    })
+})
+
+server
+  .listen()
+  .then(() => console.log('Relay listening on', server.publicKey.toString('hex')))

examples/connection-relaying/server.js

@@ -0,0 +1,14 @@
+const DHT = require('../..')
+
+const dht = new DHT()
+
+const server = dht.createServer({ shareLocalAddress: false }, (socket) => {
+  console.log('Connection from', socket.remotePublicKey.toString('hex'))
+  socket
+    .on('data', (data) => console.log(data.toString()))
+    .end()
+})
+
+server
+  .listen()
+  .then(() => console.log('Server listening on', server.publicKey.toString('hex')))

index.js

@@ -11,6 +11,7 @@ const connect = require('./lib/connect')
 const { FIREWALL, BOOTSTRAP_NODES, COMMANDS } = require('./lib/constants')
 const { hash, createKeyPair } = require('./lib/crypto')
 const RawStreamSet = require('./lib/raw-stream-set')
+const ConnectionPool = require('./lib/connection-pool')
 const { STREAM_NOT_CONNECTED } = require('./lib/errors')
 
 const maxSize = 65536
@@ -60,6 +61,10 @@ class HyperDHT extends DHT {
     return s
   }
 
+  pool () {
+    return new ConnectionPool(this)
+  }
+
   async destroy ({ force } = {}) {
     if (!force) {
       const closing = []

lib/connect.js

@@ -1,5 +1,6 @@
 const NoiseSecretStream = require('@hyperswarm/secret-stream')
 const b4a = require('b4a')
+const relay = require('protomux-bridging-relay')
 const DebuggingStream = require('debugging-stream')
 const { isPrivate } = require('bogon')
 const Semaphore = require('./semaphore')
@@ -26,16 +27,23 @@ const {
 } = require('./errors')
 
 module.exports = function connect (dht, publicKey, opts = {}) {
+  const pool = opts.pool || null
+
+  if (pool && pool.has(publicKey)) return pool.get(publicKey)
+
   const keyPair = opts.keyPair || dht.defaultKeyPair
   const encryptedSocket = (opts.createSecretStream || defaultCreateSecretStream)(true, null, {
     publicKey: keyPair.publicKey,
     remotePublicKey: publicKey,
     autoStart: false
   })
 
+  if (pool) pool._attachStream(encryptedSocket, false)
+
   const c = {
     dht,
     session: dht.session(),
+    pool,
     round: 0,
     target: hash(publicKey),
     remotePublicKey: publicKey,
@@ -45,7 +53,7 @@ module.exports = function connect (dht, publicKey, opts = {}) {
     requesting: false,
     lan: opts.localConnection !== false,
     firewall: FIREWALL.UNKNOWN,
-    rawStream: dht._rawStreams.add({ framed: true, firewall }),
+    rawStream: dht.createRawStream({ framed: true, firewall }),
     connect: null,
     query: null,
     puncher: null,
@@ -55,7 +63,14 @@ module.exports = function connect (dht, publicKey, opts = {}) {
     serverAddress: null,
     onsocket: null,
     sleeper: new Sleeper(),
-    encryptedSocket
+    encryptedSocket,
+
+    // Relay state
+    relayThrough: opts.relayThrough || null,
+    relayToken: opts.relayThrough ? relay.token() : null,
+    relaySocket: null,
+    relayClient: null,
+    relayPaired: false
   }
 
   // If the raw stream receives an error signal pre connect (ie from the firewall hook), make sure
@@ -156,7 +171,9 @@ async function holepunch (c, opts) {
     // TODO: check all addresses also obvs
   }
 
-  const onabort = () => destroyEncryptedSocket(c, HOLEPUNCH_ABORTED())
+  const onabort = () => {
+    if (c.relayToken === null) destroyEncryptedSocket(c, HOLEPUNCH_ABORTED())
+  }
 
   if (c.firewall === FIREWALL.OPEN) {
     c.passiveConnectTimeout = setTimeout(onabort, 10000)
@@ -294,7 +311,10 @@ async function connectThroughNode (c, address, socket) {
         id: c.rawStream.id,
         seq: 0
       },
-      secretStream: {}
+      secretStream: {},
+      relayThrough: c.relayThrough
+        ? { publicKey: c.relayThrough, token: c.relayToken }
+        : null
     })
     if (isDone(c)) return
   }
@@ -336,12 +356,17 @@ async function connectThroughNode (c, address, socket) {
   c.onsocket = function (socket, port, host) {
     if (c.rawStream === null) return // Already hole punched
 
-    const rawStream = c.dht._debugStream !== null
-      ? new DebuggingStream(c.rawStream, c.dht._debugStream)
-      : c.rawStream
+    if (c.rawStream.connected) {
+      c.rawStream.changeRemote(socket, c.connect.payload.udx.id, port, host)
+    } else {
+      c.rawStream.connect(socket, c.connect.payload.udx.id, port, host)
 
-    c.rawStream.connect(socket, c.connect.payload.udx.id, port, host)
-    c.encryptedSocket.start(rawStream, { handshake: hs })
+      const rawStream = c.dht._debugStream !== null
+        ? new DebuggingStream(c.rawStream, c.dht._debugStream)
+        : c.rawStream
+
+      c.encryptedSocket.start(rawStream, { handshake: hs })
+    }
 
     if (c.reusableSocket && payload.udx.reusableSocket) {
       c.dht._socketPool.routes.add(c.remotePublicKey, c.rawStream)
@@ -359,6 +384,50 @@ async function connectThroughNode (c, address, socket) {
     c.rawStream = null
   }
 
+  if (payload.relayThrough || c.relayThrough) {
+    let isInitiator
+    let publicKey
+    let token
+
+    if (payload.relayThrough) {
+      isInitiator = false
+      publicKey = payload.relayThrough.publicKey
+      token = payload.relayThrough.token
+    } else {
+      isInitiator = true
+      publicKey = c.relayThrough
+      token = c.relayToken
+    }
+
+    c.relayToken = token
+    c.relaySocket = c.dht.connect(publicKey)
+    c.relayClient = relay.Client.from(c.relaySocket, { id: c.relaySocket.publicKey })
+
+    c.relayClient
+      .pair(isInitiator, token, c.rawStream)
+      .on('error', () => c.relaySocket.destroy())
+      .on('data', (remoteId) => {
+        if (c.rawStream === null) return
+        c.relayPaired = true
+
+        const {
+          remotePort,
+          remoteHost,
+          socket
+        } = c.relaySocket.rawStream
+
+        c.rawStream
+          .on('close', () => c.relaySocket.destroy())
+          .connect(socket, remoteId, remotePort, remoteHost)
+
+        const rawStream = c.dht._debugStream !== null
+          ? new DebuggingStream(c.rawStream, c.dht._debugStream)
+          : c.rawStream
+
+        c.encryptedSocket.start(rawStream, { handshake: hs })
+      })
+  }
+
   if (c.serverSocket) {
     c.onsocket(c.serverSocket, c.serverAddress.port, c.serverAddress.host)
     return
@@ -387,6 +456,8 @@ async function updateHolepunch (c, peerAddress, relayAddr, payload) {
 
   const { error, firewall, punching, addresses, remoteToken } = remotePayload
   if (error !== ERROR.NONE) {
+    if (c.relayToken) return null // Don't throw if we're being relayed
+
     throw REMOTE_ABORTED('Remote aborted with error code ' + error)
   }
 
@@ -484,6 +555,8 @@ async function roundPunch (c, serverAddress, remoteToken, clientRelay) {
     remoteToken
   })
 
+  if (c.relayToken) return // Don't throw if we're being relayed
+
   if (!c.puncher.remoteHolepunching) {
     throw REMOTE_NOT_HOLEPUNCHING()
   }
@@ -512,7 +585,8 @@ async function abort (c, { peerAddress, relayAddress }, err) {
       remoteToken: null
     })
   } catch {}
-  destroyEncryptedSocket(c, err)
+
+  if (c.relayToken === null) destroyEncryptedSocket(c, err)
 }
 
 function clearPassiveConnectTimeout (c) {