[WIP] Refactor to use pull streams

README.md

@@ -4,14 +4,14 @@
 [![](https://img.shields.io/badge/project-IPFS-blue.svg?style=flat-square)](http://ipfs.io/)
 [![](https://img.shields.io/badge/freenode-%23ipfs-blue.svg?style=flat-square)](http://webchat.freenode.net/?channels=%23ipfs)
 [![standard-readme compliant](https://img.shields.io/badge/standard--readme-OK-green.svg?style=flat-square)](https://github.com/RichardLitt/standard-readme)
-[![Coverage Status](https://coveralls.io/repos/github/ipfs/js-libp2p-secio/badge.svg?branch=master)](https://coveralls.io/github/ipfs/js-libp2p-secio?branch=master)
-[![Travis CI](https://travis-ci.org/ipfs/js-libp2p-secio.svg?branch=master)](https://travis-ci.org/ipfs/js-libp2p-secio)
-[![Circle CI](https://circleci.com/gh/ipfs/js-libp2p-secio.svg?style=svg)](https://circleci.com/gh/ipfs/js-libp2p-secio)
-[![Dependency Status](https://david-dm.org/ipfs/js-libp2p-secio.svg?style=flat-square)](https://david-dm.org/ipfs/js-libp2p-secio) [![js-standard-style](https://img.shields.io/badge/code%20style-standard-brightgreen.svg?style=flat-square)](https://github.com/feross/standard)
+[![Coverage Status](https://coveralls.io/repos/github/libp2p/js-libp2p-secio/badge.svg?branch=master)](https://coveralls.io/github/libp2p/js-libp2p-secio?branch=master)
+[![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)
 
 > Secio implementation in JavaScript
 
-This repo contains the JavaScript implementation of secio, an encryption protocol used in libp2p. This is based on this [go implementation](https://github.com/ipfs/go-libp2p-secio).
+This repo contains the JavaScript implementation of secio, an encryption protocol used in libp2p. This is based on this [go implementation](https://github.com/libp2p/go-libp2p-secio).
 
 ## Table of Contents
 
@@ -24,17 +24,55 @@ This repo contains the JavaScript implementation of secio, an encryption protoco
 ## Install
 
 ```sh
-npm install --save libp2p-secio
+npm install libp2p-secio
 ```
 
 ## Usage
 
 ```js
-const libp2pSecio = require('libp2p-secio')
+const secio = require('libp2p-secio')
 ```
 
 ## API
 
+### `SecureSession`
+
+#### `constructor(id, key, insecure)`
+
+- `id: PeerId` - The id of the node.
+- `key: RSAPrivateKey` - The private key of the node.
+- `insecure: PullStream` - The insecure connection.
+
+### `.secure`
+
+Returns the `insecure` connection provided, wrapped with secio. This is a pull-stream.
+
+### This module uses `pull-streams`
+
+We expose a streaming interface based on `pull-streams`, rather then on the Node.js core streams implementation (aka Node.js streams). `pull-streams` offers us a better mechanism for error handling and flow control guarantees. If you would like to know more about why we did this, see the discussion at this [issue](https://github.com/ipfs/js-ipfs/issues/362).
+
+You can learn more about pull-streams at:
+
+- [The history of Node.js streams, nodebp April 2014](https://www.youtube.com/watch?v=g5ewQEuXjsQ)
+- [The history of streams, 2016](http://dominictarr.com/post/145135293917/history-of-streams)
+- [pull-streams, the simple streaming primitive](http://dominictarr.com/post/149248845122/pull-streams-pull-streams-are-a-very-simple)
+- [pull-streams documentation](https://pull-stream.github.io/)
+
+#### Converting `pull-streams` to Node.js Streams
+
+If you are a Node.js streams user, you can convert a pull-stream to a Node.js stream using the module [`pull-stream-to-stream`](https://github.com/dominictarr/pull-stream-to-stream), giving you an instance of a Node.js stream that is linked to the pull-stream. For example:
+
+```js
+const pullToStream = require('pull-stream-to-stream')
+
+const nodeStreamInstance = pullToStream(pullStreamInstance)
+// nodeStreamInstance is an instance of a Node.js Stream
+```
+
+To learn more about this utility, visit https://pull-stream.github.io/#pull-stream-to-stream.
+
+
+
 ## Contribute
 
 Feel free to join in. All welcome. Open an [issue](https://github.com/ipfs/js-libp2p-secio/issues)!

package.json

@@ -25,27 +25,27 @@
   "author": "Friedel Ziegelmayer <dignifiedqurie@gmail.com>",
   "license": "MIT",
   "dependencies": {
-    "async-buffered-reader": "^1.2.1",
     "debug": "^2.2.0",
-    "duplexify": "^3.4.3",
-    "length-prefixed-stream": "^1.5.0",
+    "interface-connection": "^0.2.1",
     "libp2p-crypto": "^0.5.0",
     "multihashing": "^0.2.1",
-    "node-forge": "^0.6.39",
+    "node-forge": "^0.6.42",
     "peer-id": "^0.7.0",
     "protocol-buffers": "^3.1.6",
-    "readable-stream": "2.1.4",
-    "run-series": "^1.1.4",
-    "through2": "^2.0.1"
+    "pull-defer": "^0.2.2",
+    "pull-handshake": "^1.1.4",
+    "pull-length-prefixed": "^1.2.0",
+    "pull-stream": "^3.4.3",
+    "pull-through": "^1.0.18",
+    "run-series": "^1.1.4"
   },
   "devDependencies": {
-    "aegir": "^6.0.0",
-    "bl": "^1.1.2",
+    "aegir": "^8.0.0",
     "chai": "^3.5.0",
-    "multistream-select": "^0.10.0",
+    "multistream-select": "^0.11.0",
     "pre-commit": "^1.1.3",
-    "run-parallel": "^1.1.6",
-    "stream-pair": "^1.0.3"
+    "pull-pair": "^1.1.0",
+    "run-parallel": "^1.1.6"
   },
   "pre-commit": [
     "lint",
@@ -65,4 +65,4 @@
   "contributors": [
     "dignifiedquire <dignifiedquire@gmail.com>"
   ]
-}
+}

src/etm.js

@@ -1,41 +1,44 @@
 'use strict'
 
-const through = require('through2')
-const lpm = require('length-prefixed-stream')
+const through = require('pull-through')
+const pull = require('pull-stream')
+const lp = require('pull-length-prefixed')
 
 const toForgeBuffer = require('./support').toForgeBuffer
 
-exports.writer = function etmWriter (insecure, cipher, mac) {
-  const encode = lpm.encode()
-  const pt = through(function (chunk, enc, cb) {
+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)
-      const macBuffer = new Buffer(mac.getMac().getBytes(), 'binary')
+      mac.update(data.toString('binary'))
+      const macBuffer = new Buffer(mac.digest().getBytes(), 'binary')
 
-      this.push(Buffer.concat([data, macBuffer]))
+      this.queue(Buffer.concat([data, macBuffer]))
       // reset hmac
       mac.start(null, null)
     }
-
-    cb()
   })
 
-  pt.pipe(encode).pipe(insecure)
-
-  return pt
+  return pull(
+    pt,
+    lp.encode(lpOpts)
+  )
 }
 
-exports.reader = function etmReader (insecure, decipher, mac) {
-  const decode = lpm.decode()
-  const pt = through(function (chunk, enc, cb) {
+exports.createUnboxStream = (decipher, mac) => {
+  const pt = through(function (chunk) {
     const l = chunk.length
     const macSize = mac.getMac().length()
 
     if (l < macSize) {
-      return cb(new Error(`buffer (${l}) shorter than MAC size (${macSize})`))
+      return this.emit('error', new Error(`buffer (${l}) shorter than MAC size (${macSize})`))
     }
 
     const mark = l - macSize
@@ -45,26 +48,26 @@ exports.reader = function etmReader (insecure, decipher, mac) {
     // Clear out any previous data
     mac.start(null, null)
 
-    mac.update(data)
+    mac.update(data.toString('binary'))
     const expected = new Buffer(mac.getMac().getBytes(), 'binary')
+
     // reset hmac
     mac.start(null, null)
     if (!macd.equals(expected)) {
-      return cb(new Error(`MAC Invalid: ${macd.toString('hex')} != ${expected.toString('hex')}`))
+      return this.emit('error', new Error(`MAC Invalid: ${macd.toString('hex')} != ${expected.toString('hex')}`))
     }
 
     // all good, decrypt
     decipher.update(toForgeBuffer(data))
 
     if (decipher.output.length() > 0) {
       const data = new Buffer(decipher.output.getBytes(), 'binary')
-      this.push(data)
+      this.queue(data)
     }
-
-    cb()
   })
 
-  insecure.pipe(decode).pipe(pt)
-
-  return pt
+  return pull(
+    lp.decode(lpOpts),
+    pt
+  )
 }

src/handshake/crypto.js

@@ -0,0 +1,163 @@
+'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 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 support = require('../support')
+
+// nonceSize is the size of our nonces (in bytes)
+const nonceSize = 16
+
+exports.createProposal = (state) => {
+  state.proposal.out = {
+    rand: support.randomBytes(nonceSize),
+    pubkey: state.key.local.public.bytes,
+    exchanges: support.exchanges.join(','),
+    ciphers: support.ciphers.join(','),
+    hashes: support.hashes.join(',')
+  }
+
+  state.proposalEncoded.out = pbm.Propose.encode(state.proposal.out)
+  return state.proposalEncoded.out
+}
+
+exports.createExchange = (state) => {
+  const res = crypto.generateEphemeralKeyPair(state.protocols.local.curveT)
+  state.ephemeralKey.local = res.key
+  state.shared.generate = res.genSharedKey
+
+  // Gather corpus to sign.
+  const selectionOut = Buffer.concat([
+    state.proposalEncoded.out,
+    state.proposalEncoded.in,
+    state.ephemeralKey.local
+  ])
+
+  state.exchange.out = {
+    epubkey: state.ephemeralKey.local,
+    signature: new Buffer(state.key.local.sign(selectionOut), 'binary')
+  }
+
+  return pbm.Exchange.encode(state.exchange.out)
+}
+
+exports.identify = (state, msg) => {
+  log('1.1 identify')
+
+  state.proposalEncoded.in = msg
+  state.proposal.in = pbm.Propose.decode(msg)
+  const pubkey = state.proposal.in.pubkey
+
+  console.log(state.proposal.in)
+
+  state.key.remote = crypto.unmarshalPublicKey(pubkey)
+  state.id.remote = PeerId.createFromPubKey(pubkey.toString('base64'))
+
+  log('1.1 identify - %s - identified remote peer as %s', state.id.local.toB58String(), state.id.remote.toB58String())
+}
+
+exports.selectProtocols = (state) => {
+  log('1.2 selection')
+
+  const local = {
+    pubKeyBytes: state.key.local.public.bytes,
+    exchanges: support.exchanges,
+    hashes: support.hashes,
+    ciphers: support.ciphers,
+    nonce: state.proposal.out.rand
+  }
+
+  const remote = {
+    pubKeyBytes: state.proposal.in.pubkey,
+    exchanges: state.proposal.in.exchanges.split(','),
+    hashes: state.proposal.in.hashes.split(','),
+    ciphers: state.proposal.in.ciphers.split(','),
+    nonce: state.proposal.in.rand
+  }
+
+  let selected = support.selectBest(local, remote)
+  // 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 = {
+    order: selected.order,
+    curveT: selected.curveT,
+    cipherT: selected.cipherT,
+    hashT: selected.hashT
+  }
+
+  state.protocols.local = {
+    order: selected.order,
+    curveT: selected.curveT,
+    cipherT: selected.cipherT,
+    hashT: selected.hashT
+  }
+}
+
+exports.verify = (state, msg) => {
+  log('2.1. verify')
+
+  state.exchange.in = pbm.Exchange.decode(msg)
+  state.ephemeralKey.remote = state.exchange.in.epubkey
+
+  const selectionIn = Buffer.concat([
+    state.proposalEncoded.in,
+    state.proposalEncoded.out,
+    state.ephemeralKey.remote
+  ])
+
+  const sigOk = state.key.remote.verify(selectionIn, state.exchange.in.signature)
+
+  if (!sigOk) {
+    throw new Error('Bad signature')
+  }
+
+  log('2.1. verify - signature verified')
+}
+
+exports.generateKeys = (state) => {
+  log('2.2. keys')
+
+  state.shared.secret = state.shared.generate(state.exchange.in.epubkey)
+
+  const keys = crypto.keyStretcher(
+    state.protocols.local.cipherT,
+    state.protocols.local.hashT,
+    state.shared.secret
+  )
+
+  // use random nonces to decide order.
+  if (state.protocols.local.order > 0) {
+    state.protocols.local.keys = keys.k1
+    state.protocols.remote.keys = keys.k2
+  } else if (state.protocols.local.order < 0) {
+    // swap
+    state.protocols.local.keys = keys.k2
+    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')
+  }
+
+  log('2.3. mac + cipher')
+
+  support.makeMacAndCipher(state.protocols.local)
+  support.makeMacAndCipher(state.protocols.remote)
+}
+
+exports.verifyNonce = (state, n2) => {
+  const n1 = state.proposal.out.rand
+
+  if (n1.equals(n2)) return
+
+  throw new Error(
+    `Failed to read our encrypted nonce: ${n1.toString('hex')} != ${n2.toString('hex')}`
+  )
+}