hyperdht - npm Package Compare versions

Comparing version 6.6.2 to 6.7.0

index.js

@@ -14,7 +14,5 @@ const DHT = require('dht-rpc')
 const RawStreamSet = require('./lib/raw-stream-set')
+const ConnectionPool = require('./lib/connection-pool')
 const { STREAM_NOT_CONNECTED } = require('./lib/errors')
 
-const maxSize = 65536
-const maxAge = 20 * 60 * 1000
-
 class HyperDHT extends DHT {
@@ -27,6 +25,3 @@ constructor (opts = {}) {
 
-    const cacheOpts = {
-      maxSize: opts.maxSize || maxSize,
-      maxAge: opts.maxAge || maxAge
-    }
+    const { router, persistent } = defaultCacheOpts(opts)
 
@@ -36,3 +31,3 @@ this.defaultKeyPair = opts.keyPair || createKeyPair(opts.seed)
 
-    this._router = new Router(this, cacheOpts)
+    this._router = new Router(this, router)
     this._socketPool = new SocketPool(this, opts.host || '0.0.0.0')
@@ -46,3 +41,3 @@ this._rawStreams = new RawStreamSet(this)
     this.once('persistent', () => {
-      this._persistent = new Persistent(this, cacheOpts)
+      this._persistent = new Persistent(this, persistent)
     })
@@ -67,2 +62,6 @@
 
+  pool () {
+    return new ConnectionPool(this)
+  }
+
   async destroy ({ force } = {}) {
@@ -466,1 +465,27 @@ if (!force) {
 }
+
+const defaultMaxSize = 65536
+const defaultMaxAge = 20 * 60 * 1000 // 20 minutes
+
+function defaultCacheOpts (opts) {
+  const maxSize = opts.maxSize || defaultMaxSize
+  const maxAge = opts.maxAge || defaultMaxAge
+
+  return {
+    router: {
+      forwards: { maxSize, maxAge }
+    },
+    persistent: {
+      records: { maxSize, maxAge },
+      refreshes: { maxSize, maxAge },
+      mutables: {
+        maxSize: maxSize / 2 | 0,
+        maxAge: opts.maxAge || 48 * 60 * 60 * 1000 // 48 hours
+      },
+      immutables: {
+        maxSize: maxSize / 2 | 0,
+        maxAge: opts.maxAge || 48 * 60 * 60 * 1000 // 48 hours
+      }
+    }
+  }
+}

lib/connect.js

 const NoiseSecretStream = require('@hyperswarm/secret-stream')
 const b4a = require('b4a')
+const relay = require('blind-relay')
 const DebuggingStream = require('debugging-stream')
@@ -29,2 +30,6 @@ const { isPrivate } = require('bogon')
 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
@@ -37,5 +42,8 @@ const encryptedSocket = (opts.createSecretStream || defaultCreateSecretStream)(true, null, {
 
+  if (pool) pool._attachStream(encryptedSocket, false)
+
   const c = {
     dht,
     session: dht.session(),
+    pool,
     round: 0,
@@ -50,3 +58,3 @@ target: hash(publicKey),
     firewall: FIREWALL.UNKNOWN,
-    rawStream: dht._rawStreams.add({ framed: true, firewall }),
+    rawStream: dht.createRawStream({ framed: true, firewall }),
     connect: null,
@@ -61,3 +69,10 @@ query: null,
     sleeper: new Sleeper(),
-    encryptedSocket
+    encryptedSocket,
+
+    // Relay state
+    relayThrough: opts.relayThrough || null,
+    relayToken: opts.relayThrough ? relay.token() : null,
+    relaySocket: null,
+    relayClient: null,
+    relayPaired: false
   }
@@ -93,2 +108,8 @@
   function firewall (socket, port, host) {
+    // Check if the traffic originated from the socket on which we're expecting relay traffic. If so,
+    // we haven't hole punched yet and the other side is just sending us traffic through the relay.
+    if (c.relaySocket && c.relaySocket.rawStream && c.relaySocket.rawStream.socket === socket) {
+      return false
+    }
+
     if (c.onsocket) {
@@ -164,3 +185,5 @@ c.onsocket(socket, port, host)
 
-  const onabort = () => destroyEncryptedSocket(c, HOLEPUNCH_ABORTED())
+  const onabort = () => {
+    if (c.relayToken === null) destroyEncryptedSocket(c, HOLEPUNCH_ABORTED())
+  }
 
@@ -303,3 +326,6 @@ if (c.firewall === FIREWALL.OPEN) {
       },
-      secretStream: {}
+      secretStream: {},
+      relayThrough: c.relayThrough
+        ? { publicKey: c.relayThrough, token: c.relayToken }
+        : null
     })
@@ -346,9 +372,14 @@ if (isDone(c)) return
 
-    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) {
@@ -370,2 +401,6 @@ c.dht._socketPool.routes.add(c.remotePublicKey, c.rawStream)
 
+  if (payload.relayThrough || c.relayThrough) {
+    relayConnection(c, c.relayThrough, payload, hs)
+  }
+
   if (c.serverSocket) {
@@ -399,3 +434,4 @@ c.onsocket(c.serverSocket, c.serverAddress.port, c.serverAddress.host)
   if (error !== ERROR.NONE) {
-    throw REMOTE_ABORTED('Remote aborted with error code ' + error)
+    // Don't throw if we're being relayed
+    if (!c.relayToken) throw REMOTE_ABORTED('Remote aborted with error code ' + error)
   }
@@ -498,7 +534,9 @@
   if (!c.puncher.remoteHolepunching) {
-    throw REMOTE_NOT_HOLEPUNCHING()
+    // Don't throw if we're being relayed
+    if (!c.relayToken) throw REMOTE_NOT_HOLEPUNCHING()
   }
 
   if (!await c.puncher.punch()) {
-    throw REMOTE_NOT_HOLEPUNCHABLE()
+    // Don't throw if we're being relayed
+    if (!c.relayToken) throw REMOTE_NOT_HOLEPUNCHABLE()
   }
@@ -526,5 +564,52 @@ }
   } catch {}
-  destroyEncryptedSocket(c, err)
+
+  if (c.relayToken === null) destroyEncryptedSocket(c, err)
 }
 
+function relayConnection (c, relayThrough, payload, hs) {
+  if (c.passiveConnectTimeout) clearPassiveConnectTimeout(c)
+
+  let isInitiator
+  let publicKey
+  let token
+
+  if (payload.relayThrough) {
+    isInitiator = false
+    publicKey = payload.relayThrough.publicKey
+    token = payload.relayThrough.token
+  } else {
+    isInitiator = true
+    publicKey = 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 })
+    })
+}
+
 function clearPassiveConnectTimeout (c) {
@@ -531,0 +616,0 @@ clearTimeout(c.passiveConnectTimeout)

lib/errors.js

@@ -90,2 +90,6 @@ module.exports = class DHTError extends Error {
   }
+
+  static DUPLICATE_CONNECTION (msg = 'Duplicate connection') {
+    return new DHTError(msg, 'DUPLICATE_CONNECTION', DHTError.DUPLICATE_CONNECTION)
+  }
 }

lib/messages.js

@@ -131,2 +131,27 @@ const c = require('compact-encoding')
 
+const relayThroughInfo = {
+  preencode (state, m) {
+    c.uint.preencode(state, 1) // version
+    c.uint.preencode(state, 0) // flags
+    c.fixed32.preencode(state, m.publicKey)
+    c.fixed32.preencode(state, m.token)
+  },
+  encode (state, m) {
+    c.uint.encode(state, 1)
+    c.uint.encode(state, 0)
+    c.fixed32.encode(state, m.publicKey)
+    c.fixed32.encode(state, m.token)
+  },
+  decode (state) {
+    const version = c.uint.decode(state)
+    c.uint.decode(state)
+
+    return {
+      version,
+      publicKey: c.fixed32.decode(state),
+      token: c.fixed32.decode(state)
+    }
+  }
+}
+
 exports.noisePayload = {
@@ -140,2 +165,3 @@ preencode (state, m) {
     if (m.secretStream) secretStreamInfo.preencode(state, m.secretStream)
+    if (m.relayThrough) relayThroughInfo.preencode(state, m.relayThrough)
   },
@@ -150,2 +176,3 @@ encode (state, m) {
     if (m.secretStream) flags |= 16
+    if (m.relayThrough) flags |= 32
 
@@ -162,2 +189,3 @@ c.uint.encode(state, 1) // version
     if (m.secretStream) secretStreamInfo.encode(state, m.secretStream)
+    if (m.relayThrough) relayThroughInfo.encode(state, m.relayThrough)
   },
@@ -178,3 +206,4 @@ decode (state) {
         udx: null,
-        secretStream: null
+        secretStream: null,
+        relayThrough: null
       }
@@ -193,3 +222,4 @@ }
       udx: (flags & 8) !== 0 ? udxInfo.decode(state) : null,
-      secretStream: (flags & 16) !== 0 ? secretStreamInfo.decode(state) : null
+      secretStream: (flags & 16) !== 0 ? secretStreamInfo.decode(state) : null,
+      relayThrough: (flags & 32) !== 0 ? relayThroughInfo.decode(state) : null
     }
@@ -196,0 +226,0 @@ }

lib/persistent.js

@@ -15,9 +15,8 @@ const c = require('compact-encoding')
 module.exports = class Persistent {
-  constructor (dht, { maxSize, maxAge }) {
+  constructor (dht, opts) {
     this.dht = dht
-    // TODO: should prob be more clear about maxSize here since to make many caches
-    this.records = new RecordCache({ maxSize, maxAge })
-    this.refreshes = new Cache({ maxSize, maxAge })
-    this.mutables = new Cache({ maxSize: Math.floor(maxSize / 2), maxAge })
-    this.immutables = new Cache({ maxSize: Math.floor(maxSize / 2), maxAge })
+    this.records = new RecordCache(opts.records)
+    this.refreshes = new Cache(opts.refreshes)
+    this.mutables = new Cache(opts.mutables)
+    this.immutables = new Cache(opts.immutables)
   }
@@ -24,0 +23,0 @@

lib/router.js

@@ -23,3 +23,3 @@ const c = require('compact-encoding')
     this.dht = dht
-    this.forwards = new Cache(opts)
+    this.forwards = new Cache(opts.forwards)
   }
@@ -26,0 +26,0 @@

lib/server.js

@@ -5,2 +5,3 @@ const { EventEmitter } = require('events')
 const b4a = require('b4a')
+const relay = require('blind-relay')
 const NoiseWrap = require('./noise-wrap')
@@ -30,2 +31,4 @@ const Announcer = require('./announcer')
     this.holepunch = opts.holepunch || (() => true)
+    this.relayThrough = opts.relayThrough || null
+    this.pool = opts.pool || null
     this.createHandshake = opts.createHandshake || defaultCreateHandshake
@@ -124,2 +127,4 @@ this.createSecretStream = opts.createSecretStream || defaultCreateSecretStream
 
+    if (this.pool) this.pool._attachServer(this)
+
     return this
@@ -142,6 +147,13 @@ }
       rawStream: null,
+      encryptedSocket: null,
       prepunching: null,
       firewalled: true,
       clearing: null,
-      onsocket: null
+      onsocket: null,
+
+      // Relay state
+      relayToken: null,
+      relaySocket: null,
+      relayClient: null,
+      relayPaired: false
     }
@@ -187,5 +199,11 @@
     if (error === ERROR.NONE) {
-      hs.rawStream = this.dht._rawStreams.add({
+      hs.rawStream = this.dht.createRawStream({
         framed: true,
         firewall (socket, port, host) {
+          // Check if the traffic originated from the socket on which we're expecting relay traffic. If so,
+          // we haven't hole punched yet and the other side is just sending us traffic through the relay.
+          if (hs.relaySocket && hs.relaySocket.rawStream && hs.relaySocket.rawStream.socket === socket) {
+            return false
+          }
+
           hs.onsocket(socket, port, host)
@@ -208,6 +226,2 @@ return false
 
-        const rawStream = this.dht._debugStream !== null
-          ? new DebuggingStream(hs.rawStream, this.dht._debugStream)
-          : hs.rawStream
-
         if (this._reusableSocket && remotePayload.udx.reusableSocket) {
@@ -218,6 +232,17 @@ this.dht._socketPool.routes.add(handshake.remotePublicKey, hs.rawStream)
         hs.rawStream.removeListener('error', autoDestroy)
-        hs.rawStream.connect(socket, remotePayload.udx.id, port, host)
 
-        this.onconnection(this.createSecretStream(false, rawStream, { handshake: h }))
+        if (hs.rawStream.connected) {
+          hs.rawStream.changeRemote(socket, remotePayload.udx.id, port, host)
+        } else {
+          hs.rawStream.connect(socket, remotePayload.udx.id, port, host)
 
+          const rawStream = this.dht._debugStream !== null
+            ? new DebuggingStream(hs.rawStream, this.dht._debugStream)
+            : hs.rawStream
+
+          hs.encryptedSocket = this.createSecretStream(false, rawStream, { handshake: h })
+
+          this.onconnection(hs.encryptedSocket)
+        }
+
         if (hs.puncher) {
@@ -236,2 +261,6 @@ hs.puncher.onabort = noop
 
+    const relayThrough = typeof this.relayThrough === 'function' ? this.relayThrough() : this.relayThrough
+
+    if (relayThrough) hs.relayToken = relay.token()
+
     try {
@@ -249,3 +278,6 @@ hs.reply = await handshake.send({
         },
-        secretStream: {}
+        secretStream: {},
+        relayThrough: relayThrough
+          ? { publicKey: relayThrough, token: hs.relayToken }
+          : null
       })
@@ -271,2 +303,6 @@ } catch (err) {
 
+    if (relayThrough || remotePayload.relayThrough) {
+      this._relayConnection(hs, relayThrough, remotePayload, h)
+    }
+
     if (remotePayload.firewall === FIREWALL.OPEN || direct) {
@@ -296,3 +332,3 @@ const sock = direct ? socket : this.dht.socket
       hs.prepunching = null
-      hs.rawStream.destroy()
+      if (hs.relayToken === null) hs.rawStream.destroy()
       this._clearLater(hs, id, k)
@@ -464,2 +500,48 @@ }
   }
+
+  _relayConnection (hs, relayThrough, remotePayload, h) {
+    let isInitiator
+    let publicKey
+    let token
+
+    if (relayThrough) {
+      isInitiator = true
+      publicKey = relayThrough
+      token = hs.relayToken
+    } else {
+      isInitiator = false
+      publicKey = remotePayload.relayThrough.publicKey
+      token = remotePayload.relayThrough.token
+    }
+
+    hs.relayToken = token
+    hs.relaySocket = this.dht.connect(publicKey)
+    hs.relayClient = relay.Client.from(hs.relaySocket, { id: hs.relaySocket.publicKey })
+
+    hs.relayClient
+      .pair(isInitiator, token, hs.rawStream)
+      .on('error', () => hs.relaySocket.destroy())
+      .on('data', (remoteId) => {
+        if (hs.rawStream === null) return
+        hs.relayPaired = true
+
+        const {
+          remotePort,
+          remoteHost,
+          socket
+        } = hs.relaySocket.rawStream
+
+        hs.rawStream
+          .on('close', () => hs.relaySocket.destroy())
+          .connect(socket, remoteId, remotePort, remoteHost)
+
+        const rawStream = this.dht._debugStream !== null
+          ? new DebuggingStream(hs.rawStream, this.dht._debugStream)
+          : hs.rawStream
+
+        hs.encryptedSocket = this.createSecretStream(false, rawStream, { handshake: h })
+
+        this.onconnection(hs.encryptedSocket)
+      })
+  }
 }
@@ -466,0 +548,0 @@

package.json

 {
   "name": "hyperdht",
-  "version": "6.6.2",
+  "version": "6.7.0",
   "description": "The DHT powering Hyperswarm",
@@ -20,2 +20,3 @@ "main": "index.js",
     "b4a": "^1.3.1",
+    "blind-relay": "^1.3.0",
     "bogon": "^1.0.0",
@@ -25,3 +26,3 @@ "compact-encoding": "^2.4.1",
     "debugging-stream": "^2.0.0",
-    "dht-rpc": "^6.7.0",
+    "dht-rpc": "^6.9.0",
     "events": "^3.3.0",
@@ -39,3 +40,3 @@ "hypercore-crypto": "^3.3.0",
     "keypear": "^1.1.1",
-    "standard": "^16.0.4"
+    "standard": "^17.1.0"
   },
@@ -42,0 +43,0 @@ "scripts": {

No alert changes

Improved metrics

Total package byte prevSize

126289

increased by8.42%

Number of package files

26

increased by4%

Lines of code

3502

increased by9.2%

Worsened metrics

Dependency count

16

increased by6.67%

Dependency changes

• + Addedblind-relay@^1.3.0

• + Addedbits-to-bytes@1.3.0(transitive)

• + Addedblind-relay@1.3.3(transitive)

• + Addedcompact-encoding-bitfield@1.0.0(transitive)

• + Addedhypertrace@1.4.2(transitive)

• + Addedprotomux@3.10.1(transitive)

• + Addedqueue-tick@1.0.1(transitive)

• Updateddht-rpc@^6.9.0