net/transport: Implement listeners for TCP and TLS + integration tests.

Cargo.lock

@@ -1459,6 +1459,12 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "ct-codecs"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f3b7eb4404b8195a9abb6356f4ac07d8ba267045c8d6d220ac4dc992e6cc75df"
+
 [[package]]
 name = "ctor"
 version = "0.1.22"
@@ -1564,6 +1570,7 @@ dependencies = [
  "darkfi-derive-internal",
  "dirs 4.0.0",
  "drk-sdk",
+ "ed25519-compact",
  "fast-socks5",
  "futures",
  "futures-rustls",
@@ -1583,7 +1590,9 @@ dependencies = [
  "num-bigint",
  "pasta_curves",
  "rand 0.8.5",
+ "rcgen",
  "rocksdb",
+ "rustls-pemfile 0.3.0",
  "serde",
  "serde_json",
  "sha2 0.10.2",
@@ -2000,6 +2009,16 @@ dependencies = [
  "signature",
 ]
 
+[[package]]
+name = "ed25519-compact"
+version = "1.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "24e1f30f0312ac83726c1197abeacd91c9557f8a623e904a009ae6bc529ae8d8"
+dependencies = [
+ "ct-codecs",
+ "getrandom 0.2.5",
+]
+
 [[package]]
 name = "ed25519-dalek"
 version = "1.0.1"
@@ -3770,6 +3789,15 @@ version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
 
+[[package]]
+name = "pem"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e9a3b09a20e374558580a4914d3b7d89bd61b954a5a5e1dcbea98753addb1947"
+dependencies = [
+ "base64 0.13.0",
+]
+
 [[package]]
 name = "percent-encoding"
 version = "2.1.0"
@@ -4358,6 +4386,18 @@ dependencies = [
  "num_cpus",
 ]
 
+[[package]]
+name = "rcgen"
+version = "0.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d7fa2d386df8533b02184941c76ae2e0d0c1d053f5d43339169d80f21275fc5e"
+dependencies = [
+ "pem",
+ "ring",
+ "time 0.3.9",
+ "yasna",
+]
+
 [[package]]
 name = "rdrand"
 version = "0.4.0"
@@ -7138,6 +7178,15 @@ dependencies = [
  "linked-hash-map",
 ]
 
+[[package]]
+name = "yasna"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "346d34a236c9d3e5f3b9b74563f238f955bbd05fa0b8b4efa53c130c43982f4c"
+dependencies = [
+ "time 0.3.9",
+]
+
 [[package]]
 name = "zeroize"
 version = "1.3.0"

Cargo.toml

@@ -53,6 +53,11 @@ native-tls = {version = "0.2.8", optional = true}
 socket2 = {version = "0.4.4", optional = true}
 futures-rustls = {version = "0.22.1", features = ["dangerous_configuration"], optional = true}
 
+# TLS cert utilities
+ed25519-compact = {version = "1.0.11", features = ["pem"], optional = true}
+rcgen = {version = "0.9.2", features = ["pem"], optional = true}
+rustls-pemfile = {version = "0.3.0", optional = true}
+
 # Encoding
 hex = {version = "0.4.3", optional = true}
 bs58 = {version = "0.4.0", optional = true}
@@ -206,6 +211,9 @@ net = [
     "fxhash",
     "socket2",
     "futures-rustls",
+    "ed25519-compact",
+    "rcgen",
+    "rustls-pemfile",
 
     "util",
     "system",
@@ -272,11 +280,6 @@ name = "net"
 path = "example/net.rs"
 required-features = ["async-runtime", "net"]
 
-[[example]]
-name = "net2"
-path = "example/net2.rs"
-required-features = ["async-runtime", "net"]
-
 [[example]]
 name = "tx"
 path = "example/tx.rs"

example/net2.rs

@@ -1,26 +0,0 @@
-use async_std::io::{ReadExt, WriteExt};
-use darkfi::net::transport::{TcpTransport, TlsTransport, Transport};
-use std::{fs::File, io::Write};
-use url::Url;
-
-#[async_std::main]
-async fn main() {
-    // nc -l 127.0.0.1 5432
-    // let tcp = TcpTransport { ttl: None };
-    // let url = Url::parse("tcp://127.0.0.1:5432").unwrap();
-
-    // let mut socket = tcp.dial(url).unwrap().await.unwrap();
-    // socket.write_all(b"ohai tcp\n").await.unwrap();
-    // socket.flush().await;
-
-    let tls = TlsTransport { ttl: None };
-    let url = Url::parse("tls://parazyd.org:70").unwrap();
-    let mut socket = tls.dial(url).unwrap().await.unwrap();
-    socket.write_all(b"/rms.png\r\n").await.unwrap();
-
-    let mut buf = vec![];
-    socket.read_to_end(&mut buf).await.unwrap();
-
-    let mut file = File::create("rms.png").unwrap();
-    file.write_all(&buf).unwrap();
-}

src/net/transport.rs

@@ -10,9 +10,17 @@ pub use tcp::TcpTransport;
 pub use tls::TlsTransport;
 
 pub trait Transport {
-    type Output;
+    type Acceptor;
+    type Connector;
+
     type Error: Error;
-    type Dial: Future<Output = Result<Self::Output, Self::Error>>;
+
+    type Listener: Future<Output = Result<Self::Acceptor, Self::Error>>;
+    type Dial: Future<Output = Result<Self::Connector, Self::Error>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>>
+    where
+        Self: Sized;
 
     fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>>
     where

src/net/transport/tcp.rs

@@ -1,6 +1,6 @@
 use std::{io, net::SocketAddr, pin::Pin};
 
-use async_std::net::TcpStream;
+use async_std::net::{TcpListener, TcpStream};
 use futures::prelude::*;
 use log::debug;
 use socket2::{Domain, Socket, Type};
@@ -8,14 +8,29 @@ use url::Url;
 
 use super::{Transport, TransportError};
 
+#[derive(Clone)]
 pub struct TcpTransport {
     pub ttl: Option<u32>,
 }
 
 impl Transport for TcpTransport {
-    type Output = TcpStream;
+    type Acceptor = TcpListener;
+    type Connector = TcpStream;
+
     type Error = io::Error;
-    type Dial = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
+
+    type Listener = Pin<Box<dyn Future<Output = Result<Self::Acceptor, Self::Error>> + Send>>;
+    type Dial = Pin<Box<dyn Future<Output = Result<Self::Connector, Self::Error>> + Send>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>> {
+        if url.scheme() != "tcp" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        debug!(target: "tcptransport", "listening on {}", socket_addr);
+        Ok(Box::pin(self.do_listen(socket_addr)))
+    }
 
     fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>> {
         if url.scheme() != "tcp" {
@@ -44,6 +59,15 @@ impl TcpTransport {
         Ok(socket)
     }
 
+    async fn do_listen(self, socket_addr: SocketAddr) -> Result<TcpListener, io::Error> {
+        let socket = self.create_socket(socket_addr)?;
+        socket.bind(&socket_addr.into())?;
+        // TODO: make backlog configurable
+        socket.listen(1024)?;
+        socket.set_nonblocking(true)?;
+        Ok(TcpListener::from(std::net::TcpListener::from(socket)))
+    }
+
     async fn do_dial(self, socket_addr: SocketAddr) -> Result<TcpStream, io::Error> {
         let socket = self.create_socket(socket_addr)?;
         socket.set_nonblocking(true)?;

src/net/transport/tls.rs

@@ -1,18 +1,20 @@
 use std::{io, net::SocketAddr, pin::Pin, sync::Arc, time::SystemTime};
 
-use async_std::net::TcpStream;
+use async_std::net::{TcpListener, TcpStream};
 use futures::prelude::*;
 use futures_rustls::{
     rustls,
     rustls::{
         client::{ServerCertVerified, ServerCertVerifier},
         kx_group::X25519,
+        server::{ClientCertVerified, ClientCertVerifier},
         version::TLS13,
-        Certificate, ClientConfig, ServerName,
+        Certificate, ClientConfig, DistinguishedNames, ServerConfig, ServerName,
     },
-    TlsConnector, TlsStream,
+    TlsAcceptor, TlsConnector, TlsStream,
 };
 use log::debug;
+use rustls_pemfile::pkcs8_private_keys;
 use socket2::{Domain, Socket, Type};
 use url::Url;
 
@@ -33,7 +35,6 @@ fn cipher_suite() -> rustls::SupportedCipherSuite {
 }
 
 struct ServerCertificateVerifier;
-
 impl ServerCertVerifier for ServerCertificateVerifier {
     fn verify_server_cert(
         &self,
@@ -49,14 +50,45 @@ impl ServerCertVerifier for ServerCertificateVerifier {
     }
 }
 
+struct ClientCertificateVerifier;
+impl ClientCertVerifier for ClientCertificateVerifier {
+    fn client_auth_root_subjects(&self) -> Option<DistinguishedNames> {
+        Some(vec![])
+    }
+
+    fn verify_client_cert(
+        &self,
+        _end_entity: &Certificate,
+        _intermediates: &[Certificate],
+        _now: SystemTime,
+    ) -> Result<ClientCertVerified, rustls::Error> {
+        // TODO: upsycle
+        Ok(ClientCertVerified::assertion())
+    }
+}
+
+#[derive(Clone)]
 pub struct TlsTransport {
     pub ttl: Option<u32>,
 }
 
 impl Transport for TlsTransport {
-    type Output = TlsStream<TcpStream>;
+    type Acceptor = (TlsAcceptor, TcpListener);
+    type Connector = TlsStream<TcpStream>;
+
     type Error = io::Error;
-    type Dial = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
+
+    type Listener = Pin<Box<dyn Future<Output = Result<Self::Acceptor, Self::Error>> + Send>>;
+    type Dial = Pin<Box<dyn Future<Output = Result<Self::Connector, Self::Error>> + Send>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>> {
+        if url.scheme() != "tls" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+
+        debug!(target: "tlstransport", "listening on {}", url);
+        Ok(Box::pin(self.do_listen(url)))
+    }
 
     fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>> {
         if url.scheme() != "tls" {
@@ -84,6 +116,46 @@ impl TlsTransport {
         Ok(socket)
     }
 
+    async fn do_listen(self, url: Url) -> Result<(TlsAcceptor, TcpListener), io::Error> {
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        let socket = self.create_socket(socket_addr)?;
+        socket.bind(&socket_addr.into())?;
+        // TODO: make backlog configurable
+        socket.listen(1024)?;
+        socket.set_nonblocking(true)?;
+
+        // TODO: This should be in the struct
+        // TODO: Client auth (see upsycle)
+        let keypair_pem = ed25519_compact::KeyPair::generate().to_pem();
+        let secret_key = pkcs8_private_keys(&mut keypair_pem.as_bytes())?;
+        let secret_key = rustls::PrivateKey(secret_key[0].clone());
+
+        // TODO: Into util
+        let altnames = vec![String::from("example.com")];
+        let mut cert_params = rcgen::CertificateParams::new(altnames);
+        cert_params.alg = &rcgen::PKCS_ED25519;
+        cert_params.key_pair = Some(rcgen::KeyPair::from_pem(&keypair_pem).unwrap());
+
+        let certificate = rcgen::Certificate::from_params(cert_params).unwrap();
+        let cert_der = certificate.serialize_der().unwrap();
+        let certificate = rustls::Certificate(cert_der);
+
+        let _client_cert_verifier = Arc::new(ClientCertificateVerifier {});
+        let config = ServerConfig::builder()
+            .with_cipher_suites(&[cipher_suite()])
+            .with_kx_groups(&[&X25519])
+            .with_protocol_versions(&[&TLS13])
+            .unwrap()
+            // TODO: .with_client_cert_verifier(client_cert_verifier)
+            .with_no_client_auth()
+            .with_single_cert(vec![certificate], secret_key)
+            .unwrap();
+
+        let listener = TcpListener::from(std::net::TcpListener::from(socket));
+        let acceptor = TlsAcceptor::from(Arc::new(config));
+        Ok((acceptor, listener))
+    }
+
     async fn do_dial(self, url: Url) -> Result<TlsStream<TcpStream>, io::Error> {
         let socket_addr = url.socket_addrs(|| None)?[0];
         // TODO: Handle host

tests/network_transports.rs

@@ -0,0 +1,62 @@
+use async_std::{
+    io,
+    io::{ReadExt, WriteExt},
+    stream::StreamExt,
+    task,
+};
+use url::Url;
+
+use darkfi::net::transport::{TcpTransport, TlsTransport, Transport};
+
+#[async_std::test]
+async fn tcp_transport() {
+    let tcp = TcpTransport { ttl: None };
+    let url = Url::parse("tcp://127.0.0.1:5432").unwrap();
+
+    let listener = tcp.clone().listen_on(url.clone()).unwrap().await.unwrap();
+
+    let _ = task::spawn(async move {
+        let mut incoming = listener.incoming();
+        while let Some(stream) = incoming.next().await {
+            let stream = stream.unwrap();
+            let (reader, writer) = &mut (&stream, &stream);
+            io::copy(reader, writer).await.unwrap();
+        }
+    });
+
+    let payload = b"ohai tcp";
+
+    let mut client = tcp.dial(url).unwrap().await.unwrap();
+    client.write_all(payload).await.unwrap();
+    let mut buf = vec![0_u8; 8];
+    client.read_exact(&mut buf).await.unwrap();
+
+    assert_eq!(buf, payload);
+}
+
+#[async_std::test]
+async fn tls_transport() {
+    let tls = TlsTransport { ttl: None };
+    let url = Url::parse("tls://127.0.0.1:5433").unwrap();
+
+    let (acceptor, listener) = tls.clone().listen_on(url.clone()).unwrap().await.unwrap();
+
+    let _ = task::spawn(async move {
+        let mut incoming = listener.incoming();
+        while let Some(stream) = incoming.next().await {
+            let stream = stream.unwrap();
+            let stream = acceptor.accept(stream).await.unwrap();
+            let (mut reader, mut writer) = smol::io::split(stream);
+            io::copy(&mut reader, &mut writer).await.unwrap();
+        }
+    });
+
+    let payload = b"ohai tls";
+
+    let mut client = tls.dial(url).unwrap().await.unwrap();
+    client.write_all(payload).await.unwrap();
+    let mut buf = vec![0_u8; 8];
+    client.read_exact(&mut buf).await.unwrap();
+
+    assert_eq!(buf, payload);
+}