Add PHT node put/get for search stack

index.js

@@ -16,6 +16,8 @@ const { decode } = require('hypercore-id-encoding')
 const RawStreamSet = require('./lib/raw-stream-set')
 const ConnectionPool = require('./lib/connection-pool')
 const { STREAM_NOT_CONNECTED } = require('./lib/errors')
+const { PrefixHashTree } = require('prefix-hash-tree')
+const { label, isNode } = require('prefix-hash-tree/node')
 
 const DEFAULTS = {
   ...DHT.DEFAULTS,
@@ -60,6 +62,7 @@ class HyperDHT extends DHT {
     this._randomPunchInterval = opts.randomPunchInterval || DEFAULTS.randomPunchInterval // min 20s between random punches...
     this._randomPunches = 0
     this._randomPunchLimit = 1 // set to one for extra safety for now
+    this._experimentalPHT = opts.experimentalPHT === true
 
     this.once('persistent', () => {
       this._persistent = new Persistent(this, persistent)
@@ -389,6 +392,79 @@ class HyperDHT extends DHT {
     return { publicKey: keyPair.publicKey, closestNodes: query.closestNodes, seq, signature }
   }
 
+  async phtNodeGet(target, opts = {}) {
+    if (!this._experimentalPHT) return
+
+    opts = { ...opts, map: mapPHTNode }
+
+    const query = this.query(
+      {
+        target,
+        command: COMMANDS.PHT_NODE_GET,
+        value: null
+      },
+      opts
+    )
+
+    for await (const node of query) {
+      const { publicKey, treeID, phtNode, signature } = node
+
+      const hash = b4a.allocUnsafe(32)
+      sodium.crypto_generichash(hash, b4a.concat([publicKey, treeID]))
+      const indexID = b4a.toString(hash, 'hex')
+      const t = new PrefixHashTree({ indexID })._labelHash(label(phtNode))
+
+      if (
+        isNode(phtNode) &&
+        t.equals(target) &&
+        Persistent.verifyPHTNode(signature, treeID, phtNode, publicKey)
+      )
+        return node
+    }
+
+    return null
+  }
+
+  async phtNodePut(keyPair, treeID, phtNode, opts = {}) {
+    if (!this._experimentalPHT) return
+
+    const publicKey = opts.publicKey || keyPair.publicKey
+
+    const signPHTNode = opts.signPHTNode || Persistent.signPHTNode
+
+    const hash = b4a.allocUnsafe(32)
+    sodium.crypto_generichash(hash, b4a.concat([publicKey, treeID]))
+    const indexID = b4a.toString(hash, 'hex')
+    const target = new PrefixHashTree({ indexID })._labelHash(label(phtNode))
+
+    const signature = opts.signature || (await signPHTNode(phtNode, treeID, keyPair))
+    const connectionKey = opts.connectionKey || b4a.alloc(32)
+
+    const signed = c.encode(m.phtNodePutRequest, {
+      publicKey,
+      treeID,
+      phtNode,
+      signature,
+      connectionKey
+    })
+
+    opts = {
+      ...opts,
+      map: mapPHTNode,
+      commit(reply, dht) {
+        return dht.request(
+          { token: reply.token, target, command: COMMANDS.PHT_NODE_PUT, value: signed },
+          reply.from
+        )
+      }
+    }
+
+    const query = this.query({ target, command: COMMANDS.PHT_NODE_GET, value: null }, opts)
+    await query.finished()
+
+    return { target, indexID, closestNodes: query.closestNodes, signature, connectionKey }
+  }
+
   onrequest(req) {
     switch (req.command) {
       case COMMANDS.PEER_HANDSHAKE: {
@@ -436,6 +512,14 @@ class HyperDHT extends DHT {
         this._persistent.onimmutableget(req)
         return true
       }
+      case COMMANDS.PHT_NODE_PUT: {
+        this._persistent.onphtnodeput(req)
+        return true
+      }
+      case COMMANDS.PHT_NODE_GET: {
+        this._persistent.onphtnodeget(req)
+        return true
+      }
     }
 
     return false
@@ -580,6 +664,27 @@ function mapMutable(node) {
   }
 }
 
+function mapPHTNode(node) {
+  if (!node.value) return null
+
+  try {
+    const p = c.decode(m.phtNodeGetResponse, node.value)
+    const { publicKey, treeID, phtNode, signature } = p
+
+    return {
+      token: node.token,
+      from: node.from,
+      to: node.to,
+      publicKey,
+      treeID,
+      phtNode,
+      signature
+    }
+  } catch {
+    return null
+  }
+}
+
 function noop() {}
 
 function filterNode(node) {
@@ -604,6 +709,7 @@ function defaultCacheOpts(opts) {
     },
     relayAddresses: { maxSize: Math.min(maxSize, 512), maxAge: 0 },
     persistent: {
+      experimentalPHT: opts.experimentalPHT,
       records: { maxSize, maxAge },
       refreshes: { maxSize, maxAge },
       mutables: {
@@ -614,6 +720,10 @@ function defaultCacheOpts(opts) {
         maxSize: (maxSize / 2) | 0,
         maxAge: opts.maxAge || 48 * 60 * 60 * 1000 // 48 hours
       },
+      phtNodes: {
+        maxSize: (maxSize / 2) | 0,
+        maxAge: opts.maxAge || 48 * 60 * 60 * 1000 // 48 hours
+      },
       bumps: { maxSize, maxAge }
     }
   }

lib/constants.js

@@ -10,7 +10,9 @@ const COMMANDS = (exports.COMMANDS = {
   MUTABLE_PUT: 6,
   MUTABLE_GET: 7,
   IMMUTABLE_PUT: 8,
-  IMMUTABLE_GET: 9
+  IMMUTABLE_GET: 9,
+  PHT_NODE_PUT: 10,
+  PHT_NODE_GET: 11
 })
 
 exports.BOOTSTRAP_NODES = global.Pear?.config.dht?.bootstrap || [
@@ -39,19 +41,27 @@ exports.ERROR = {
   SEQ_TOO_LOW: 17
 }
 
-const [NS_ANNOUNCE, NS_UNANNOUNCE, NS_MUTABLE_PUT, NS_PEER_HANDSHAKE, NS_PEER_HOLEPUNCH] =
-  crypto.namespace('hyperswarm/dht', [
-    COMMANDS.ANNOUNCE,
-    COMMANDS.UNANNOUNCE,
-    COMMANDS.MUTABLE_PUT,
-    COMMANDS.PEER_HANDSHAKE,
-    COMMANDS.PEER_HOLEPUNCH
-  ])
+const [
+  NS_ANNOUNCE,
+  NS_UNANNOUNCE,
+  NS_MUTABLE_PUT,
+  NS_PEER_HANDSHAKE,
+  NS_PEER_HOLEPUNCH,
+  NS_PHT_NODE_PUT
+] = crypto.namespace('hyperswarm/dht', [
+  COMMANDS.ANNOUNCE,
+  COMMANDS.UNANNOUNCE,
+  COMMANDS.MUTABLE_PUT,
+  COMMANDS.PEER_HANDSHAKE,
+  COMMANDS.PEER_HOLEPUNCH,
+  COMMANDS.PHT_NODE_PUT
+])
 
 exports.NS = {
   ANNOUNCE: NS_ANNOUNCE,
   UNANNOUNCE: NS_UNANNOUNCE,
   MUTABLE_PUT: NS_MUTABLE_PUT,
   PEER_HANDSHAKE: NS_PEER_HANDSHAKE,
-  PEER_HOLEPUNCH: NS_PEER_HOLEPUNCH
+  PEER_HOLEPUNCH: NS_PEER_HOLEPUNCH,
+  PHT_NODE_PUT: NS_PHT_NODE_PUT
 }

lib/messages.js

@@ -419,3 +419,69 @@ exports.mutableGetResponse = {
     }
   }
 }
+
+exports.phtNodeSignable = {
+  preencode(state, m) {
+    c.buffer.preencode(state, m.treeID)
+    c.json.preencode(state, m.phtNode)
+  },
+  encode(state, m) {
+    c.buffer.encode(state, m.treeID)
+    c.json.encode(state, m.phtNode)
+  },
+  decode(state, m) {
+    return {
+      treeID: c.buffer.decode(state),
+      node: c.json.decode(state)
+    }
+  }
+}
+
+exports.phtNodePutRequest = {
+  preencode(state, m) {
+    c.fixed32.preencode(state, m.publicKey)
+    c.buffer.preencode(state, m.treeID)
+    c.json.preencode(state, m.phtNode)
+    c.fixed64.preencode(state, m.signature)
+    c.fixed32.preencode(state, m.connectionKey)
+  },
+  encode(state, m) {
+    c.fixed32.encode(state, m.publicKey)
+    c.buffer.encode(state, m.treeID)
+    c.json.encode(state, m.phtNode)
+    c.fixed64.encode(state, m.signature)
+    c.fixed32.encode(state, m.connectionKey)
+  },
+  decode(state) {
+    return {
+      publicKey: c.fixed32.decode(state),
+      treeID: c.buffer.decode(state),
+      phtNode: c.json.decode(state),
+      signature: c.fixed64.decode(state),
+      connectionKey: c.fixed32.decode(state)
+    }
+  }
+}
+
+exports.phtNodeGetResponse = {
+  preencode(state, m) {
+    c.fixed32.preencode(state, m.publicKey)
+    c.buffer.preencode(state, m.treeID)
+    c.json.preencode(state, m.phtNode)
+    c.fixed64.preencode(state, m.signature)
+  },
+  encode(state, m) {
+    c.fixed32.encode(state, m.publicKey)
+    c.buffer.encode(state, m.treeID)
+    c.json.encode(state, m.phtNode)
+    c.fixed64.encode(state, m.signature)
+  },
+  decode(state) {
+    return {
+      publicKey: c.fixed32.decode(state),
+      treeID: c.buffer.decode(state),
+      phtNode: c.json.decode(state),
+      signature: c.fixed64.decode(state)
+    }
+  }
+}

lib/persistent.js

@@ -4,6 +4,8 @@ const RecordCache = require('record-cache')
 const Cache = require('xache')
 const b4a = require('b4a')
 const unslab = require('unslab')
+const { PrefixHashTree } = require('prefix-hash-tree')
+const { label, isNode } = require('prefix-hash-tree/node')
 
 const { encodeUnslab } = require('./encode')
 const m = require('./messages')
@@ -21,6 +23,7 @@ module.exports = class Persistent {
     this.refreshes = new Cache(opts.refreshes)
     this.mutables = new Cache(opts.mutables)
     this.immutables = new Cache(opts.immutables)
+    this.phtNodes = opts.experimentalPHT === true ? new Cache(opts.phtNodes) : null
   }
 
   onlookup(req) {
@@ -227,11 +230,49 @@ module.exports = class Persistent {
     req.reply(null)
   }
 
+  onphtnodeget(req) {
+    if (!req.target || !this.phtNodes) return
+
+    const k = b4a.toString(req.target, 'hex')
+    const p = this.phtNodes.get(k)
+
+    if (p === null) return req.reply(null)
+
+    const phtGetResponse = c.encode(m.phtNodeGetResponse, p)
+    req.reply(phtGetResponse)
+  }
+
+  onphtnodeput(req) {
+    if (!req.target || !req.token || !req.value || !this.phtNodes) return
+
+    const unslabbed = unslab(req.value)
+    const p = decode(m.phtNodePutRequest, unslabbed)
+    if (p === null) return
+
+    const { publicKey, treeID, phtNode, signature, connectionKey } = p
+
+    if (!isNode(phtNode)) return
+
+    const hash = b4a.allocUnsafe(32)
+    sodium.crypto_generichash(hash, b4a.concat([publicKey, treeID]))
+    const indexID = b4a.toString(hash, 'hex')
+    const t = new PrefixHashTree({ indexID })._labelHash(label(phtNode))
+
+    if (!t.equals(req.target) || !Persistent.verifyPHTNode(signature, treeID, phtNode, publicKey))
+      return
+
+    const k = b4a.toString(req.target, 'hex')
+    this.phtNodes.set(k, { publicKey, treeID, phtNode, signature })
+
+    req.reply(null)
+  }
+
   destroy() {
     this.records.destroy()
     this.refreshes.destroy()
     this.mutables.destroy()
     this.immutables.destroy()
+    if (this.phtNodes) this.phtNodes.destroy()
   }
 
   static signMutable(seq, value, keyPair) {
@@ -255,6 +296,26 @@ module.exports = class Persistent {
   static signUnannounce(target, token, id, ann, keyPair) {
     return sign(annSignable(target, token, id, ann, NS.UNANNOUNCE), keyPair)
   }
+
+  static signPHTNode(phtNode, treeID, keyPair) {
+    const signable = b4a.allocUnsafe(32 + 32)
+    const hash = signable.subarray(32)
+
+    signable.set(NS.PHT_NODE_PUT, 0)
+
+    sodium.crypto_generichash(hash, c.encode(m.phtNodeSignable, { treeID, phtNode }))
+    return sign(signable, keyPair)
+  }
+
+  static verifyPHTNode(signature, treeID, phtNode, publicKey) {
+    const signable = b4a.allocUnsafe(32 + 32)
+    const hash = signable.subarray(32)
+
+    signable.set(NS.PHT_NODE_PUT, 0)
+
+    sodium.crypto_generichash(hash, c.encode(m.phtNodeSignable, { treeID, phtNode }))
+    return sodium.crypto_sign_verify_detached(signature, signable, publicKey)
+  }
 }
 
 function verifyMutable(signature, seq, value, publicKey) {

package.json

@@ -36,6 +36,7 @@
     "hypercore-id-encoding": "^1.2.0",
     "noise-curve-ed": "^2.0.0",
     "noise-handshake": "^4.0.0",
+    "prefix-hash-tree": "^0.0.2",
     "record-cache": "^1.1.1",
     "safety-catch": "^1.0.1",
     "signal-promise": "^1.0.3",

test/helpers/index.js

@@ -12,8 +12,8 @@ async function toArray(iterable) {
   return result
 }
 
-async function swarm(t, n = 32, bootstrap = []) {
-  return createTestnet(n, { bootstrap, teardown: t.teardown })
+async function swarm(t, n = 32, bootstrap = [], dhtOpts = {}) {
+  return createTestnet(n, { bootstrap, teardown: t.teardown }, dhtOpts)
 }
 
 async function* spawnFixture(t, args) {