refactor: crypto and pnet (#469)

* feat: add initial plaintext 2 module

* refactor: initial refactor of pnet

* chore: fix lint

* fix: update plaintext api usage

* test: use plaintext for test crypto

* chore: update deps

test: update dialer suite scope

* feat: add connection protection to the upgrader

* refactor: cleanup and lint fix

* chore: remove unncessary transforms

* chore: temporarily disable bundlesize

* chore: add missing dep

* fix: use it-handshake to prevent overreading

* chore(fix): PR feedback updates

* chore: apply suggestions from code review

Co-Authored-By: Vasco Santos <vasco.santos@moxy.studio>

.aegir.js

@@ -1,23 +1,38 @@
 'use strict'
 
-const TransportManager = require('./src/transport-manager')
+const Libp2p = require('./src')
-const mockUpgrader = require('./test/utils/mockUpgrader')
 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({
+  // Use the last peer
-    upgrader: mockUpgrader,
+  const peerId = await PeerId.createFromJSON(Peers[Peers.length - 1])
-    onConnection: () => {}
+  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)
+  // Add the echo protocol
-  await tm.listen(MULTIADDRS_WEBSOCKETS)
+  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,6 +1,5 @@
 'use strict'
 
-const pull = require('pull-stream')
 const debug = require('debug')
 const Errors = require('./errors')
 const xsalsa20 = require('xsalsa20')
@@ -8,45 +7,40 @@ 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(
+  return (source) => (async function * () {
-    ensureBuffer(),
+    for await (const chunk of source) {
-    pull.map((chunk) => {
+      yield Buffer.from(xor.update(chunk.slice()))
-      return xor.update(chunk, chunk)
+    }
-    })
+  })()
-  )
 }
 
 /**
- * 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) => {
+module.exports.createUnboxStream = (nonce, psk) => {
-  let xor
+  return (source) => (async function * () {
-  return pull(
+    const xor = xsalsa20(nonce, psk)
-    ensureBuffer(),
-    pull.map((chunk) => {
-      if (!xor) {
-        xor = xsalsa20(remote.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 pipe = require('it-pipe')
-const { Connection } = require('libp2p-interfaces/src/connection')
 const assert = require('assert')
-
+const duplexPair = require('it-pair/duplex')
+const crypto = require('libp2p-crypto')
 const Errors = require('./errors')
-const State = require('./state')
+const {
-const decodeV1PSK = require('./crypto').decodeV1PSK
+  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 {*} A protected duplex iterable
-   * @returns {Connection} The protected connection
    */
-  protect (connection, callback) {
+  async protect (connection) {
     assert(connection, Errors.NO_HANDSHAKE_CONNECTION)
 
-    const protectedConnection = new Connection(undefined, connection)
+    // Exchange nonces
-    const state = new State(this.psk)
-
     log('protecting the connection')
+    const localNonce = crypto.randomBytes(NONCE_LENGTH)
 
-    // Run the connection through an encryptor
+    const shake = handshake(connection)
-    pull(
+    shake.write(localNonce)
-      connection,
-      state.encrypt((err, encryptedOuterStream) => {
-        if (err) {
-          log.err('There was an error attempting to protect the connection', err)
-          return callback(err)
-        }
 
-        connection.getPeerInfo(() => {
+    const result = await shake.reader.next(NONCE_LENGTH)
-          protectedConnection.setInnerConn(new Connection(encryptedOuterStream, connection))
+    const remoteNonce = result.value.slice()
-          log('the connection has been successfully wrapped by the protector')
+    shake.rest()
-          callback()
+
-        })
+    // Create the boxing/unboxing pipe
-      }),
+    log('exchanged nonces')
-      connection
+    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
   }
 }
 

src/pnet/state.js

@@ -1,110 +0,0 @@
-'use strict'
-
-const crypto = require('crypto')
-const debug = require('debug')
-const pair = require('pull-pair')
-const Reader = require('pull-reader')
-const cat = require('pull-cat')
-const pull = require('pull-stream')
-const deferred = require('pull-defer')
-
-const cryptoStreams = require('./crypto')
-const NONCE_LENGTH = require('./key-generator').NONCE_LENGTH
-
-const log = debug('libp2p:pnet')
-log.err = debug('libp2p:pnet:err')
-log.trace = debug('libp2p:pnet:trace')
-
-/**
- * Keeps track of the state of a given connection, such as the local psk
- * and local and remote nonces for encryption/decryption
- */
-class State {
-  /**
-   * @param {Buffer} psk The key buffer used for encryption
-   * @constructor
-   */
-  constructor (psk) {
-    this.local = {
-      nonce: Buffer.from(
-        crypto.randomBytes(NONCE_LENGTH)
-      ),
-      psk: psk
-    }
-    this.remote = { nonce: null }
-
-    this.rawReader = Reader(60e3)
-    this.encryptedReader = Reader(60e3)
-
-    this.rawPairStream = pair()
-    this.encryptedPairStream = pair()
-
-    // The raw, pair stream
-    this.innerRawStream = null
-    this.outerRawStream = {
-      sink: this.rawReader,
-      source: cat([
-        pull.values([
-          this.local.nonce
-        ]),
-        this.rawPairStream.source
-      ])
-    }
-
-    // The encrypted, pair stream
-    this.innerEncryptedStream = {
-      sink: this.encryptedReader,
-      source: this.encryptedPairStream.source
-    }
-    this.outerEncryptedStream = null
-  }
-
-  /**
-   * Creates encryption streams for the given state
-   *
-   * @param {function(Error, Connection)} callback
-   * @returns {void}
-   */
-  encrypt (callback) {
-    // The outer stream needs to be returned before we setup the
-    // rest of the streams, so we're delaying the execution
-    setTimeout(() => {
-      // Read the nonce first, once we have it resolve the
-      // deferred source, so we keep reading
-      const deferredSource = deferred.source()
-      this.rawReader.read(NONCE_LENGTH, (err, data) => {
-        if (err) {
-          log.err('There was an error attempting to read the nonce', err)
-        }
-        log.trace('remote nonce received')
-        this.remote.nonce = data
-        deferredSource.resolve(this.rawReader.read())
-      })
-
-      this.innerRawStream = {
-        sink: this.rawPairStream.sink,
-        source: deferredSource
-      }
-
-      // Create the pull exchange between the two inner streams
-      pull(
-        this.innerRawStream,
-        cryptoStreams.createUnboxStream(this.remote, this.local.psk),
-        this.innerEncryptedStream,
-        cryptoStreams.createBoxStream(this.local.nonce, this.local.psk),
-        this.innerRawStream
-      )
-
-      this.outerEncryptedStream = {
-        sink: this.encryptedPairStream.sink,
-        source: this.encryptedReader.read()
-      }
-
-      callback(null, this.outerEncryptedStream)
-    }, 0)
-
-    return this.outerRawStream
-  }
-}
-
-module.exports = State

src/upgrader.js

@@ -32,11 +32,20 @@ class Upgrader {
    * @param {PeerId} options.localPeer
    * @param {Map<string, Crypto>} options.cryptos
    * @param {Map<string, Muxer>} options.muxers
+   * @param {function(Connection)} options.onConnection Called when a connection is upgraded
+   * @param {function(Connection)} options.onConnectionEnd
    */
-  constructor ({ localPeer, cryptos, muxers, onConnectionEnd = () => {}, onConnection = () => {} }) {
+  constructor ({
+    localPeer,
+    cryptos,
+    muxers,
+    onConnectionEnd = () => {},
+    onConnection = () => {}
+  }) {
     this.localPeer = localPeer
     this.cryptos = cryptos || new Map()
     this.muxers = muxers || new Map()
+    this.protector = null
     this.protocols = new Map()
     this.onConnection = onConnection
     this.onConnectionEnd = onConnectionEnd
@@ -55,13 +64,21 @@ class Upgrader {
     let Muxer
     let cryptoProtocol
 
+    log('Starting the inbound connection upgrade')
+
+    // Protect
+    let protectedConn = maConn
+    if (this.protector) {
+      protectedConn = await this.protector.protect(maConn)
+    }
+
     try {
       // Encrypt the connection
       ({
         conn: encryptedConn,
         remotePeer,
         protocol: cryptoProtocol
-      } = await this._encryptInbound(this.localPeer, maConn, this.cryptos))
+      } = await this._encryptInbound(this.localPeer, protectedConn, this.cryptos))
 
       // Multiplex the connection
       ;({ stream: muxedConnection, Muxer } = await this._multiplexInbound(encryptedConn, this.muxers))
@@ -104,13 +121,21 @@ class Upgrader {
     let cryptoProtocol
     let Muxer
 
+    log('Starting the outbound connection upgrade')
+
+    // Protect
+    let protectedConn = maConn
+    if (this.protector) {
+      protectedConn = await this.protector.protect(maConn)
+    }
+
     try {
       // Encrypt the connection
       ({
         conn: encryptedConn,
         remotePeer,
         protocol: cryptoProtocol
-      } = await this._encryptOutbound(this.localPeer, maConn, remotePeerId, this.cryptos))
+      } = await this._encryptOutbound(this.localPeer, protectedConn, remotePeerId, this.cryptos))
 
       // Multiplex the connection
       ;({ stream: muxedConnection, Muxer } = await this._multiplexOutbound(encryptedConn, this.muxers))
@@ -241,7 +266,7 @@ class Upgrader {
   async _encryptInbound (localPeer, connection, cryptos) {
     const mss = new Multistream.Listener(connection)
     const protocols = Array.from(cryptos.keys())
-    log('selecting inbound crypto protocol', protocols)
+    log('handling inbound crypto protocol selection', protocols)
 
     try {
       const { stream, protocol } = await mss.handle(protocols)

test/dialing/direct.node.js

@@ -7,6 +7,7 @@ const { expect } = chai
 const sinon = require('sinon')
 const Transport = require('libp2p-tcp')
 const Muxer = require('libp2p-mplex')
+const Crypto = require('../../src/insecure/plaintext')
 const multiaddr = require('multiaddr')
 const PeerId = require('peer-id')
 const PeerInfo = require('peer-info')
@@ -18,9 +19,10 @@ const Libp2p = require('../../src')
 const Dialer = require('../../src/dialer')
 const TransportManager = require('../../src/transport-manager')
 const { codes: ErrorCodes } = require('../../src/errors')
+const Protector = require('../../src/pnet')
+const swarmKeyBuffer = Buffer.from(require('../fixtures/swarm.key'))
 
 const mockUpgrader = require('../utils/mockUpgrader')
-const mockCrypto = require('../utils/mockCrypto')
 const Peers = require('../fixtures/peers')
 
 const listenAddr = multiaddr('/ip4/127.0.0.1/tcp/0')
@@ -49,9 +51,7 @@ describe('Dialing (direct, TCP)', () => {
     remoteAddr = remoteTM.getAddrs()[0]
   })
 
-  after(async () => {
+  after(() => remoteTM.close())
-    await remoteTM.close()
-  })
 
   afterEach(() => {
     sinon.restore()
@@ -171,7 +171,6 @@ describe('Dialing (direct, TCP)', () => {
     expect(dialer.queue.pending).to.equal(0)
     expect(dialer.queue.size).to.equal(0)
   })
-})
 
   describe('libp2p.dialer', () => {
     let peerInfo
@@ -194,7 +193,7 @@ describe('libp2p.dialer', () => {
         modules: {
           transport: [Transport],
           streamMuxer: [Muxer],
-        connEncryption: [mockCrypto]
+          connEncryption: [Crypto]
         }
       })
       remoteLibp2p.handle('/echo/1.0.0', ({ stream }) => pipe(stream, stream))
@@ -209,9 +208,7 @@ describe('libp2p.dialer', () => {
       libp2p = null
     })
 
-  after(async () => {
+    after(() => remoteLibp2p.stop())
-    await remoteLibp2p.stop()
-  })
 
     it('should use the dialer for connecting', async () => {
       libp2p = new Libp2p({
@@ -219,7 +216,7 @@ describe('libp2p.dialer', () => {
         modules: {
           transport: [Transport],
           streamMuxer: [Muxer],
-        connEncryption: [mockCrypto]
+          connEncryption: [Crypto]
         }
       })
 
@@ -233,4 +230,29 @@ describe('libp2p.dialer', () => {
       await connection.close()
       expect(libp2p.dialer.connectToMultiaddr.callCount).to.equal(1)
     })
+
+    it('should use the protectors when provided for connecting', async () => {
+      const protector = new Protector(swarmKeyBuffer)
+      libp2p = new Libp2p({
+        peerInfo,
+        modules: {
+          transport: [Transport],
+          streamMuxer: [Muxer],
+          connEncryption: [Crypto],
+          connProtector: protector
+        }
+      })
+
+      sinon.spy(libp2p.upgrader.protector, 'protect')
+      sinon.stub(remoteLibp2p.upgrader, 'protector').value(new Protector(swarmKeyBuffer))
+
+      const connection = await libp2p.dialer.connectToMultiaddr(remoteAddr)
+      expect(connection).to.exist()
+      const { stream, protocol } = await connection.newStream('/echo/1.0.0')
+      expect(stream).to.exist()
+      expect(protocol).to.equal('/echo/1.0.0')
+      await connection.close()
+      expect(libp2p.upgrader.protector.protect.callCount).to.equal(1)
+    })
+  })
 })

test/dialing/direct.spec.js

@@ -9,6 +9,7 @@ const pDefer = require('p-defer')
 const delay = require('delay')
 const Transport = require('libp2p-websockets')
 const Muxer = require('libp2p-mplex')
+const Crypto = require('../../src/insecure/plaintext')
 const multiaddr = require('multiaddr')
 const PeerId = require('peer-id')
 const PeerInfo = require('peer-info')
@@ -22,7 +23,6 @@ const Libp2p = require('../../src')
 const Peers = require('../fixtures/peers')
 const { MULTIADDRS_WEBSOCKETS } = require('../fixtures/browser')
 const mockUpgrader = require('../utils/mockUpgrader')
-const mockCrypto = require('../utils/mockCrypto')
 const unsupportedAddr = multiaddr('/ip4/127.0.0.1/tcp/9999/ws')
 const remoteAddr = MULTIADDRS_WEBSOCKETS[0]
 
@@ -162,11 +162,11 @@ describe('Dialing (direct, WebSockets)', () => {
     expect(dialer.queue.pending).to.equal(0)
     expect(dialer.queue.size).to.equal(0)
   })
-})
 
-describe.skip('libp2p.dialer', () => {
+  describe('libp2p.dialer', () => {
     let peerInfo
     let libp2p
+    let remoteLibp2p
 
     before(async () => {
       const peerId = await PeerId.createFromJSON(Peers[0])
@@ -179,13 +179,17 @@ describe.skip('libp2p.dialer', () => {
       libp2p = null
     })
 
+    after(async () => {
+      remoteLibp2p && await remoteLibp2p.stop()
+    })
+
     it('should create a dialer', () => {
       libp2p = new Libp2p({
         peerInfo,
         modules: {
           transport: [Transport],
           streamMuxer: [Muxer],
-        connEncryption: [mockCrypto]
+          connEncryption: [Crypto]
         }
       })
 
@@ -200,12 +204,19 @@ describe.skip('libp2p.dialer', () => {
         modules: {
           transport: [Transport],
           streamMuxer: [Muxer],
-        connEncryption: [mockCrypto]
+          connEncryption: [Crypto]
         }
       })
 
+      sinon.spy(libp2p.dialer, 'connectToMultiaddr')
+
       const connection = await libp2p.dialer.connectToMultiaddr(remoteAddr)
       expect(connection).to.exist()
+      const { stream, protocol } = await connection.newStream('/echo/1.0.0')
+      expect(stream).to.exist()
+      expect(protocol).to.equal('/echo/1.0.0')
       await connection.close()
+      expect(libp2p.dialer.connectToMultiaddr.callCount).to.equal(1)
+    })
   })
 })

test/fixtures/browser.js

@@ -3,5 +3,5 @@
 const multiaddr = require('multiaddr')
 
 module.exports.MULTIADDRS_WEBSOCKETS = [
-  multiaddr('/ip4/127.0.0.1/tcp/15001/ws')
+  multiaddr('/ip4/127.0.0.1/tcp/15001/ws/p2p/QmckxVrJw1Yo8LqvmDJNUmdAsKtSbiKWmrXJFyKmUraBoN')
 ]

test/fixtures/swarm.key.js

@@ -0,0 +1,5 @@
+'use strict'
+
+module.exports = '/key/swarm/psk/1.0.0/\n' +
+  '/base16/\n' +
+  '411f0a244cbbc25ecbb2b070d00a1832516ded521eb3ee3aa13189efe2e2b9a2'

test/insecure/compliance.spec.js

@@ -0,0 +1,13 @@
+'use strict'
+/* eslint-env mocha */
+
+const tests = require('libp2p-interfaces/src/crypto/tests')
+const plaintext = require('../../src/insecure/plaintext')
+
+describe('plaintext compliance', () => {
+  tests({
+    setup () {
+      return plaintext
+    }
+  })
+})

test/insecure/plaintext.spec.js

@@ -0,0 +1,69 @@
+'use strict'
+/* eslint-env mocha */
+
+const chai = require('chai')
+chai.use(require('dirty-chai'))
+const { expect } = chai
+const sinon = require('sinon')
+
+const PeerId = require('peer-id')
+const duplexPair = require('it-pair/duplex')
+
+const peers = require('../fixtures/peers')
+const plaintext = require('../../src/insecure/plaintext')
+const {
+  InvalidCryptoExchangeError,
+  UnexpectedPeerError
+} = require('libp2p-interfaces/src/crypto/errors')
+
+describe('plaintext', () => {
+  let localPeer
+  let remotePeer
+  let wrongPeer
+
+  before(async () => {
+    [localPeer, remotePeer, wrongPeer] = await Promise.all([
+      PeerId.createFromJSON(peers[0]),
+      PeerId.createFromJSON(peers[1]),
+      PeerId.createFromJSON(peers[2])
+    ])
+  })
+
+  afterEach(() => {
+    sinon.restore()
+  })
+
+  it('should verify the public key and id match', () => {
+    const [localConn, remoteConn] = duplexPair()
+
+    // When we attempt to get the remote peer key, return the wrong peers pub key
+    sinon.stub(remotePeer, 'marshalPubKey').callsFake(() => {
+      return wrongPeer.marshalPubKey()
+    })
+
+    return Promise.all([
+      plaintext.secureInbound(remotePeer, localConn),
+      plaintext.secureOutbound(localPeer, remoteConn, remotePeer)
+    ]).then(() => expect.fail('should have failed'), (err) => {
+      expect(err).to.exist()
+      expect(err).to.have.property('code', UnexpectedPeerError.code)
+    })
+  })
+
+  it('should fail if the peer does not provide its public key', () => {
+    const [localConn, remoteConn] = duplexPair()
+
+    // When we attempt to get the remote peer key, return the wrong peers pub key
+    sinon.stub(remotePeer, 'marshalPubKey').callsFake(() => {
+      return Buffer.alloc(0)
+    })
+
+    return Promise.all([
+      plaintext.secureInbound(remotePeer, localConn),
+      plaintext.secureOutbound(localPeer, remoteConn, remotePeer)
+    ]).then(() => expect.fail('should have failed'), (err) => {
+      expect(err).to.exist()
+      expect(err).to.have.property('code', InvalidCryptoExchangeError.code)
+    })
+  })
+})

test/pnet/index.spec.js

@@ -0,0 +1,94 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+const dirtyChai = require('dirty-chai')
+chai.use(dirtyChai)
+const expect = chai.expect
+const duplexPair = require('it-pair/duplex')
+const pipe = require('it-pipe')
+const { collect } = require('streaming-iterables')
+
+const Protector = require('../../src/pnet')
+const Errors = Protector.errors
+const generate = Protector.generate
+
+const swarmKeyBuffer = Buffer.alloc(95)
+const wrongSwarmKeyBuffer = Buffer.alloc(95)
+
+// Write new psk files to the buffers
+generate(swarmKeyBuffer)
+generate(wrongSwarmKeyBuffer)
+
+describe('private network', () => {
+  it('should accept a valid psk buffer', () => {
+    const protector = new Protector(swarmKeyBuffer)
+
+    expect(protector.tag).to.equal('/key/swarm/psk/1.0.0/')
+    expect(protector.psk.byteLength).to.equal(32)
+  })
+
+  it('should protect a simple connection', async () => {
+    const [inbound, outbound] = duplexPair()
+    const protector = new Protector(swarmKeyBuffer)
+
+    const [aToB, bToA] = await Promise.all([
+      protector.protect(inbound),
+      protector.protect(outbound)
+    ])
+
+    pipe(
+      [Buffer.from('hello world'), Buffer.from('doo dah')],
+      aToB
+    )
+
+    const output = await pipe(
+      bToA,
+      source => (async function * () {
+        for await (const chunk of source) {
+          yield chunk.slice()
+        }
+      })(),
+      collect
+    )
+
+    expect(output).to.eql([Buffer.from('hello world'), Buffer.from('doo dah')])
+  })
+
+  it('should not be able to share correct data with different keys', async () => {
+    const [inbound, outbound] = duplexPair()
+    const protector = new Protector(swarmKeyBuffer)
+    const protectorB = new Protector(wrongSwarmKeyBuffer)
+
+    const [aToB, bToA] = await Promise.all([
+      protector.protect(inbound),
+      protectorB.protect(outbound)
+    ])
+
+    pipe(
+      [Buffer.from('hello world'), Buffer.from('doo dah')],
+      aToB
+    )
+
+    const output = await pipe(
+      bToA,
+      collect
+    )
+
+    expect(output).to.not.eql([Buffer.from('hello world'), Buffer.from('doo dah')])
+  })
+
+  describe('invalid psks', () => {
+    it('should not accept a bad psk', () => {
+      expect(() => {
+        return new Protector(Buffer.from('not-a-key'))
+      }).to.throw(Errors.INVALID_PSK)
+    })
+
+    it('should not accept a psk of incorrect length', () => {
+      expect(() => {
+        return new Protector(Buffer.from('/key/swarm/psk/1.0.0/\n/base16/\ndffb7e'))
+      }).to.throw(Errors.INVALID_PSK)
+    })
+  })
+})

test/upgrading/upgrader.spec.js

@@ -13,12 +13,14 @@ const pipe = require('it-pipe')
 const { collect } = require('streaming-iterables')
 const pSettle = require('p-settle')
 const Transport = require('libp2p-websockets')
+const Crypto = require('../../src/insecure/plaintext')
+const Protector = require('../../src/pnet')
+const swarmKeyBuffer = Buffer.from(require('../fixtures/swarm.key'))
 
 const Libp2p = require('../../src')
 const Upgrader = require('../../src/upgrader')
 const { codes } = require('../../src/errors')
 
-const mockCrypto = require('../utils/mockCrypto')
 const mockMultiaddrConnPair = require('../utils/mockMultiaddrConn')
 const Peers = require('../fixtures/peers')
 const addrs = [
@@ -63,7 +65,7 @@ describe('Upgrader', () => {
     sinon.stub(localUpgrader, 'muxers').value(muxers)
     sinon.stub(remoteUpgrader, 'muxers').value(muxers)
 
-    const cryptos = new Map([[mockCrypto.tag, mockCrypto]])
+    const cryptos = new Map([[Crypto.protocol, Crypto]])
     sinon.stub(localUpgrader, 'cryptos').value(cryptos)
     sinon.stub(remoteUpgrader, 'cryptos').value(cryptos)
 
@@ -85,7 +87,7 @@ describe('Upgrader', () => {
     sinon.stub(localUpgrader, 'muxers').value(muxers)
     sinon.stub(remoteUpgrader, 'muxers').value(muxers)
 
-    const cryptos = new Map([[mockCrypto.tag, mockCrypto]])
+    const cryptos = new Map([[Crypto.protocol, Crypto]])
     sinon.stub(localUpgrader, 'cryptos').value(cryptos)
     sinon.stub(remoteUpgrader, 'cryptos').value(cryptos)
 
@@ -114,6 +116,48 @@ describe('Upgrader', () => {
     expect(result).to.eql([hello])
   })
 
+  it('should use a private connection protector when provided', async () => {
+    const { inbound, outbound } = mockMultiaddrConnPair({ addrs, remotePeer })
+
+    const muxers = new Map([[Muxer.multicodec, Muxer]])
+    sinon.stub(localUpgrader, 'muxers').value(muxers)
+    sinon.stub(remoteUpgrader, 'muxers').value(muxers)
+
+    const cryptos = new Map([[Crypto.protocol, Crypto]])
+    sinon.stub(localUpgrader, 'cryptos').value(cryptos)
+    sinon.stub(remoteUpgrader, 'cryptos').value(cryptos)
+
+    const protector = new Protector(swarmKeyBuffer)
+    sinon.stub(localUpgrader, 'protector').value(protector)
+    sinon.stub(remoteUpgrader, 'protector').value(protector)
+    sinon.spy(protector, 'protect')
+
+    const connections = await Promise.all([
+      localUpgrader.upgradeOutbound(outbound),
+      remoteUpgrader.upgradeInbound(inbound)
+    ])
+
+    expect(connections).to.have.length(2)
+
+    const { stream, protocol } = await connections[0].newStream('/echo/1.0.0')
+    expect(protocol).to.equal('/echo/1.0.0')
+
+    const hello = Buffer.from('hello there!')
+    const result = await pipe(
+      [hello],
+      stream,
+      function toBuffer (source) {
+        return (async function * () {
+          for await (const val of source) yield val.slice()
+        })()
+      },
+      collect
+    )
+
+    expect(result).to.eql([hello])
+    expect(protector.protect.callCount).to.eql(2)
+  })
+
   it('should fail if crypto fails', async () => {
     const { inbound, outbound } = mockMultiaddrConnPair({ addrs, remotePeer })
 
@@ -153,7 +197,7 @@ describe('Upgrader', () => {
     sinon.stub(localUpgrader, 'muxers').value(muxersLocal)
     sinon.stub(remoteUpgrader, 'muxers').value(muxersRemote)
 
-    const cryptos = new Map([[mockCrypto.tag, mockCrypto]])
+    const cryptos = new Map([[Crypto.protocol, Crypto]])
     sinon.stub(localUpgrader, 'cryptos').value(cryptos)
     sinon.stub(remoteUpgrader, 'cryptos').value(cryptos)
 
@@ -178,7 +222,7 @@ describe('Upgrader', () => {
     sinon.stub(localUpgrader, 'muxers').value(muxers)
     sinon.stub(remoteUpgrader, 'muxers').value(muxers)
 
-    const cryptos = new Map([[mockCrypto.tag, mockCrypto]])
+    const cryptos = new Map([[Crypto.protocol, Crypto]])
     sinon.stub(localUpgrader, 'cryptos').value(cryptos)
     sinon.stub(remoteUpgrader, 'cryptos').value(cryptos)
 
@@ -212,7 +256,7 @@ describe('Upgrader', () => {
     sinon.stub(localUpgrader, 'muxers').value(muxers)
     sinon.stub(remoteUpgrader, 'muxers').value(muxers)
 
-    const cryptos = new Map([[mockCrypto.tag, mockCrypto]])
+    const cryptos = new Map([[Crypto.protocol, Crypto]])
     sinon.stub(localUpgrader, 'cryptos').value(cryptos)
     sinon.stub(remoteUpgrader, 'cryptos').value(cryptos)
 
@@ -246,7 +290,7 @@ describe('Upgrader', () => {
     sinon.stub(localUpgrader, 'muxers').value(muxers)
     sinon.stub(remoteUpgrader, 'muxers').value(muxers)
 
-    const cryptos = new Map([[mockCrypto.tag, mockCrypto]])
+    const cryptos = new Map([[Crypto.protocol, Crypto]])
     sinon.stub(localUpgrader, 'cryptos').value(cryptos)
     sinon.stub(remoteUpgrader, 'cryptos').value(cryptos)
 
@@ -288,18 +332,21 @@ describe('libp2p.upgrader', () => {
   })
 
   it('should create an Upgrader', () => {
+    const protector = new Protector(swarmKeyBuffer)
     libp2p = new Libp2p({
       peerInfo: peers[0],
       modules: {
         transport: [Transport],
         streamMuxer: [Muxer],
-        connEncryption: [mockCrypto]
+        connEncryption: [Crypto],
+        connProtector: protector
       }
     })
 
     expect(libp2p.upgrader).to.exist()
     expect(libp2p.upgrader.muxers).to.eql(new Map([[Muxer.multicodec, Muxer]]))
-    expect(libp2p.upgrader.cryptos).to.eql(new Map([[mockCrypto.tag, mockCrypto]]))
+    expect(libp2p.upgrader.cryptos).to.eql(new Map([[Crypto.protocol, Crypto]]))
+    expect(libp2p.upgrader.protector).to.equal(protector)
     // Ensure the transport manager also has the upgrader
     expect(libp2p.upgrader).to.equal(libp2p.transportManager.upgrader)
   })
@@ -310,7 +357,7 @@ describe('libp2p.upgrader', () => {
       modules: {
         transport: [Transport],
         streamMuxer: [Muxer],
-        connEncryption: [mockCrypto]
+        connEncryption: [Crypto]
       }
     })
 
@@ -335,14 +382,14 @@ describe('libp2p.upgrader', () => {
       modules: {
         transport: [Transport],
         streamMuxer: [Muxer],
-        connEncryption: [mockCrypto]
+        connEncryption: [Crypto]
       }
     })
 
     const remoteUpgrader = new Upgrader({
       localPeer: remotePeer.id,
       muxers: new Map([[Muxer.multicodec, Muxer]]),
-      cryptos: new Map([[mockCrypto.tag, mockCrypto]])
+      cryptos: new Map([[Crypto.protocol, Crypto]])
     })
 
     const { inbound, outbound } = mockMultiaddrConnPair({ addrs, remotePeer: remotePeer.id })

test/utils/mockCrypto.js

@@ -4,7 +4,7 @@ const PeerId = require('peer-id')
 const Peers = require('../fixtures/peers')
 
 module.exports = {
-  tag: '/insecure',
+  protocol: '/insecure',
   secureInbound: (localPeer, stream) => {
     return {
       conn: stream,