refactor: crypto and pnet

.aegir.js

@@ -1,23 +1,38 @@
 'use strict'
 
-const TransportManager = require('./src/transport-manager')
-const mockUpgrader = require('./test/utils/mockUpgrader')
+const Libp2p = require('./src')
 const { MULTIADDRS_WEBSOCKETS } = require('./test/fixtures/browser')
-let tm
-
+const Peers = require('./test/fixtures/peers')
+const PeerId = require('peer-id')
+const PeerInfo = require('peer-info')
 const WebSockets = require('libp2p-websockets')
+const Muxer = require('libp2p-mplex')
+const Crypto = require('./src/insecure/plaintext')
+const pipe = require('it-pipe')
+let libp2p
 
 const before = async () => {
-  tm = new TransportManager({
-    upgrader: mockUpgrader,
-    onConnection: () => {}
+  // Use the last peer
+  const peerId = await PeerId.createFromJSON(Peers[Peers.length - 1])
+  const peerInfo = new PeerInfo(peerId)
+  peerInfo.multiaddrs.add(MULTIADDRS_WEBSOCKETS[0])
+
+  libp2p = new Libp2p({
+    peerInfo,
+    modules: {
+      transport: [WebSockets],
+      streamMuxer: [Muxer],
+      connEncryption: [Crypto]
+    }
   })
-  tm.add(WebSockets.prototype[Symbol.toStringTag], WebSockets)
-  await tm.listen(MULTIADDRS_WEBSOCKETS)
+  // Add the echo protocol
+  libp2p.handle('/echo/1.0.0', ({ stream }) => pipe(stream, stream))
+
+  await libp2p.start()
 }
 
 const after = async () => {
-  await tm.close()
+  await libp2p.stop()
 }
 
 module.exports = {

.travis.yml

@@ -20,7 +20,7 @@ jobs:
   include:
     - stage: check
       script:
-        - npx aegir build --bundlesize
+        # - npx aegir build --bundlesize
         - npx aegir dep-check -- -i wrtc -i electron-webrtc
         - npm run lint
 

package.json

@@ -49,10 +49,12 @@
     "err-code": "^1.1.2",
     "fsm-event": "^2.1.0",
     "hashlru": "^2.3.0",
+    "it-handshake": "^1.0.1",
+    "it-length-prefixed": "jacobheun/pull-length-prefixed#feat/fromReader",
     "it-pipe": "^1.0.1",
     "latency-monitor": "~0.2.1",
     "libp2p-crypto": "^0.16.2",
-    "libp2p-interfaces": "^0.1.1",
+    "libp2p-interfaces": "^0.1.3",
     "mafmt": "^7.0.0",
     "merge-options": "^1.0.1",
     "moving-average": "^1.0.0",
@@ -99,7 +101,7 @@
     "libp2p-secio": "^0.11.1",
     "libp2p-spdy": "^0.13.2",
     "libp2p-tcp": "^0.14.1",
-    "libp2p-websockets": "^0.13.0",
+    "libp2p-websockets": "^0.13.1",
     "lodash.times": "^4.3.2",
     "nock": "^10.0.6",
     "p-defer": "^3.0.0",

src/index.js

@@ -90,7 +90,7 @@ class Libp2p extends EventEmitter {
     if (this._modules.connEncryption) {
       const cryptos = this._modules.connEncryption
       cryptos.forEach((crypto) => {
-        this.upgrader.cryptos.set(crypto.tag, crypto)
+        this.upgrader.cryptos.set(crypto.protocol, crypto)
       })
     }
 
@@ -108,7 +108,7 @@ class Libp2p extends EventEmitter {
 
     // Attach private network protector
     if (this._modules.connProtector) {
-      this._switch.protector = this._modules.connProtector
+      this.upgrader.protector = this._modules.connProtector
     } else if (process.env.LIBP2P_FORCE_PNET) {
       throw new Error('Private network is enforced, but no protector was provided')
     }
@@ -229,6 +229,7 @@ class Libp2p extends EventEmitter {
 
     try {
       await this.transportManager.close()
+      await this._switch.stop()
     } catch (err) {
       if (err) {
         log.error(err)

src/insecure/plaintext.js

@@ -0,0 +1,67 @@
+'use strict'
+
+const handshake = require('it-handshake')
+const lp = require('it-length-prefixed')
+const PeerId = require('peer-id')
+const debug = require('debug')
+const log = debug('libp2p:plaintext')
+log.error = debug('libp2p:plaintext:error')
+const { UnexpectedPeerError, InvalidCryptoExchangeError } = require('libp2p-interfaces/src/crypto/errors')
+
+const { Exchange, KeyType } = require('./proto')
+const protocol = '/plaintext/2.0.0'
+
+function lpEncodeExchange (exchange) {
+  const pb = Exchange.encode(exchange)
+  return lp.encode.single(pb)
+}
+
+async function encrypt (localId, conn, remoteId) {
+  const shake = handshake(conn)
+
+  // Encode the public key and write it to the remote peer
+  shake.write(lpEncodeExchange({
+    id: localId.toBytes(),
+    pubkey: {
+      Type: KeyType.RSA, // TODO: dont hard code
+      Data: localId.marshalPubKey()
+    }
+  }))
+
+  log('write pubkey exchange to peer %j', remoteId)
+
+  // Get the Exchange message
+  const response = (await lp.decodeFromReader(shake.reader).next()).value
+  const id = Exchange.decode(response.slice())
+  log('read pubkey exchange from peer %j', remoteId)
+
+  let peerId
+  try {
+    peerId = await PeerId.createFromPubKey(id.pubkey.Data)
+  } catch (err) {
+    log.error(err)
+    throw new InvalidCryptoExchangeError('Remote did not provide its public key')
+  }
+
+  if (remoteId && !peerId.isEqual(remoteId)) {
+    throw new UnexpectedPeerError()
+  }
+
+  log('plaintext key exchange completed successfully with peer %j', peerId)
+
+  shake.rest()
+  return {
+    conn: shake.stream,
+    remotePeer: peerId
+  }
+}
+
+module.exports = {
+  protocol,
+  secureInbound: (localId, conn, remoteId) => {
+    return encrypt(localId, conn, remoteId)
+  },
+  secureOutbound: (localId, conn, remoteId) => {
+    return encrypt(localId, conn, remoteId)
+  }
+}

src/insecure/proto.js

@@ -0,0 +1,22 @@
+'use strict'
+
+const protobuf = require('protons')
+
+module.exports = protobuf(`
+message Exchange {
+  optional bytes id = 1;
+  optional PublicKey pubkey = 2;
+}
+
+enum KeyType {
+  RSA = 0;
+  Ed25519 = 1;
+  Secp256k1 = 2;
+  ECDSA = 3;
+}
+
+message PublicKey {
+  required KeyType Type = 1;
+  required bytes Data = 2;
+}
+`)

src/pnet/crypto.js

@@ -1,52 +1,46 @@
 'use strict'
 
-const pull = require('pull-stream')
 const debug = require('debug')
 const Errors = require('./errors')
 const xsalsa20 = require('xsalsa20')
 const KEY_LENGTH = require('./key-generator').KEY_LENGTH
 
 const log = debug('libp2p:pnet')
 log.trace = debug('libp2p:pnet:trace')
-log.err = debug('libp2p:pnet:err')
+log.error = debug('libp2p:pnet:err')
 
 /**
- * Creates a pull stream to encrypt messages in a private network
+  * Creates a stream iterable to encrypt messages in a private network
  *
  * @param {Buffer} nonce The nonce to use in encryption
  * @param {Buffer} psk The private shared key to use in encryption
- * @returns {PullStream} a through stream
+ * @returns {*} a through iterable
  */
 module.exports.createBoxStream = (nonce, psk) => {
   const xor = xsalsa20(nonce, psk)
-  return pull(
-    ensureBuffer(),
-    pull.map((chunk) => {
-      return xor.update(chunk, chunk)
-    })
-  )
+  return (source) => (async function * () {
+    for await (const chunk of source) {
+      yield Buffer.from(xor.update(chunk.slice()))
+    }
+  })()
 }
 
 /**
- * Creates a pull stream to decrypt messages in a private network
+  * Creates a stream iterable to decrypt messages in a private network
  *
- * @param {Object} remote Holds the nonce of the peer
+ * @param {Buffer} nonce The nonce of the remote peer
  * @param {Buffer} psk The private shared key to use in decryption
- * @returns {PullStream} a through stream
+ * @returns {*} a through iterable
  */
-module.exports.createUnboxStream = (remote, psk) => {
-  let xor
-  return pull(
-    ensureBuffer(),
-    pull.map((chunk) => {
-      if (!xor) {
-        xor = xsalsa20(remote.nonce, psk)
-        log.trace('Decryption enabled')
-      }
+module.exports.createUnboxStream = (nonce, psk) => {
+  return (source) => (async function * () {
+    const xor = xsalsa20(nonce, psk)
+    log.trace('Decryption enabled')
 
-      return xor.update(chunk, chunk)
-    })
-  )
+    for await (const chunk of source) {
+      yield Buffer.from(xor.update(chunk.slice()))
+    }
+  })()
 }
 
 /**
@@ -61,7 +55,7 @@ module.exports.decodeV1PSK = (pskBuffer) => {
     // This should pull from multibase/multicodec to allow for
     // more encoding flexibility. Ideally we'd consume the codecs
     // from the buffer line by line to evaluate the next line
-    // programatically instead of making assumptions about the
+    // programmatically instead of making assumptions about the
     // encodings of each line.
     const metadata = pskBuffer.toString().split(/(?:\r\n|\r|\n)/g)
     const pskTag = metadata.shift()
@@ -78,21 +72,7 @@ module.exports.decodeV1PSK = (pskBuffer) => {
       psk: psk
     }
   } catch (err) {
+    log.error(err)
     throw new Error(Errors.INVALID_PSK)
   }
 }
-
-/**
- * Returns a through pull-stream that ensures the passed chunks
- * are buffers instead of strings
- * @returns {PullStream} a through stream
- */
-function ensureBuffer () {
-  return pull.map((chunk) => {
-    if (typeof chunk === 'string') {
-      return Buffer.from(chunk, 'utf-8')
-    }
-
-    return chunk
-  })
-}

src/pnet/index.js

@@ -1,12 +1,17 @@
 'use strict'
 
-const pull = require('pull-stream')
-const { Connection } = require('libp2p-interfaces/src/connection')
+const pipe = require('it-pipe')
 const assert = require('assert')
-
+const duplexPair = require('it-pair/duplex')
+const crypto = require('libp2p-crypto')
 const Errors = require('./errors')
-const State = require('./state')
-const decodeV1PSK = require('./crypto').decodeV1PSK
+const {
+  createBoxStream,
+  createUnboxStream,
+  decodeV1PSK
+} = require('./crypto')
+const handshake = require('it-handshake')
+const { NONCE_LENGTH } = require('./key-generator')
 const debug = require('debug')
 const log = debug('libp2p:pnet')
 log.err = debug('libp2p:pnet:err')
@@ -27,41 +32,41 @@ class Protector {
   }
 
   /**
-   * Takes a given Connection and creates a privaste encryption stream
+   * Takes a given Connection and creates a private encryption stream
    * between its two peers from the PSK the Protector instance was
    * created with.
    *
    * @param {Connection} connection The connection to protect
-   * @param {function(Error)} callback
-   * @returns {Connection} The protected connection
+   * @returns {*} A protected duplex iterable
    */
-  protect (connection, callback) {
+  async protect (connection) {
     assert(connection, Errors.NO_HANDSHAKE_CONNECTION)
 
-    const protectedConnection = new Connection(undefined, connection)
-    const state = new State(this.psk)
-
+    // Exchange nonces
     log('protecting the connection')
+    const localNonce = crypto.randomBytes(NONCE_LENGTH)
+
+    const shake = handshake(connection)
+    shake.write(localNonce)
 
-    // Run the connection through an encryptor
-    pull(
-      connection,
-      state.encrypt((err, encryptedOuterStream) => {
-        if (err) {
-          log.err('There was an error attempting to protect the connection', err)
-          return callback(err)
-        }
+    const result = await shake.reader.next(NONCE_LENGTH)
+    const remoteNonce = result.value.slice()
+    shake.rest()
 
-        connection.getPeerInfo(() => {
-          protectedConnection.setInnerConn(new Connection(encryptedOuterStream, connection))
-          log('the connection has been successfully wrapped by the protector')
-          callback()
-        })
-      }),
-      connection
+    // Create the boxing/unboxing pipe
+    log('exchanged nonces')
+    const [internal, external] = duplexPair()
+    pipe(
+      external,
+      // Encrypt all outbound traffic
+      createBoxStream(localNonce, this.psk),
+      shake.stream,
+      // Decrypt all inbound traffic
+      createUnboxStream(remoteNonce, this.psk),
+      external
     )
 
-    return protectedConnection
+    return internal
   }
 }