fix: dht validate if receiving stream

README.md

@@ -259,8 +259,8 @@ const ipfs = ipfsClient({
   - [`ipfs.bitswap.unwant(cid)`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/BITSWAP.md#bitswapunwant)
 
 - [dht](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md)
-  - [`ipfs.dht.findpeer(peerId, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md#dhtfindpeer)
-  - [`ipfs.dht.findprovs(hash, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md#dhtfindprovs)
+  - [`ipfs.dht.findPeer(peerId, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md#dhtfindpeer)
+  - [`ipfs.dht.findProvs(hash, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md#dhtfindprovs)
   - [`ipfs.dht.get(key, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md#dhtget)
   - [`ipfs.dht.provide(cid, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md#dhtprovide)
   - [`ipfs.dht.put(key, value, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DHT.md#dhtput)

package.json

@@ -36,6 +36,7 @@
     "debug": "^4.1.0",
     "detect-node": "^2.0.4",
     "end-of-stream": "^1.4.1",
+    "err-code": "^1.1.2",
     "flatmap": "0.0.3",
     "glob": "^7.1.3",
     "ipfs-block": "~0.8.0",
@@ -47,14 +48,14 @@
     "is-stream": "^1.1.0",
     "libp2p-crypto": "~0.14.0",
     "lodash": "^4.17.11",
-    "lru-cache": "^4.1.3",
-    "multiaddr": "^5.0.2",
+    "lru-cache": "^5.1.1",
+    "multiaddr": "^6.0.0",
     "multibase": "~0.6.0",
     "multihashes": "~0.4.14",
     "ndjson": "^1.5.0",
     "once": "^1.4.0",
     "peer-id": "~0.12.0",
-    "peer-info": "~0.14.1",
+    "peer-info": "~0.15.0",
     "promisify-es6": "^1.0.3",
     "pull-defer": "~0.2.3",
     "pull-pushable": "^2.2.0",
@@ -66,7 +67,7 @@
     "stream-to-pull-stream": "^1.7.2",
     "streamifier": "~0.1.1",
     "tar-stream": "^1.6.2",
-    "through2": "^2.0.3"
+    "through2": "^3.0.0"
   },
   "engines": {
     "node": ">=8.0.0",
@@ -85,7 +86,7 @@
     "eslint-plugin-react": "^7.11.1",
     "go-ipfs-dep": "~0.4.18",
     "gulp": "^3.9.1",
-    "interface-ipfs-core": "~0.90.0",
+    "interface-ipfs-core": "~0.91.0",
     "ipfsd-ctl": "~0.40.0",
     "nock": "^10.0.2",
     "pull-stream": "^3.6.9",

src/dht/findpeer.js

@@ -1,7 +1,12 @@
 'use strict'
 
 const promisify = require('promisify-es6')
-const streamToValue = require('../utils/stream-to-value')
+const streamToValueWithTransformer = require('../utils/stream-to-value-with-transformer')
+
+const multiaddr = require('multiaddr')
+const PeerId = require('peer-id')
+const PeerInfo = require('peer-info')
+const errcode = require('err-code')
 
 module.exports = (send) => {
   return promisify((peerId, opts, callback) => {
@@ -17,10 +22,41 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    const handleResult = (res, callback) => {
+      // Inconsistent return values in the browser
+      if (Array.isArray(res)) {
+        res = res[0]
+      }
+
+      // Type 2 keys
+      if (res.Type !== 2) {
+        const errMsg = `key was not found (type 2)`
+
+        return callback(errcode(new Error(errMsg), 'ERR_KEY_TYPE_2_NOT_FOUND'))
+      }
+
+      const responseReceived = res.Responses[0]
+      const peerInfo = new PeerInfo(PeerId.createFromB58String(responseReceived.ID))
+
+      responseReceived.Addrs.forEach((addr) => {
+        const ma = multiaddr(addr)
+
+        peerInfo.multiaddrs.add(ma)
+      })
+
+      callback(null, peerInfo)
+    }
+
+    send({
       path: 'dht/findpeer',
       args: peerId,
       qs: opts
-    }, streamToValue, callback)
+    }, (err, result) => {
+      if (err) {
+        return callback(err)
+      }
+
+      streamToValueWithTransformer(result, handleResult, callback)
+    })
   })
 }

src/dht/findprovs.js

@@ -1,7 +1,12 @@
 'use strict'
 
 const promisify = require('promisify-es6')
-const streamToValue = require('../utils/stream-to-value')
+const streamToValueWithTransformer = require('../utils/stream-to-value-with-transformer')
+
+const multiaddr = require('multiaddr')
+const PeerId = require('peer-id')
+const PeerInfo = require('peer-info')
+const errcode = require('err-code')
 
 module.exports = (send) => {
   return promisify((cid, opts, callback) => {
@@ -17,10 +22,44 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    const handleResult = (res, callback) => {
+      // Inconsistent return values in the browser vs node
+      if (Array.isArray(res)) {
+        res = res[0]
+      }
+
+      // Type 4 keys
+      if (res.Type !== 4) {
+        const errMsg = `key was not found (type 4)`
+
+        return callback(errcode(new Error(errMsg), 'ERR_KEY_TYPE_4_NOT_FOUND'))
+      }
+
+      const responses = res.Responses.map((r) => {
+        const peerInfo = new PeerInfo(PeerId.createFromB58String(r.ID))
+
+        r.Addrs.forEach((addr) => {
+          const ma = multiaddr(addr)
+
+          peerInfo.multiaddrs.add(ma)
+        })
+
+        return peerInfo
+      })
+
+      callback(null, responses)
+    }
+
+    send({
       path: 'dht/findprovs',
       args: cid,
       qs: opts
-    }, streamToValue, callback)
+    }, (err, result) => {
+      if (err) {
+        return callback(err)
+      }
+
+      streamToValueWithTransformer(result, handleResult, callback)
+    })
   })
 }

src/dht/index.js

@@ -8,8 +8,8 @@ module.exports = (arg) => {
   return {
     get: require('./get')(send),
     put: require('./put')(send),
-    findprovs: require('./findprovs')(send),
-    findpeer: require('./findpeer')(send),
+    findProvs: require('./findprovs')(send),
+    findPeer: require('./findpeer')(send),
     provide: require('./provide')(send),
     // find closest peerId to given peerId
     query: require('./query')(send)

src/dht/query.js

@@ -1,7 +1,10 @@
 'use strict'
 
 const promisify = require('promisify-es6')
-const streamToValue = require('../utils/stream-to-value')
+const streamToValueWithTransformer = require('../utils/stream-to-value-with-transformer')
+
+const PeerId = require('peer-id')
+const PeerInfo = require('peer-info')
 
 module.exports = (send) => {
   return promisify((peerId, opts, callback) => {
@@ -17,10 +20,22 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    const handleResult = (res, callback) => {
+      const peerIds = res.map((r) => (new PeerInfo(PeerId.createFromB58String(r.ID))))
+
+      callback(null, peerIds)
+    }
+
+    send({
       path: 'dht/query',
       args: peerId,
       qs: opts
-    }, streamToValue, callback)
+    }, (err, result) => {
+      if (err) {
+        return callback(err)
+      }
+
+      streamToValueWithTransformer(result, handleResult, callback)
+    })
   })
 }

src/object/data.js

@@ -8,7 +8,7 @@ const lruOptions = {
   max: 128
 }
 
-const cache = LRU(lruOptions)
+const cache = new LRU(lruOptions)
 
 module.exports = (send) => {
   return promisify((cid, options, callback) => {

src/object/get.js

@@ -11,7 +11,7 @@ const lruOptions = {
   max: 128
 }
 
-const cache = LRU(lruOptions)
+const cache = new LRU(lruOptions)
 
 module.exports = (send) => {
   return promisify((cid, options, callback) => {

src/object/links.js

@@ -9,7 +9,7 @@ const lruOptions = {
   max: 128
 }
 
-const cache = LRU(lruOptions)
+const cache = new LRU(lruOptions)
 
 module.exports = (send) => {
   return promisify((multihash, options, callback) => {

src/utils/stream-to-value-with-transformer.js

@@ -0,0 +1,18 @@
+'use strict'
+
+const streamToValue = require('./stream-to-value')
+
+function streamToValueWithTransformer (response, transformer, callback) {
+  if (typeof response.pipe === 'function') {
+    streamToValue(response, (err, res) => {
+      if (err) {
+        return callback(err)
+      }
+      transformer(res, callback)
+    })
+  } else {
+    transformer(response, callback)
+  }
+}
+
+module.exports = streamToValueWithTransformer

test/sub-modules.spec.js

@@ -49,8 +49,8 @@ describe('submodules', () => {
 
     expect(dht.get).to.be.a('function')
     expect(dht.put).to.be.a('function')
-    expect(dht.findprovs).to.be.a('function')
-    expect(dht.findpeer).to.be.a('function')
+    expect(dht.findProvs).to.be.a('function')
+    expect(dht.findPeer).to.be.a('function')
     expect(dht.provide).to.be.a('function')
     expect(dht.query).to.be.a('function')
   })