[WIP] Add experimental QUIC support.

Cargo.toml

@@ -39,14 +39,18 @@ tokio-io = "0.1"
 [target.'cfg(not(any(target_os = "emscripten", target_os = "unknown")))'.dependencies]
 libp2p-dns = { version = "0.2.0", path = "./transports/dns" }
 libp2p-mdns = { version = "0.2.0", path = "./misc/mdns" }
+libp2p-quic = { version = "0.1.0", path = "./transports/quic" }
 libp2p-tcp = { version = "0.2.0", path = "./transports/tcp" }
 
 [target.'cfg(any(target_os = "emscripten", target_os = "unknown"))'.dependencies]
 stdweb = { version = "0.4", default-features = false }
 
 [dev-dependencies]
 env_logger = "0.6.0"
+openssl = "0.10"
+quicli = "0.4"
 rand = "0.6"
+structopt = "0.2"
 tokio = "0.1"
 tokio-stdin-stdout = "0.1"
 void = "1.0"
@@ -71,6 +75,7 @@ members = [
     "protocols/plaintext",
     "protocols/secio",
     "transports/dns",
+    "transports/quic",
     "transports/ratelimit",
     "transports/tcp",
     "transports/uds",

core/src/lib.rs

@@ -87,7 +87,7 @@ pub use self::peer_id::PeerId;
 pub use self::protocols_handler::{ProtocolsHandler, ProtocolsHandlerEvent};
 pub use self::public_key::PublicKey;
 pub use self::swarm::Swarm;
-pub use self::transport::Transport;
+pub use self::transport::{Transport, TransportError};
 pub use self::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo, UpgradeError, ProtocolName};
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]

examples/chat-quic.rs

@@ -0,0 +1,164 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use env_logger;
+use futures::prelude::*;
+use libp2p::{self, NetworkBehaviour, core::PublicKey, tokio_codec::{FramedRead, LinesCodec}};
+use openssl::{rsa::Rsa, x509::X509};
+use quicli::prelude::*;
+use structopt::StructOpt;
+use std::io;
+
+// Brief usage instructions:
+//
+// 1. Generate a private RSA key and self-signed certificate for the listener:
+//     $ openssl req -newkey rsa:4096 -nodes -keyout serverkey.pem -x509 -out servercert.pem
+//
+// 2. Generate a private RSA key and self-signed certificate for the dialer:
+//     $ openssl req -newkey rsa:4096 -nodes -keyout clientkey.pem -x509 -out clientcert.pem
+//
+// 3. Get the PeerId of the listener:
+//     $ cargo run --example chat-quic -- peer -c servercert.pem
+//
+// 4. Start the listener:
+//     $ cargo run --example chat-quic -- run -k serverkey.pem -c servercert.pem
+//
+// 5. Notice the listen port in the output of 4. and start the dialer by using the peer ID from 3.
+//     $ cargo run --example chat-quic -- run -k clientkey.pem -c clientcert.pem -d "/ip4/127.0.0.1/udp/<port>/quic/p2p/<peer-id>"
+
+#[derive(Debug, StructOpt)]
+enum Cli {
+    #[structopt(name = "peer")]
+    PeerId(PeerId),
+    #[structopt(name = "run")]
+    Run(Run)
+}
+
+#[derive(Debug, StructOpt)]
+struct PeerId {
+    #[structopt(long = "cert", short = "c")]
+    cert: String
+}
+
+#[derive(Debug, StructOpt)]
+struct Run {
+    #[structopt(long = "key", short = "k")]
+    key: String,
+
+    #[structopt(long = "cert", short = "c")]
+    cert: String,
+
+    #[structopt(long = "dial", short = "d")]
+    dial: Option<String>
+}
+
+fn main() -> CliResult {
+    env_logger::init();
+
+    match Cli::from_args() {
+        Cli::PeerId(args) => peerid(args),
+        Cli::Run(args) => run(args)
+    }
+}
+
+fn peerid(args: PeerId) -> CliResult {
+    let certificate = X509::from_pem(read_file(&args.cert)?.as_bytes())?;
+    let public_key = PublicKey::Rsa(certificate.public_key()?.public_key_to_der()?);
+    println!("Peer ID: {}", public_key.into_peer_id().to_base58());
+    Ok(())
+}
+
+fn run(args: Run) -> CliResult {
+    let private_key = Rsa::private_key_from_pem(read_file(&args.key)?.as_bytes())?;
+    let certificate = X509::from_pem(read_file(&args.cert)?.as_bytes())?;
+    let public_key = PublicKey::Rsa(certificate.public_key()?.public_key_to_der()?);
+
+    let rt = tokio::runtime::Runtime::new()?;
+
+    let transport = libp2p_quic::QuicConfig::new(rt.executor(), &private_key, &certificate)?;
+
+    let floodsub_topic = libp2p::floodsub::TopicBuilder::new("chat").build();
+
+    #[derive(NetworkBehaviour)]
+    struct MyBehaviour<TSubstream: libp2p::tokio_io::AsyncRead + libp2p::tokio_io::AsyncWrite> {
+        floodsub: libp2p::floodsub::Floodsub<TSubstream>
+    }
+
+    impl<TSubstream: libp2p::tokio_io::AsyncRead + libp2p::tokio_io::AsyncWrite> libp2p::core::swarm::NetworkBehaviourEventProcess<libp2p::floodsub::FloodsubEvent> for MyBehaviour<TSubstream> {
+        // Called when `floodsub` produces an event.
+        fn inject_event(&mut self, message: libp2p::floodsub::FloodsubEvent) {
+            if let libp2p::floodsub::FloodsubEvent::Message(message) = message {
+                println!("Received: '{:?}' from {:?}", String::from_utf8_lossy(&message.data), message.source);
+            }
+        }
+    }
+
+    // Create a Swarm to manage peers and events
+    let mut swarm = {
+        let mut behaviour = MyBehaviour {
+            floodsub: libp2p::floodsub::Floodsub::new(public_key.clone().into_peer_id())
+        };
+        behaviour.floodsub.subscribe(floodsub_topic.clone());
+        libp2p::Swarm::new(transport, behaviour, libp2p::core::topology::MemoryTopology::empty(public_key))
+    };
+
+    let addr = libp2p::Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/udp/0/quic".parse()?)?;
+    println!("Listening on {:?}", addr);
+
+    // Reach out to another node if specified
+    if let Some(to_dial) = args.dial {
+        match to_dial.parse() {
+            Ok(addr) => {
+                match libp2p::Swarm::dial_addr(&mut swarm, addr) {
+                    Ok(_) => println!("Dialed {:?}", to_dial),
+                    Err(e) => println!("Dial {:?} failed: {:?}", to_dial, e)
+                }
+            }
+            Err(err) => println!("Failed to parse address to dial: {:?}", err)
+        }
+    }
+
+    // Read full lines from stdin
+    let stdin = tokio_stdin_stdout::stdin(0);
+    let mut framed_stdin = FramedRead::new(stdin, LinesCodec::new());
+
+    // Kick it off
+    rt.block_on_all(futures::future::poll_fn(move || {
+        loop {
+            match framed_stdin.poll()? {
+                Async::Ready(Some(line)) => swarm.floodsub.publish(&floodsub_topic, line.as_bytes()),
+                Async::Ready(None) => return Err(io::Error::new(io::ErrorKind::Other, "stdin closed")),
+                Async::NotReady => break
+            };
+        }
+
+        loop {
+            match swarm.poll()? {
+                Async::Ready(Some(_)) => {}
+                Async::Ready(None) | Async::NotReady => break
+            }
+        }
+
+        Ok(Async::NotReady)
+    }))?;
+
+    Ok(())
+}
+

src/lib.rs

@@ -162,6 +162,9 @@ pub use libp2p_mdns as mdns;
 pub use libp2p_ping as ping;
 #[doc(inline)]
 pub use libp2p_plaintext as plaintext;
+#[cfg(not(any(target_os = "emscripten", target_os = "unknown")))]
+#[doc(inline)]
+pub use libp2p_quic as quic;
 #[doc(inline)]
 pub use libp2p_ratelimit as ratelimit;
 #[doc(inline)]

transports/quic/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "libp2p-quic"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+license = "MIT"
+edition = "2018"
+
+[dependencies]
+bytes = "0.4"
+failure = "0.1"
+libp2p-core = { path = "../../core" }
+log = "0.4"
+fnv = "1.0"
+futures = "0.1"
+multiaddr = { package = "parity-multiaddr", version = "0.1.0", path = "../../misc/multiaddr" }
+multihash = { package = "parity-multihash", version = "0.1.0", path = "../../misc/multihash" }
+openssl = "0.10"
+parking_lot = "0.5"
+picoquic = { git = "https://github.com/bkchr/picoquic-rs", rev = "60016b1be4c417a9c4e357201afdd3c42ca31e82" }
+tokio-executor = "0.1"

transports/quic/src/error.rs

@@ -0,0 +1,78 @@
+use failure::Fail;
+use openssl::error::ErrorStack;
+use picoquic;
+use std::{error::Error, fmt, io};
+
+#[derive(Debug)]
+pub struct QuicError {
+    kind: ErrorKind,
+    source: Option<Box<dyn Error + Send + Sync>>
+}
+
+#[derive(Debug)]
+pub enum ErrorKind {
+    Io,
+    PicoQuic,
+    OpenSsl,
+    InvalidPeerId,
+    PeerIdMismatch,
+    #[doc(hidden)]
+    __Nonexhaustive
+}
+
+impl Into<QuicError> for ErrorKind {
+    fn into(self) -> QuicError {
+        QuicError { kind: self, source: None }
+    }
+}
+
+impl fmt::Display for QuicError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self.kind {
+            ErrorKind::Io => write!(f, "i/o: {:?}", self.source),
+            ErrorKind::PicoQuic => write!(f, "picoquic: {:?}", self.source),
+            ErrorKind::OpenSsl => write!(f, "openssl: {:?}", self.source),
+            ErrorKind::InvalidPeerId => f.write_str("invalid peer ID"),
+            ErrorKind::PeerIdMismatch => f.write_str("peer IDs do not match"),
+            ErrorKind::__Nonexhaustive => f.write_str("__Nonexhaustive")
+        }
+    }
+}
+
+impl Error for QuicError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        if let Some(ref s) = self.source {
+            Some(s.as_ref())
+        } else {
+            None
+        }
+    }
+}
+
+impl From<io::Error> for QuicError {
+    fn from(e: io::Error) -> Self {
+        QuicError {
+            kind: ErrorKind::Io,
+            source: Some(Box::new(e))
+        }
+    }
+}
+
+impl From<ErrorStack> for QuicError {
+    fn from(e: ErrorStack) -> Self {
+        QuicError {
+            kind: ErrorKind::OpenSsl,
+            source: Some(Box::new(e))
+        }
+    }
+}
+
+impl From<picoquic::Error> for QuicError {
+    fn from(e: picoquic::Error) -> Self {
+        QuicError {
+            kind: ErrorKind::PicoQuic,
+            source: Some(Box::new(e.compat()))
+        }
+    }
+}
+