feat(transport): TCP-TLS

Cargo.toml

@@ -1,11 +1,11 @@
 [workspace]
 members = [
-    "msg",
-    "msg-socket",
-    "msg-wire",
-    "msg-transport",
-    "msg-common",
-    "msg-sim",
+  "msg",
+  "msg-socket",
+  "msg-wire",
+  "msg-transport",
+  "msg-common",
+  "msg-sim",
 ]
 resolver = "2"
 
@@ -19,19 +19,19 @@ authors = ["Chainbound Developers <dev@chainbound.io>"]
 homepage = "https://github.com/chainbound/msg-rs"
 repository = "https://github.com/chainbound/msg-rs"
 keywords = [
-    "messaging",
-    "distributed",
-    "systems",
-    "networking",
-    "quic",
-    "quinn",
-    "tokio",
-    "async",
-    "simulation",
-    "pnet",
-    "udp",
-    "tcp",
-    "socket",
+  "messaging",
+  "distributed",
+  "systems",
+  "networking",
+  "quic",
+  "quinn",
+  "tokio",
+  "async",
+  "simulation",
+  "pnet",
+  "udp",
+  "tcp",
+  "socket",
 ]
 
 [workspace.dependencies]
@@ -45,8 +45,9 @@ msg-sim = { path = "./msg-sim" }
 async-trait = "0.1"
 tokio = { version = "1", features = ["full"] }
 tokio-util = { version = "0.7", features = ["codec"] }
-futures = "0.3"
 tokio-stream = { version = "0.1", features = ["sync"] }
+tokio-openssl = { version = "0.6" }
+futures = "0.3"
 parking_lot = "0.12"
 
 # general
@@ -55,10 +56,19 @@ thiserror = "1"
 tracing = "0.1"
 rustc-hash = "1"
 rand = "0.8"
+derive_more = { version = "2.0.1", features = [
+  "from",
+  "into",
+  "deref",
+  "deref_mut",
+] }
 
 # networking
 quinn = "0.11.9"
 rcgen = "0.14"
+# (rustls needs to be the same version as the one used by quinn)
+rustls = { version = "0.21", features = ["quic", "dangerous_configuration"] }
+openssl = { version = "0.10" }
 
 # benchmarking & profiling
 criterion = { version = "0.5", features = ["async_tokio"] }

msg-socket/Cargo.toml

@@ -28,7 +28,8 @@ parking_lot.workspace = true
 
 [dev-dependencies]
 rand.workspace = true
-msg-transport = { workspace = true, features = ["quic"] }
+msg-transport = { workspace = true, features = ["quic", "tcp-tls"] }
+openssl.workspace = true
 
 msg-sim.workspace = true
 

msg-socket/tests/it/reqrep.rs

@@ -1,10 +1,64 @@
 use bytes::Bytes;
 use msg_socket::{RepSocket, ReqSocket};
-use msg_transport::tcp::Tcp;
+use msg_transport::{
+    tcp::Tcp,
+    tcp_tls::{self, TcpTls},
+};
 use tokio_stream::StreamExt;
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn test_reqrep() {
+/// Helper functions.
+mod helpers {
+    use std::{path::PathBuf, str::FromStr as _};
+
+    use openssl::ssl::{
+        SslAcceptor, SslAcceptorBuilder, SslConnector, SslConnectorBuilder, SslFiletype, SslMethod,
+    };
+
+    /// Creates a default SSL acceptor builder for testing, with a trusted CA.
+    pub fn default_acceptor_builder() -> SslAcceptorBuilder {
+        let certificate_path =
+            PathBuf::from_str("../testdata/certificates/server-cert.pem").unwrap();
+        let private_key_path =
+            PathBuf::from_str("../testdata/certificates/server-key.pem").unwrap();
+        let ca_certificate_path =
+            PathBuf::from_str("../testdata/certificates/ca-cert.pem").unwrap();
+
+        assert!(certificate_path.exists(), "Certificate file does not exist");
+        assert!(private_key_path.exists(), "Private key file does not exist");
+        assert!(ca_certificate_path.exists(), "CA Certificate file does not exist");
+
+        let mut acceptor_builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
+        acceptor_builder.set_certificate_file(certificate_path, SslFiletype::PEM).unwrap();
+        acceptor_builder.set_private_key_file(private_key_path, SslFiletype::PEM).unwrap();
+        acceptor_builder.set_ca_file(ca_certificate_path).unwrap();
+        acceptor_builder
+    }
+
+    /// Creates a default SSL connector builder for testing, with a trusted CA.
+    /// It also has client certificate and private key set for mTLS testing.
+    pub fn default_connector_builder() -> SslConnectorBuilder {
+        let certificate_path =
+            PathBuf::from_str("../testdata/certificates/client-cert.pem").unwrap();
+        let private_key_path =
+            PathBuf::from_str("../testdata/certificates/client-key.pem").unwrap();
+        let ca_certificate_path =
+            PathBuf::from_str("../testdata/certificates/ca-cert.pem").unwrap();
+
+        assert!(certificate_path.exists(), "Certificate file does not exist");
+        assert!(private_key_path.exists(), "Private key file does not exist");
+        assert!(ca_certificate_path.exists(), "CA Certificate file does not exist");
+
+        let mut connector_builder = SslConnector::builder(SslMethod::tls()).unwrap();
+        connector_builder.set_certificate_file(certificate_path, SslFiletype::PEM).unwrap();
+        connector_builder.set_private_key_file(private_key_path, SslFiletype::PEM).unwrap();
+        connector_builder.set_ca_file(ca_certificate_path).unwrap();
+
+        connector_builder
+    }
+}
+
+#[tokio::test]
+async fn reqrep_works() {
     let _ = tracing_subscriber::fmt::try_init();
 
     let mut rep = RepSocket::new(Tcp::default());
@@ -21,6 +75,72 @@ async fn test_reqrep() {
         }
     });
 
-    let response = req.request(Bytes::from_static(b"hello")).await.unwrap();
-    tracing::info!("Response: {:?}", response);
+    let hello = Bytes::from_static(b"hello");
+    let response = req.request(hello.clone()).await.unwrap();
+    assert_eq!(hello, response, "expected {:?}, got {:?}", hello, response);
+}
+
+#[tokio::test]
+async fn reqrep_tls_works() {
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let server_config = tcp_tls::config::Server::new(helpers::default_acceptor_builder().build());
+    let tcp_tls_server = TcpTls::new_server(server_config);
+    let mut rep = RepSocket::new(tcp_tls_server);
+
+    rep.bind("0.0.0.0:0").await.unwrap();
+
+    let domain = "localhost".to_string();
+    let ssl_connector = helpers::default_connector_builder().build();
+    let tcp_tls_client =
+        TcpTls::new_client(tcp_tls::config::Client::new(domain).with_ssl_connector(ssl_connector));
+    let mut req = ReqSocket::new(tcp_tls_client);
+
+    req.connect(rep.local_addr().unwrap()).await.unwrap();
+
+    tokio::spawn(async move {
+        while let Some(request) = rep.next().await {
+            let msg = request.msg().clone();
+            request.respond(msg).unwrap();
+        }
+    });
+
+    let hello = Bytes::from_static(b"hello");
+    let response = req.request(hello.clone()).await.unwrap();
+    assert_eq!(hello, response, "expected {:?}, got {:?}", hello, response);
+}
+
+#[tokio::test]
+async fn reqrep_mutual_tls_works() {
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let mut acceptor_builder = helpers::default_acceptor_builder();
+    // By specifying peer verification mode, we essentially toggle mTLS.
+    acceptor_builder.set_verify(
+        openssl::ssl::SslVerifyMode::PEER | openssl::ssl::SslVerifyMode::FAIL_IF_NO_PEER_CERT,
+    );
+    let server_config = tcp_tls::config::Server::new(acceptor_builder.build());
+    let tcp_tls_server = TcpTls::new_server(server_config);
+    let mut rep = RepSocket::new(tcp_tls_server);
+
+    rep.bind("0.0.0.0:0").await.unwrap();
+
+    let domain = "localhost".to_string();
+    let ssl_connector = helpers::default_connector_builder().build();
+    let tcp_tls_client =
+        TcpTls::new_client(tcp_tls::config::Client::new(domain).with_ssl_connector(ssl_connector));
+    let mut req = ReqSocket::new(tcp_tls_client);
+
+    req.connect(rep.local_addr().unwrap()).await.unwrap();
+
+    tokio::spawn(async move {
+        while let Some(request) = rep.next().await {
+            let msg = request.msg().clone();
+            request.respond(msg).unwrap();
+        }
+    });
+
+    let hello = Bytes::from_static(b"hello");
+    let response = req.request(hello.clone()).await.unwrap();
+    assert_eq!(hello, response, "expected {:?}, got {:?}", hello, response);
 }