Merge pull request #67 from libp2p/secio

[WIP] Add secio & move to pull-streams

Author: daviddias

README.md

@@ -59,13 +59,13 @@ const sw = new Swarm(peerInfo)
 
 ## API
 
-peerInfo is a [PeerInfo](https://github.com/diasdavid/js-peer-info) object that represents the peer creating this swarm instance.
+peerInfo is a [PeerInfo](https://github.com/libp2p/js-peer-info) object that represents the peer creating this swarm instance.
 
 ### Transports
 
 ##### `swarm.transport.add(key, transport, options, callback)`
 
-libp2p-swarm expects transports that implement [interface-transport](https://github.com/diasdavid/abstract-transport). For example [libp2p-tcp](https://github.com/diasdavid/js-libp2p-tcp).
+libp2p-swarm expects transports that implement [interface-transport](https://github.com/libp2p/abstract-transport). For example [libp2p-tcp](https://github.com/libp2p/js-libp2p-tcp).
 
 - `key` - the transport identifier.
 - `transport` -
@@ -100,20 +100,31 @@ Close the listeners of a given transport.
 
 ##### `swarm.connection.addUpgrade()`
 
-A connection upgrade must be able to receive and return something that implements the [interface-connection](https://github.com/diasdavid/interface-connection) specification.
+A connection upgrade must be able to receive and return something that implements the [interface-connection](https://github.com/libp2p/interface-connection) specification.
 
 > **WIP**
 
 ##### `swarm.connection.addStreamMuxer(muxer)`
 
-Upgrading a connection to use a stream muxer is still considered an upgrade, but a special case since once this connection is applied, the returned obj will implement the [interface-stream-muxer](https://github.com/diasdavid/interface-stream-muxer) spec.
+Upgrading a connection to use a stream muxer is still considered an upgrade, but a special case since once this connection is applied, the returned obj will implement the [interface-stream-muxer](https://github.com/libp2p/interface-stream-muxer) spec.
 
 - `muxer`
 
 ##### `swarm.connection.reuse()`
 
 Enable the identify protocol.
 
+##### `swarm.connection.crypto([tag, encrypt])`
+
+Enable a specified crypto protocol. By default no encryption is used, aka `plaintext`. If called with no arguments it resets to use `plaintext`.
+
+You can use for example [libp2p-secio](https://github.com/libp2p/js-libp2p-secio) like this
+
+```js
+const secio = require('libp2p-secio')
+swarm.connection.crypto(secio.tag, secio.encrypt)
+```
+
 ### `swarm.dial(pi, protocol, callback)`
 
 dial uses the best transport (whatever works first, in the future we can have some criteria), and jump starts the connection until the point where we have to negotiate the protocol. If a muxer is available, then drop the muxer onto that connection. Good to warm up connections or to check for connectivity. If we have already a muxer for that peerInfo, then do nothing.
@@ -152,17 +163,43 @@ Close all the listeners and muxers.
 
 - `callback`
 
+### 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.
+
+
+
 ## Design
 
 ### Multitransport
 
 libp2p is designed to support multiple transports at the same time. While peers are identified by their ID (which are generated from their public keys), the addresses of each pair may vary, depending the device where they are being run or the network in which they are accessible through.
 
-In order for a transport to be supported, it has to follow the [interface-transport](https://github.com/diasdavid/interface-transport) spec.
+In order for a transport to be supported, it has to follow the [interface-transport](https://github.com/libp2p/interface-transport) spec.
 
 ### Connection upgrades
 
-Each connection in libp2p follows the [interface-connection](https://github.com/diasdavid/interface-connection) spec. This design decision enables libp2p to have upgradable transports.
+Each connection in libp2p follows the [interface-connection](https://github.com/libp2p/interface-connection) spec. This design decision enables libp2p to have upgradable transports.
 
 We think of `upgrade` as a very important notion when we are talking about connections, we can see mechanisms like: stream multiplexing, congestion control, encrypted channels, multipath, simulcast, etc, as `upgrades` to a connection. A connection can be a simple and with no guarantees, drop a packet on the network with a destination thing, a transport in the other hand can be a connection and or a set of different upgrades that are mounted on top of each other, giving extra functionality to that connection and therefore `upgrading` it.
 

gulpfile.js

@@ -4,6 +4,7 @@ const gulp = require('gulp')
 const PeerInfo = require('peer-info')
 const PeerId = require('peer-id')
 const WebSockets = require('libp2p-websockets')
+const pull = require('pull-stream')
 
 const Swarm = require('./src')
 const spdy = require('libp2p-spdy')
@@ -60,7 +61,7 @@ gulp.task('test:browser:before', (done) => {
   sigS = sigServer.start(15555, ready)
 
   function echo (conn) {
-    conn.pipe(conn)
+    pull(conn, conn)
   }
 })
 

package.json

@@ -37,37 +37,36 @@
     "node": "^4.3.0"
   },
   "devDependencies": {
-    "aegir": "^4.0.0",
+    "aegir": "^8.0.0",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
     "gulp": "^3.9.1",
-    "istanbul": "^0.4.3",
     "libp2p-multiplex": "^0.2.1",
-    "libp2p-spdy": "^0.8.1",
-    "libp2p-tcp": "^0.7.4",
-    "libp2p-webrtc-star": "^0.3.2",
-    "libp2p-websockets": "^0.7.1",
-    "pre-commit": "^1.1.2",
-    "stream-pair": "^1.0.3",
+    "libp2p-secio": "^0.4.2",
+    "libp2p-spdy": "^0.9.0",
+    "libp2p-tcp": "^0.8.1",
+    "libp2p-webrtc-star": "^0.4.4",
+    "libp2p-websockets": "^0.8.1",
+    "pre-commit": "^1.1.3",
+    "pull-goodbye": "0.0.1",
+    "pull-stream": "^3.4.5",
     "webrtcsupport": "^2.2.0"
   },
   "dependencies": {
-    "babel-runtime": "^6.6.1",
-    "bl": "^1.1.2",
+    "babel-runtime": "^6.11.6",
     "browserify-zlib": "github:ipfs/browserify-zlib",
     "debug": "^2.2.0",
-    "duplexify": "^3.4.3",
-    "interface-connection": "^0.1.7",
-    "ip-address": "^5.8.0",
-    "length-prefixed-stream": "^1.5.0",
-    "libp2p-identify": "^0.1.3",
+    "interface-connection": "^0.2.1",
+    "ip-address": "^5.8.2",
+    "libp2p-identify": "^0.2.0",
     "lodash.contains": "^2.4.3",
-    "multiaddr": "^2.0.0",
-    "multistream-select": "^0.9.0",
+    "multiaddr": "^2.0.2",
+    "multistream-select": "^0.11.0",
     "peer-id": "^0.7.0",
     "peer-info": "^0.7.0",
     "protocol-buffers": "^3.1.6",
-    "run-parallel": "^1.1.6"
+    "run-parallel": "^1.1.6",
+    "run-waterfall": "^1.1.3"
   },
   "contributors": [
     "David Dias <[email protected]>",
@@ -77,4 +76,4 @@
     "Richard Littauer <[email protected]>",
     "dignifiedquire <[email protected]>"
   ]
-}
+}

src/connection.js

@@ -1,8 +1,13 @@
 'use strict'
 
-const protocolMuxer = require('./protocol-muxer')
 const identify = require('libp2p-identify')
 const multistream = require('multistream-select')
+const waterfall = require('run-waterfall')
+const debug = require('debug')
+const log = debug('libp2p:swarm:connection')
+
+const protocolMuxer = require('./protocol-muxer')
+const plaintext = require('./plaintext')
 
 module.exports = function connection (swarm) {
   return {
@@ -14,7 +19,7 @@ module.exports = function connection (swarm) {
 
       // for listening
       swarm.handle(muxer.multicodec, (conn) => {
-        const muxedConn = muxer(conn, true)
+        const muxedConn = muxer.listener(conn)
 
         muxedConn.on('stream', (conn) => {
           protocolMuxer(swarm.protocols, conn)
@@ -29,28 +34,23 @@ module.exports = function connection (swarm) {
           conn.getPeerInfo = (cb) => {
             const conn = muxedConn.newStream()
             const ms = new multistream.Dialer()
-            ms.handle(conn, (err) => {
-              if (err) { return cb(err) }
-
-              ms.select(identify.multicodec, (err, conn) => {
-                if (err) { return cb(err) }
 
-                identify.exec(conn, (err, peerInfo, observedAddrs) => {
-                  if (err) { return cb(err) }
-
-                  observedAddrs.forEach((oa) => {
-                    swarm._peerInfo.multiaddr.addSafe(oa)
-                  })
-
-                  cb(null, peerInfo)
+            waterfall([
+              (cb) => ms.handle(conn, cb),
+              (cb) => ms.select(identify.multicodec, cb),
+              (conn, cb) => identify.dialer(conn, cb),
+              (peerInfo, observedAddrs, cb) => {
+                observedAddrs.forEach((oa) => {
+                  swarm._peerInfo.multiaddr.addSafe(oa)
                 })
-              })
-            })
+                cb(null, peerInfo)
+              }
+            ], cb)
           }
 
           conn.getPeerInfo((err, peerInfo) => {
             if (err) {
-              return console.log('Identify not successful')
+              return log('Identify not successful')
             }
             swarm.muxedConns[peerInfo.id.toB58String()] = {
               muxer: muxedConn
@@ -63,12 +63,33 @@ module.exports = function connection (swarm) {
             })
           })
         }
+
+        return conn
       })
     },
 
     reuse () {
       swarm.identify = true
-      swarm.handle(identify.multicodec, identify.handler(swarm._peerInfo))
+      swarm.handle(identify.multicodec, (conn) => {
+        identify.listener(conn, swarm._peerInfo)
+      })
+    },
+
+    crypto (tag, encrypt) {
+      if (!tag && !encrypt) {
+        tag = plaintext.tag
+        encrypt = plaintext.encrypt
+      }
+
+      swarm.unhandle(swarm.crypto.tag)
+      swarm.handle(tag, (conn) => {
+        const id = swarm._peerInfo.id
+        const secure = encrypt(id, id.privKey, conn)
+
+        protocolMuxer(swarm.protocols, secure)
+      })
+
+      swarm.crypto = {tag, encrypt}
     }
   }
 }

src/dial.js

@@ -2,6 +2,8 @@
 
 const multistream = require('multistream-select')
 const Connection = require('interface-connection').Connection
+const debug = require('debug')
+const log = debug('libp2p:swarm:dial')
 
 const protocolMuxer = require('./protocol-muxer')
 
@@ -19,6 +21,7 @@ module.exports = function dial (swarm) {
     const proxyConn = new Connection()
 
     const b58Id = pi.id.toB58String()
+    log('dialing %s', b58Id)
 
     if (!swarm.muxedConns[b58Id]) {
       if (!swarm.conns[b58Id]) {
@@ -44,7 +47,6 @@ module.exports = function dial (swarm) {
 
     function gotWarmedUpConn (conn) {
       conn.setPeerInfo(pi)
-
       attemptMuxerUpgrade(conn, (err, muxer) => {
         if (!protocol) {
           if (err) {
@@ -97,13 +99,22 @@ module.exports = function dial (swarm) {
           cryptoDial()
 
           function cryptoDial () {
-            // currently, no crypto channel is implemented
             const ms = new multistream.Dialer()
             ms.handle(conn, (err) => {
               if (err) {
                 return cb(err)
               }
-              ms.select('/plaintext/1.0.0', cb)
+
+              const id = swarm._peerInfo.id
+              log('selecting crypto: %s', swarm.crypto.tag)
+              ms.select(swarm.crypto.tag, (err, conn) => {
+                if (err) {
+                  return cb(err)
+                }
+
+                const wrapped = swarm.crypto.encrypt(id, id.privKey, conn)
+                cb(null, wrapped)
+              })
             })
           }
         })
@@ -129,6 +140,7 @@ module.exports = function dial (swarm) {
           if (err) {
             return callback(new Error('multistream not supported'))
           }
+          log('selecting %s', key)
           ms.select(key, (err, conn) => {
             if (err) {
               if (muxers.length === 0) {
@@ -139,7 +151,7 @@ module.exports = function dial (swarm) {
               return
             }
 
-            const muxedConn = swarm.muxers[key](conn, false)
+            const muxedConn = swarm.muxers[key].dialer(conn)
             swarm.muxedConns[b58Id] = {}
             swarm.muxedConns[b58Id].muxer = muxedConn
             // should not be needed anymore - swarm.muxedConns[b58Id].conn = conn

src/index.js

@@ -9,6 +9,7 @@ const transport = require('./transport')
 const connection = require('./connection')
 const dial = require('./dial')
 const protocolMuxer = require('./protocol-muxer')
+const plaintext = require('./plaintext')
 
 exports = module.exports = Swarm
 
@@ -50,6 +51,9 @@ function Swarm (peerInfo) {
   // is the Identify protocol enabled?
   this.identify = false
 
+  // Crypto details
+  this.crypto = plaintext
+
   this.transport = transport(this)
   this.connection = connection(this)
 
@@ -90,12 +94,13 @@ function Swarm (peerInfo) {
     this.protocols[protocol] = handler
   }
 
-  // our crypto handshake :)
-  this.handle('/plaintext/1.0.0', (conn) => {
-    protocolMuxer(this.protocols, conn)
+  this.handle(this.crypto.tag, (conn) => {
+    const id = this._peerInfo.id
+    const wrapped = this.crypto.encrypt(id, id.privKey, conn)
+    return protocolMuxer(this.protocols, wrapped)
   })
 
-  this.unhandle = (protocol, handler) => {
+  this.unhandle = (protocol) => {
     if (this.protocols[protocol]) {
       delete this.protocols[protocol]
     }
@@ -122,14 +127,13 @@ function Swarm (peerInfo) {
 
     const transports = this.transports
 
-    parallel(Object.keys(transports).map((key) => {
-      return (cb) => {
+    parallel(
+      Object.keys(transports).map((key) => (cb) => {
         parallel(transports[key].listeners.map((listener) => {
-          return (cb) => {
-            listener.close(cb)
-          }
+          return (cb) => listener.close(cb)
         }), cb)
-      }
-    }), callback)
+      }),
+      callback
+    )
   }
 }