[WIP] TLS transport upgrade

Cargo.toml

@@ -29,6 +29,7 @@ libp2p-ratelimit = { version = "0.8.0", path = "./transports/ratelimit" }
 libp2p-core = { version = "0.8.1", path = "./core" }
 libp2p-core-derive = { version = "0.8.0", path = "./misc/core-derive" }
 libp2p-secio = { version = "0.8.0", path = "./protocols/secio", default-features = false }
+libp2p-tls = { version = "0.1.0", path = "./protocols/tls" }
 libp2p-uds = { version = "0.8.0", path = "./transports/uds" }
 libp2p-wasm-ext = { version = "0.1.0", path = "./transports/wasm-ext" }
 libp2p-websocket = { version = "0.8.0", path = "./transports/websocket", optional = true }

core/src/lib.rs

@@ -84,7 +84,7 @@ pub use muxing::StreamMuxer;
 pub use nodes::raw_swarm::ConnectedPoint;
 pub use peer_id::PeerId;
 pub use protocols_handler::{ProtocolsHandler, ProtocolsHandlerEvent};
-pub use identity::PublicKey;
+pub use identity::{PublicKey, Keypair};
 pub use swarm::Swarm;
 pub use transport::Transport;
 pub use translation::address_translation;

examples/chat.rs

@@ -67,7 +67,8 @@ fn main() {
     println!("Local peer id: {:?}", local_peer_id);
 
     // Set up a an encrypted DNS-enabled TCP Transport over the Mplex and Yamux protocols
-    let transport = libp2p::build_development_transport(local_key);
+    let transport =
+        libp2p::build_tcp_ws_tls_mplex_yamux(local_key);
 
     // Create a Floodsub topic
     let floodsub_topic = libp2p::floodsub::TopicBuilder::new("chat").build();
@@ -145,7 +146,10 @@ fn main() {
     tokio::run(futures::future::poll_fn(move || -> Result<_, ()> {
         loop {
             match framed_stdin.poll().expect("Error while polling stdin") {
-                Async::Ready(Some(line)) => swarm.floodsub.publish(&floodsub_topic, line.as_bytes()),
+                Async::Ready(Some(line)) => {
+                    println!("stdin: {}", line);
+                    swarm.floodsub.publish(&floodsub_topic, line.as_bytes())
+                },
                 Async::Ready(None) => panic!("Stdin closed"),
                 Async::NotReady => break,
             };

protocols/secio/src/lib.rs

@@ -251,6 +251,7 @@ where
 
     #[inline]
     fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        println!("{}", item.len());
         self.inner.start_send(item)
     }
 

protocols/tls/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "libp2p-tls"
+edition = "2018"
+description = "TLS encryption protocol for libp2p"
+version = "0.1.0"
+authors = ["Grant Wuerker <[email protected]>"]
+license = "MIT"
+repository = "https://github.com/libp2p/rust-libp2p"
+keywords = ["peer-to-peer", "libp2p", "networking"]
+categories = ["network-programming", "asynchronous"]
+
+[dependencies]
+futures = "0.1"
+libp2p-core = { version = "0.8.0", path = "../../core" }
+rw-stream-sink = { version = "0.1.1", path = "../../misc/rw-stream-sink" }
+tokio-io = "0.1.0"
+bytes = "0.4"
+log = "0.4.6"
+openssl = "0.10"
+rustls = { version = "0.15", features = ["dangerous_configuration"] }
+rand = "0.6"
+quinn-proto = "0.2"
+simple_asn1 = "0.3"
+webpki = "0.19"
+tokio-threadpool = "0.1"

protocols/tls/src/codec/decode.rs

@@ -0,0 +1,43 @@
+use futures::prelude::*;
+use bytes::BytesMut;
+use crate::error::TlsError;
+
+pub struct DecoderMiddleware<S> {
+    pub raw_stream: S,
+}
+
+impl<S> Sink for DecoderMiddleware<S>
+where
+    S: Sink,
+{
+    type SinkItem = S::SinkItem;
+    type SinkError = S::SinkError;
+
+    #[inline]
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        self.raw_stream.start_send(item)
+    }
+
+    #[inline]
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        self.raw_stream.poll_complete()
+    }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.raw_stream.close()
+    }
+}
+
+impl<S> Stream for DecoderMiddleware<S>
+where
+    S: Stream,
+{
+    type Item = S::Item;
+    type Error = S::Error;
+
+    #[inline]
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.raw_stream.poll()
+    }
+}

protocols/tls/src/codec/encode.rs

@@ -0,0 +1,42 @@
+use futures::prelude::*;
+use bytes::BytesMut;
+use rustls::ServerSession;
+
+pub struct EncoderMiddleware<S> {
+    pub raw_sink: S,
+    tls_session: ServerSession,
+
+}
+
+impl<S> Sink for EncoderMiddleware<S>
+where
+    S: Sink,
+{
+    type SinkItem = S::SinkItem;
+    type SinkError = S::SinkError;
+
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        self.raw_sink.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Result<Async<()>, Self::SinkError> {
+        self.raw_sink.poll_complete()
+    }
+
+    fn close(&mut self) -> Result<Async<()>, Self::SinkError> {
+        self.raw_sink.close()
+    }
+}
+
+impl<S> Stream for EncoderMiddleware<S>
+where
+    S: Stream,
+{
+    type Item = S::Item;
+    type Error = S::Error;
+
+    #[inline]
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.raw_sink.poll()
+    }
+}

protocols/tls/src/codec/mod.rs

@@ -0,0 +1,111 @@
+use tokio_io::{AsyncRead, AsyncWrite};
+use futures::{Sink, StartSend, Poll, Stream, Async, AsyncSink};
+use rustls::TLSError;
+use libp2p_core::Negotiated;
+use bytes::BufMut;
+use std::io::{Error as IoError, ErrorKind as IoErrorKind, Write};
+use rustls::{RootCertStore, Session, NoClientAuth, AllowAnyAuthenticatedClient,
+             AllowAnyAnonymousOrAuthenticatedClient};
+use std::io::Read;
+use tokio_threadpool::blocking;
+
+pub struct FullCodec<S>
+    where
+        S: AsyncRead + AsyncWrite + Send + 'static,
+{
+    server_session: Option<rustls::ServerSession>,
+    client_session: Option<rustls::ClientSession>,
+    socket: S,
+}
+
+impl<S> FullCodec<S>
+    where
+        S: AsyncRead + AsyncWrite + Send + 'static,
+{
+    pub fn from_server(mut socket: S, mut session: rustls::ServerSession) -> Self {
+        FullCodec {
+            server_session: Some(session),
+            client_session: None,
+            socket,
+        }
+    }
+
+    pub fn from_client(mut socket: S, mut session: rustls::ClientSession) -> Self {
+        let client_stream = rustls::Stream::new(&mut session, &mut socket);
+        FullCodec {
+            server_session: None,
+            client_session: Some(session),
+            socket,
+        }
+    }
+
+    fn as_server_stream(&mut self) -> Option<rustls::Stream<rustls::ServerSession, S>> {
+        if let Some(ref mut server_session) = self.server_session {
+            return Some(rustls::Stream::new(server_session, &mut self.socket));
+        }
+        None
+    }
+
+    fn as_client_stream(&mut self) -> Option<rustls::Stream<rustls::ClientSession, S>> {
+        if let Some(ref mut client_session) = self.client_session {
+            return Some(rustls::Stream::new(client_session, &mut self.socket));
+        }
+        None
+    }
+}
+
+impl<S> Sink for FullCodec<S>
+    where
+        S: AsyncRead + AsyncWrite + Send + 'static,
+{
+    type SinkItem = Vec<u8>;
+    type SinkError = IoError;
+
+    #[inline]
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        println!("sending tls...");
+        if let Some(ref mut stream) = self.as_server_stream() {
+            let n = stream.write(&item).unwrap();
+            println!("sent {}", n);
+        } else if let Some(ref mut stream) = self.as_client_stream() {
+            let n = stream.write(&item).unwrap();
+            println!("sent {}", n);
+        }
+
+        Ok(AsyncSink::Ready)
+    }
+
+    #[inline]
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(Async::Ready(()))
+    }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(Async::Ready(()))
+    }
+}
+
+impl<S> Stream for FullCodec<S>
+    where
+        S: AsyncRead + AsyncWrite + Send,
+{
+    type Item = Vec<u8>;
+    type Error = IoError;
+
+    #[inline]
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        println!("reading tls...");
+        let mut plaintext = Vec::new();
+        if let Some(ref mut stream) = self.as_server_stream() {
+            let n = stream.read_to_end(&mut plaintext).unwrap();
+            println!("read {}", n);
+        } else if let Some(ref mut stream) = self.as_client_stream() {
+            let n = stream.read_to_end(&mut plaintext).unwrap();
+            println!("read {}", n);
+        }
+        Ok(Async::Ready(Some(plaintext)))
+    }
+}
+
+

protocols/tls/src/error.rs

@@ -0,0 +1,36 @@
+//! Defines the `TlsError` enum that groups all possible errors in TLS.
+
+use std::error;
+use std::io::Error as IoError;
+use std::fmt;
+
+#[derive(Debug)]
+pub enum TlsError {
+    IoError(IoError)
+}
+
+impl error::Error for TlsError {
+    fn cause(&self) -> Option<&dyn error::Error> {
+        match *self {
+            TlsError::IoError(ref err) => Some(err),
+            _ => None,
+        }
+    }
+}
+
+impl From<IoError> for TlsError {
+    #[inline]
+    fn from(err: IoError) -> TlsError {
+        TlsError::IoError(err)
+    }
+}
+
+impl fmt::Display for TlsError {
+    #[inline]
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
+        match self {
+            TlsError::IoError(e) =>
+                write!(f, "I/O error: {}", e),
+        }
+    }
+}