Async Crypto Endeavour (+benchmarks!) (#19)

* feat: add benchmarks and integrate with new libp2p-crypto

* start removing node-forge

* less forge

* cleanup and fixes

* benchmarked and tested

* remove unused .aegir

* Replace lib multihashing with multihashing-async

* cleanup package.json

* Using string instead of buffer for proto file

* chore: fix benchmarks

* docs(readme): update README URLs based on HTTP redirects

* ready for next aegir

* update aegir

* ready

* chore: update deps

.aegir.js

@@ -1,16 +0,0 @@
-'use strict'
-
-const path = require('path')
-
-module.exports = {
-  webpack: {
-    resolve: {
-      alias: {
-        'node-forge': path.resolve(
-          path.dirname(require.resolve('libp2p-crypto')),
-          '../vendor/forge.bundle.js'
-        )
-      }
-    }
-  }
-}

.gitignore

@@ -31,5 +31,4 @@ build
 # https://www.npmjs.org/doc/misc/npm-faq.html#should-i-check-my-node_modules-folder-into-git
 node_modules
 
-lib
 dist

.travis.yml

@@ -1,9 +1,15 @@
 sudo: false
 language: node_js
-node_js:
+matrix:
-  - 4
+  include:
-  - 5
+  - node_js: 4
-  - stable
+    env: CXX=g++-4.8
+  - node_js: 6
+    env:
+      - SAUCE=true
+      - CXX=g++-4.8
+  - node_js: stable
+    env: CXX=g++-4.8
 
 # Make sure we have new NPM.
 before_install:
@@ -14,7 +20,6 @@ script:
   - npm test
   - npm run coverage
 
-
 before_script:
   - export DISPLAY=:99.0
   - sh -e /etc/init.d/xvfb start
@@ -22,9 +27,6 @@ before_script:
 after_success:
   - npm run coverage-publish
 
-env:
-  - CXX=g++-4.8
-
 addons:
   firefox: 'latest'
   apt:

README.md

@@ -8,6 +8,10 @@
 [![Travis CI](https://travis-ci.org/libp2p/js-libp2p-secio.svg?branch=master)](https://travis-ci.org/libp2p/js-libp2p-secio)
 [![Circle CI](https://circleci.com/gh/libp2p/js-libp2p-secio.svg?style=svg)](https://circleci.com/gh/libp2p/js-libp2p-secio)
 [![Dependency Status](https://david-dm.org/libp2p/js-libp2p-secio.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-secio) [![js-standard-style](https://img.shields.io/badge/code%20style-standard-brightgreen.svg?style=flat-square)](https://github.com/feross/standard)
+![](https://img.shields.io/badge/npm-%3E%3D3.0.0-orange.svg?style=flat-square)
+![](https://img.shields.io/badge/Node.js-%3E%3D4.0.0-orange.svg?style=flat-square)
+
+[![Sauce Test Status](https://saucelabs.com/browser-matrix/libp2p-js-secio.svg)](https://saucelabs.com/u/libp2p-js-secio)
 
 > Secio implementation in JavaScript
 

benchmarks/send.js

@@ -0,0 +1,87 @@
+'use strict'
+
+const Benchmark = require('benchmark')
+const pull = require('pull-stream')
+const pair = require('pull-pair/duplex')
+const PeerId = require('peer-id')
+const crypto = require('libp2p-crypto')
+
+const secio = require('../src')
+
+const suite = new Benchmark.Suite('secio')
+const ids = []
+
+suite.add('createKey', function (d) {
+  crypto.generateKeyPair('RSA', 2048, (err, key) => {
+    if (err) {
+      throw err
+    }
+    PeerId.createFromPrivKey(key.bytes, (err, id) => {
+      if (err) {
+        throw err
+      }
+      ids.push(id)
+      d.resolve()
+    })
+  })
+}, {
+  defer: true
+})
+.add('send', function (deferred) {
+  const p = pair()
+
+  createSession(p[0], (err, local) => {
+    if (err) {
+      throw err
+    }
+    createSession(p[1], (err, remote) => {
+      if (err) {
+        throw err
+      }
+      sendMessages(local, remote)
+    })
+  })
+
+  function sendMessages (local, remote) {
+    pull(
+      pull.infinite(),
+      pull.take(100),
+      pull.map((val) => Buffer(val.toString())),
+      local
+    )
+
+    pull(
+      remote,
+      pull.take(100),
+      pull.collect((err, chunks) => {
+        if (err) throw err
+        if (chunks.length !== 100) throw new Error('Did not receive enough chunks')
+        deferred.resolve()
+      })
+    )
+  }
+}, {
+  defer: true
+})
+.on('cycle', (event) => {
+  console.log(String(event.target))
+})
+// run async
+.run({
+  async: true
+})
+
+function createSession (insecure, cb) {
+  crypto.generateKeyPair('RSA', 2048, (err, key) => {
+    if (err) {
+      return cb(err)
+    }
+    PeerId.createFromPrivKey(key.bytes, (err, id) => {
+      if (err) {
+        return cb(err)
+      }
+
+      cb(null, secio.encrypt(id, key, insecure))
+    })
+  })
+}

package.json

@@ -2,8 +2,7 @@
   "name": "libp2p-secio",
   "version": "0.5.0",
   "description": "Secio implementation in JavaScript",
-  "main": "lib/index.js",
+  "main": "src/index.js",
-  "jsnext:main": "src/index.js",
   "scripts": {
     "lint": "aegir-lint",
     "build": "aegir-build",
@@ -14,7 +13,8 @@
     "release-minor": "aegir-release --type minor",
     "release-major": "aegir-release --type major",
     "coverage": "aegir-coverage",
-    "coverage-publish": "aegir-coverage publish"
+    "coverage-publish": "aegir-coverage publish",
+    "bench": "node benchmarks/send.js"
   },
   "keywords": [
     "IPFS",
@@ -22,37 +22,36 @@
     "crypto",
     "rsa"
   ],
-  "author": "Friedel Ziegelmayer <dignifiedqurie@gmail.com>",
+  "author": "Friedel Ziegelmayer <dignifiedquire@gmail.com>",
   "license": "MIT",
   "dependencies": {
+    "async": "^2.1.2",
     "debug": "^2.2.0",
-    "interface-connection": "^0.2.1",
+    "interface-connection": "^0.3.0",
-    "libp2p-crypto": "^0.6.1",
+    "libp2p-crypto": "^0.7.0",
-    "multihashing": "^0.2.1",
+    "multihashing-async": "^0.2.0",
-    "node-forge": "^0.6.42",
+    "peer-id": "^0.8.0",
-    "peer-id": "^0.7.0",
     "protocol-buffers": "^3.1.6",
     "pull-defer": "^0.2.2",
     "pull-handshake": "^1.1.4",
     "pull-length-prefixed": "^1.2.0",
-    "pull-stream": "^3.4.3",
+    "pull-stream": "^3.5.0"
-    "pull-through": "^1.0.18",
-    "run-series": "^1.1.4"
   },
   "devDependencies": {
-    "aegir": "^8.0.0",
+    "aegir": "^9.0.1",
+    "benchmark": "^2.1.2",
     "chai": "^3.5.0",
-    "multistream-select": "^0.11.0",
+    "multistream-select": "^0.11.1",
     "pre-commit": "^1.1.3",
     "pull-pair": "^1.1.0",
-    "run-parallel": "^1.1.6"
+    "pull-pushable": "^2.0.1"
   },
   "pre-commit": [
     "lint",
     "test"
   ],
   "engines": {
-    "node": "^4.0.0"
+    "node": ">=4.0.0"
   },
   "repository": {
     "type": "git",

src/etm.js

@@ -1,73 +1,67 @@
 'use strict'
 
-const through = require('pull-through')
 const pull = require('pull-stream')
 const lp = require('pull-length-prefixed')
 
-const toForgeBuffer = require('./support').toForgeBuffer
-
 const lpOpts = {
   fixed: true,
   bytes: 4
 }
 
 exports.createBoxStream = (cipher, mac) => {
-  const pt = through(function (chunk) {
-    cipher.update(toForgeBuffer(chunk))
-
-    if (cipher.output.length() > 0) {
-      const data = new Buffer(cipher.output.getBytes(), 'binary')
-      mac.update(data.toString('binary'))
-      const macBuffer = new Buffer(mac.digest().getBytes(), 'binary')
-
-      this.queue(Buffer.concat([data, macBuffer]))
-      // reset hmac
-      mac.start(null, null)
-    }
-  })
-
   return pull(
-    pt,
+    pull.asyncMap((chunk, cb) => {
+      cipher.encrypt(chunk, (err, data) => {
+        if (err) {
+          return cb(err)
+        }
+
+        mac.digest(data, (err, digest) => {
+          if (err) {
+            return cb(err)
+          }
+
+          cb(null, Buffer.concat([data, digest]))
+        })
+      })
+    }),
     lp.encode(lpOpts)
   )
 }
 
 exports.createUnboxStream = (decipher, mac) => {
-  const pt = through(function (chunk) {
+  return pull(
+    lp.decode(lpOpts),
+    pull.asyncMap((chunk, cb) => {
       const l = chunk.length
-    const macSize = mac.getMac().length()
+      const macSize = mac.length
 
       if (l < macSize) {
-      return this.emit('error', new Error(`buffer (${l}) shorter than MAC size (${macSize})`))
+        return cb(new Error(`buffer (${l}) shorter than MAC size (${macSize})`))
       }
 
       const mark = l - macSize
       const data = chunk.slice(0, mark)
       const macd = chunk.slice(mark)
 
-    // Clear out any previous data
+      mac.digest(data, (err, expected) => {
-    mac.start(null, null)
+        if (err) {
+          return cb(err)
+        }
 
-    mac.update(data.toString('binary'))
-    const expected = new Buffer(mac.getMac().getBytes(), 'binary')
-
-    // reset hmac
-    mac.start(null, null)
         if (!macd.equals(expected)) {
-      return this.emit('error', new Error(`MAC Invalid: ${macd.toString('hex')} != ${expected.toString('hex')}`))
+          return cb(new Error(`MAC Invalid: ${macd.toString('hex')} != ${expected.toString('hex')}`))
         }
 
         // all good, decrypt
-    decipher.update(toForgeBuffer(data))
+        decipher.decrypt(data, (err, decrypted) => {
-
+          if (err) {
-    if (decipher.output.length() > 0) {
+            return cb(err)
-      const data = new Buffer(decipher.output.getBytes(), 'binary')
-      this.queue(data)
           }
-  })
 
-  return pull(
+          cb(null, decrypted)
-    lp.decode(lpOpts),
+        })
-    pt
+      })
+    })
   )
 }

src/handshake/crypto.js

@@ -1,15 +1,15 @@
 'use strict'
 
 const protobuf = require('protocol-buffers')
-const path = require('path')
-const fs = require('fs')
 const PeerId = require('peer-id')
 const crypto = require('libp2p-crypto')
+const parallel = require('async/parallel')
+const waterfall = require('async/waterfall')
 const debug = require('debug')
 const log = debug('libp2p:secio')
 log.error = debug('libp2p:secio:error')
 
-const pbm = protobuf(fs.readFileSync(path.join(__dirname, 'secio.proto')))
+const pbm = protobuf(require('./secio.proto'))
 
 const support = require('../support')
 
@@ -29,8 +29,12 @@ exports.createProposal = (state) => {
   return state.proposalEncoded.out
 }
 
-exports.createExchange = (state) => {
+exports.createExchange = (state, callback) => {
-  const res = crypto.generateEphemeralKeyPair(state.protocols.local.curveT)
+  crypto.generateEphemeralKeyPair(state.protocols.local.curveT, (err, res) => {
+    if (err) {
+      return callback(err)
+    }
+
     state.ephemeralKey.local = res.key
     state.shared.generate = res.genSharedKey
 
@@ -41,15 +45,22 @@ exports.createExchange = (state) => {
       state.ephemeralKey.local
     ])
 
-  state.exchange.out = {
+    state.key.local.sign(selectionOut, (err, sig) => {
-    epubkey: state.ephemeralKey.local,
+      if (err) {
-    signature: new Buffer(state.key.local.sign(selectionOut), 'binary')
+        return callback(err)
       }
 
-  return pbm.Exchange.encode(state.exchange.out)
+      state.exchange.out = {
+        epubkey: state.ephemeralKey.local,
+        signature: sig
+      }
+
+      callback(null, pbm.Exchange.encode(state.exchange.out))
+    })
+  })
 }
 
-exports.identify = (state, msg) => {
+exports.identify = (state, msg, callback) => {
   log('1.1 identify')
 
   state.proposalEncoded.in = msg
@@ -57,12 +68,19 @@ exports.identify = (state, msg) => {
   const pubkey = state.proposal.in.pubkey
 
   state.key.remote = crypto.unmarshalPublicKey(pubkey)
-  state.id.remote = PeerId.createFromPubKey(pubkey.toString('base64'))
+  PeerId.createFromPubKey(pubkey.toString('base64'), (err, remoteId) => {
+    if (err) {
+      return callback(err)
+    }
+
+    state.id.remote = remoteId
 
     log('1.1 identify - %s - identified remote peer as %s', state.id.local.toB58String(), state.id.remote.toB58String())
+    callback()
+  })
 }
 
-exports.selectProtocols = (state) => {
+exports.selectProtocols = (state, callback) => {
   log('1.2 selection')
 
   const local = {
@@ -81,7 +99,10 @@ exports.selectProtocols = (state) => {
     nonce: state.proposal.in.rand
   }
 
-  let selected = support.selectBest(local, remote)
+  support.selectBest(local, remote, (err, selected) => {
+    if (err) {
+      return callback(err)
+    }
     // we use the same params for both directions (must choose same curve)
     // WARNING: if they dont SelectBest the same way, this won't work...
     state.protocols.remote = {
@@ -97,9 +118,11 @@ exports.selectProtocols = (state) => {
       cipherT: selected.cipherT,
       hashT: selected.hashT
     }
+    callback()
+  })
 }
 
-exports.verify = (state, msg) => {
+exports.verify = (state, msg, callback) => {
   log('2.1. verify')
 
   state.exchange.in = pbm.Exchange.decode(msg)
@@ -111,26 +134,36 @@ exports.verify = (state, msg) => {
     state.ephemeralKey.remote
   ])
 
-  const sigOk = state.key.remote.verify(selectionIn, state.exchange.in.signature)
+  state.key.remote.verify(selectionIn, state.exchange.in.signature, (err, sigOk) => {
+    if (err) {
+      return callback(err)
+    }
 
     if (!sigOk) {
-    throw new Error('Bad signature')
+      return callback(new Error('Bad signature'))
     }
 
     log('2.1. verify - signature verified')
+    callback()
+  })
 }
 
-exports.generateKeys = (state) => {
+exports.generateKeys = (state, callback) => {
   log('2.2. keys')
 
-  state.shared.secret = state.shared.generate(state.exchange.in.epubkey)
+  waterfall([
+    (cb) => state.shared.generate(state.exchange.in.epubkey, cb),
+    (secret, cb) => {
+      state.shared.secret = secret
 
-  const keys = crypto.keyStretcher(
+      crypto.keyStretcher(
         state.protocols.local.cipherT,
         state.protocols.local.hashT,
-    state.shared.secret
+        state.shared.secret,
+        cb
       )
-
+    },
+    (keys, cb) => {
       // use random nonces to decide order.
       if (state.protocols.local.order > 0) {
         state.protocols.local.keys = keys.k1
@@ -141,13 +174,17 @@ exports.generateKeys = (state) => {
         state.protocols.remote.keys = keys.k1
       } else {
         // we should've bailed before state. but if not, bail here.
-    throw new Error('you are trying to talk to yourself')
+        return cb(new Error('you are trying to talk to yourself'))
       }
 
       log('2.3. mac + cipher')
 
-  support.makeMacAndCipher(state.protocols.local)
+      parallel([
-  support.makeMacAndCipher(state.protocols.remote)
+        (cb) => support.makeMacAndCipher(state.protocols.local, cb),
+        (cb) => support.makeMacAndCipher(state.protocols.remote, cb)
+      ], cb)
+    }
+  ], callback)
 }
 
 exports.verifyNonce = (state, n2) => {

src/handshake/exchange.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const debug = require('debug')
+const waterfall = require('async/waterfall')
 
 const support = require('../support')
 const crypto = require('./crypto')
@@ -14,18 +15,19 @@ module.exports = function exchange (state, cb) {
   log('2. exchange - start')
 
   log('2. exchange - writing exchange')
-  support.write(state, crypto.createExchange(state))
+  waterfall([
-  support.read(state.shake, (err, msg) => {
+    (cb) => crypto.createExchange(state, cb),
-    if (err) {
+    (ex, cb) => {
-      return cb(err)
+      support.write(state, ex)
-    }
+      support.read(state.shake, cb)
-
+    },
+    (msg, cb) => {
       log('2. exchange - reading exchange')
-
+      crypto.verify(state, msg, cb)
-    try {
+    },
-      crypto.verify(state, msg)
+    (cb) => crypto.generateKeys(state, cb)
-      crypto.generateKeys(state)
+  ], (err) => {
-    } catch (err) {
+    if (err) {
       return cb(err)
     }
 

src/handshake/finish.js

@@ -17,7 +17,11 @@ module.exports = function finish (state, cb) {
 
   const proto = state.protocols
   const stream = state.shake.rest()
-  const shake = handshake({timeout: state.timeout})
+  const shake = handshake({timeout: state.timeout}, (err) => {
+    if (err) {
+      throw err
+    }
+  })
 
   pull(
     stream,

src/handshake/index.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const series = require('run-series')
+const series = require('async/series')
 
 const propose = require('./propose')
 const exchange = require('./exchange')

src/handshake/propose.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const debug = require('debug')
+const waterfall = require('async/waterfall')
 
 const support = require('../support')
 const crypto = require('./crypto')
@@ -15,22 +16,20 @@ module.exports = function propose (state, cb) {
 
   log('1. propose - writing proposal')
   support.write(state, crypto.createProposal(state))
-  support.read(state.shake, (err, msg) => {
+
+  waterfall([
+    (cb) => support.read(state.shake, cb),
+    (msg, cb) => {
+      log('1. propose - reading proposal', msg)
+      crypto.identify(state, msg, cb)
+    },
+    (cb) => crypto.selectProtocols(state, cb)
+  ], (err) => {
     if (err) {
       return cb(err)
     }
 
-    log('1. propose - reading proposal', msg)
-
-    try {
-      crypto.identify(state, msg)
-      crypto.selectProtocols(state)
-    } catch (err) {
-      return cb(err)
-    }
-
     log('1. propose - finish')
-
     cb()
   })
 }

src/handshake/secio.proto

@@ -1,12 +0,0 @@
-message Propose {
-	optional bytes rand = 1;
-	optional bytes pubkey = 2;
-	optional string exchanges = 3;
-	optional string ciphers = 4;
-	optional string hashes = 5;
-}
-
-message Exchange {
-	optional bytes epubkey = 1;
-	optional bytes signature = 2;
-}

src/handshake/secio.proto.js

@@ -0,0 +1,14 @@
+'use strict'
+
+module.exports = `message Propose {
+  optional bytes rand = 1;
+  optional bytes pubkey = 2;
+  optional string exchanges = 3;
+  optional string ciphers = 4;
+  optional string hashes = 5;
+}
+
+message Exchange {
+  optional bytes epubkey = 1;
+  optional bytes signature = 2;
+}`

src/index.js

@@ -8,7 +8,7 @@ const State = require('./state')
 
 module.exports = {
   tag: '/secio/1.0.0',
-  encrypt (local, key, insecure) {
+  encrypt (local, key, insecure, callback) {
     if (!local) {
       throw new Error('no local id provided')
     }
@@ -21,7 +21,13 @@ module.exports = {
       throw new Error('no insecure stream provided')
     }
 
-    const state = new State(local, key)
+    if (!callback) {
+      callback = (err) => {
+        throw err
+      }
+    }
+
+    const state = new State(local, key, callback)
 
     pull(
       insecure,

src/state.js

@@ -5,6 +5,11 @@ const deferred = require('pull-defer')
 
 class State {
   constructor (id, key, timeout, cb) {
+    if (typeof timeout === 'function') {
+      cb = timeout
+      timeout = undefined
+    }
+
     this.setup()
     this.id.local = id
     this.key.local = key

src/support.js

@@ -1,9 +1,10 @@
 'use strict'
 
-const mh = require('multihashing')
+const mh = require('multihashing-async')
-const forge = require('node-forge')
 const lp = require('pull-length-prefixed')
 const pull = require('pull-stream')
+const crypto = require('libp2p-crypto')
+const parallel = require('async/parallel')
 
 exports.exchanges = [
   'P-256',
@@ -47,78 +48,73 @@ exports.theBest = (order, p1, p2) => {
   throw new Error('No algorithms in common!')
 }
 
-exports.makeMacAndCipher = (target) => {
+exports.makeMacAndCipher = (target, callback) => {
-  target.mac = makeMac(target.hashT, target.keys.macKey)
+  parallel([
-  target.cipher = makeCipher(target.cipherT, target.keys.iv, target.keys.cipherKey)
+    (cb) => makeMac(target.hashT, target.keys.macKey, cb),
-}
+    (cb) => makeCipher(target.cipherT, target.keys.iv, target.keys.cipherKey, cb)
-
+  ], (err, macAndCipher) => {
-const hashMap = {
+    if (err) {
-  SHA1: 'sha1',
+      return callback(err)
-  SHA256: 'sha256',
-  // workaround for https://github.com/digitalbazaar/forge/issues/401
-  SHA512: forge.md.sha512.create()
-}
-
-const toForgeBuffer = exports.toForgeBuffer = (buf) => (
-  forge.util.createBuffer(buf.toString('binary'))
-)
-
-function makeMac (hashType, key) {
-  const hash = hashMap[hashType]
-
-  if (!hash) {
-    throw new Error(`unsupported hash type: ${hashType}`)
     }
 
-  const mac = forge.hmac.create()
+    target.mac = macAndCipher[0]
-  mac.start(hash, toForgeBuffer(key))
+    target.cipher = macAndCipher[1]
-  return mac
+    callback()
+  })
 }
 
-function makeCipher (cipherType, iv, key) {
+function makeMac (hash, key, callback) {
+  crypto.hmac.create(hash, key, callback)
+}
+
+function makeCipher (cipherType, iv, key, callback) {
   if (cipherType === 'AES-128' || cipherType === 'AES-256') {
-    // aes in counter (CTR) mode because that is what
+    return crypto.aes.create(key, iv, callback)
-    // is used in go (cipher.NewCTR)
-    const cipher = forge.cipher.createCipher('AES-CTR', toForgeBuffer(key))
-    cipher.start({iv: toForgeBuffer(iv)})
-    return cipher
   }
 
-  // TODO: Blowfish is not supported in node-forge, figure out if
+  // TODO: figure out if Blowfish is needed and if so find a library for it.
-  // it's needed and if so find a library for it.
+  callback(new Error(`unrecognized cipher type: ${cipherType}`))
-
-  throw new Error(`unrecognized cipher type: ${cipherType}`)
 }
 
 exports.randomBytes = (nonceSize) => {
-  return new Buffer(forge.random.getBytesSync(nonceSize), 'binary')
+  return Buffer.from(crypto.webcrypto.getRandomValues(new Uint8Array(nonceSize)))
 }
 
-exports.selectBest = (local, remote) => {
+exports.selectBest = (local, remote, cb) => {
-  const oh1 = exports.digest(Buffer.concat([
+  exports.digest(Buffer.concat([
     remote.pubKeyBytes,
     local.nonce
-  ]))
+  ]), (err, oh1) => {
-  const oh2 = exports.digest(Buffer.concat([
+    if (err) {
+      return cb(err)
+    }
+
+    exports.digest(Buffer.concat([
       local.pubKeyBytes,
       remote.nonce
-  ]))
+    ]), (err, oh2) => {
+      if (err) {
+        return cb(err)
+      }
+
       const order = Buffer.compare(oh1, oh2)
 
       if (order === 0) {
-    throw new Error('you are trying to talk to yourself')
+        cb(new Error('you are trying to talk to yourself'))
       }
 
-  return {
+      cb(null, {
         curveT: exports.theBest(order, local.exchanges, remote.exchanges),
         cipherT: exports.theBest(order, local.ciphers, remote.ciphers),
         hashT: exports.theBest(order, local.hashes, remote.hashes),
         order
-  }
+      })
+    })
+  })
 }
 
-exports.digest = (buf) => {
+exports.digest = (buf, cb) => {
-  return mh.digest(buf, 'sha2-256', buf.length)
+  mh.digest(buf, 'sha2-256', buf.length, cb)
 }
 
 exports.write = function write (state, msg, cb) {

test/index.spec.js

@@ -1,3 +1,4 @@
+/* eslint max-nested-callbacks: ["error", 8] */
 /* eslint-env mocha */
 'use strict'
 
@@ -5,8 +6,8 @@ const pair = require('pull-pair/duplex')
 const expect = require('chai').expect
 const PeerId = require('peer-id')
 const crypto = require('libp2p-crypto')
-const parallel = require('run-parallel')
+const parallel = require('async/parallel')
-const series = require('run-series')
+const series = require('async/series')
 const ms = require('multistream-select')
 const pull = require('pull-stream')
 const Listener = ms.Listener
@@ -21,12 +22,18 @@ describe('libp2p-secio', () => {
 
   it('upgrades a connection', (done) => {
     const p = pair()
+    createSession(p[0], (err, local) => {
+      if (err) {
+        return done(err)
+      }
 
-    const local = createSession(p[0])
+      createSession(p[1], (err, remote) => {
-    const remote = createSession(p[1])
+        if (err) {
+          return done(err)
+        }
 
         pull(
-      pull.values(['hello world']),
+          pull.values([new Buffer('hello world')]),
           local
         )
 
@@ -39,14 +46,15 @@ describe('libp2p-secio', () => {
           })
         )
       })
+    })
+  })
 
   it('works over multistream', (done) => {
     const p = pair()
 
     const listener = new Listener()
     const dialer = new Dialer()
-    let local
+
-    let remote
     series([
       (cb) => parallel([
         (cb) => listener.handle(p[0], cb),
@@ -54,7 +62,10 @@ describe('libp2p-secio', () => {
       ], cb),
       (cb) => {
         listener.addHandler('/banana/1.0.0', (conn) => {
-          local = createSession(conn)
+          createSession(conn, (err, local) => {
+            if (err) {
+              return done(err)
+            }
             pull(
               local,
               pull.collect((err, chunks) => {
@@ -64,25 +75,45 @@ describe('libp2p-secio', () => {
               })
             )
           })
+        })
         cb()
       },
       (cb) => dialer.select('/banana/1.0.0', (err, conn) => {
-        remote = createSession(conn)
+        if (err) {
+          return cb(err)
+        }
+
+        createSession(conn, (err, remote) => {
+          if (err) {
+            return cb(err)
+          }
           pull(
-          pull.values(['hello world']),
+            pull.values([new Buffer('hello world')]),
             remote
           )
-        cb(err)
+          cb()
+        })
       })
     ], (err) => {
-      if (err) throw err
+      if (err) {
+        throw err
+      }
     })
   })
 })
 
-function createSession (insecure) {
+function createSession (insecure, cb) {
-  const key = crypto.generateKeyPair('RSA', 2048)
+  crypto.generateKeyPair('RSA', 2048, (err, key) => {
-  const id = PeerId.createFromPrivKey(key.bytes)
+    if (err) {
+      return cb(err)
+    }
 
-  return secio.encrypt(id, key, insecure)
+    key.public.hash((err, digest) => {
+      if (err) {
+        return cb(err)
+      }
+
+      cb(null, secio.encrypt(new PeerId(digest, key), key, insecure))
+    })
+  })
 }