feat(networking): native mTLS with subject-name authorization for fabric inter-node communication

Cargo.toml

@@ -231,6 +231,7 @@ rocksdb = { version = "0.46.1", package = "rust-rocksdb", features = [
 ], git = "https://github.com/restatedev/rust-rocksdb", rev = "dcfba7946697d740e60f0b1060b6624dc1c7e94a" }
 rstest = "0.26.1"
 rustls = { version = "0.23.35", default-features = false, features = ["ring"] }
+rustls-pemfile = "2"
 schemars = { version = "1.2", features = ["bytes1"] }
 semver = { version = "1.0", features = ["serde"] }
 serde = { version = "1.0", features = ["derive"] }
@@ -258,6 +259,7 @@ tokio = { version = "1.48.0", default-features = false, features = [
     "macros",
     "parking_lot",
 ] }
+tokio-rustls = { version = "0.26", default-features = false, features = ["ring"] }
 tokio-stream = "0.1.17"
 tokio-util = { version = "0.7.17" }
 toml = { version = "0.9" }

crates/admin/src/service.rs

@@ -212,7 +212,7 @@ where
             TaskCenter::with_current(|tc| opts.advertised_address(tc.address_book()))
         );
 
-        net_util::run_hyper_server(self.listeners, service, || ())
+        net_util::run_hyper_server(self.listeners, service, || (), None)
             .await
             .map_err(Into::into)
     }

crates/core/Cargo.toml

@@ -62,9 +62,12 @@ static_assertions = { workspace = true }
 strum = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["tracing"] }
+tokio-rustls = { workspace = true }
 tokio-stream = { workspace = true, features = ["net"] }
 tokio-util = { workspace = true, features = ["net"] }
 tonic = { workspace = true, features = ["transport", "codegen", "gzip", "zstd", "router"] }
+rustls = { workspace = true }
+rustls-pemfile = { workspace = true }
 tonic-prost = { workspace = true }
 tonic-reflection = { workspace = true }
 tower = { workspace = true }
@@ -81,6 +84,7 @@ restate-metadata-store = { workspace = true, features = ["test-util"] }
 restate-test-util = { workspace = true }
 
 googletest = { workspace = true }
+tempfile = { workspace = true }
 test-log = { workspace = true }
 tracing-subscriber = { workspace = true }
 tracing-test = { workspace = true }

crates/core/src/network/grpc/connector.rs

@@ -11,8 +11,10 @@
 use futures::Stream;
 use http::Uri;
 use hyper_util::rt::TokioIo;
+use rustls::pki_types::ServerName;
 use tokio::io;
 use tokio::net::UnixStream;
+use tokio_rustls::TlsConnector;
 use tokio_stream::StreamExt;
 use tonic::codec::CompressionEncoding;
 use tonic::transport::Endpoint;
@@ -26,13 +28,20 @@ use restate_types::net::connect_opts::GrpcConnectionOptions;
 use crate::network::grpc::DEFAULT_GRPC_COMPRESSION;
 use crate::network::protobuf::core_node_svc::core_node_svc_client::CoreNodeSvcClient;
 use crate::network::protobuf::network::Message;
+use crate::network::tls::TlsCertResolver;
 use crate::network::transport_connector::find_node;
 use crate::network::{ConnectError, Destination, Swimlane, TransportConnect};
 use crate::{Metadata, TaskCenter, TaskKind};
 
 #[derive(Clone, Default)]
 pub struct GrpcConnector {
-    _private: (),
+    tls: Option<TlsCertResolver>,
+}
+
+impl GrpcConnector {
+    pub fn new(tls: Option<TlsCertResolver>) -> Self {
+        Self { tls }
+    }
 }
 
 impl TransportConnect for GrpcConnector {
@@ -53,7 +62,7 @@ impl TransportConnect for GrpcConnector {
 
         debug!("Connecting to {} at {}", destination, address);
         let networking = &Configuration::pinned().networking;
-        let channel = create_channel(address, swimlane, networking);
+        let channel = create_channel(address, swimlane, networking, &self.tls);
 
         // Establish the connection
         let client = CoreNodeSvcClient::new(channel)
@@ -85,8 +94,11 @@ fn create_channel<P: ListenerPort + GrpcPort>(
     address: AdvertisedAddress<P>,
     _swimlane: Swimlane,
     options: &NetworkingOptions,
+    tls: &Option<TlsCertResolver>,
 ) -> Channel {
     let address = address.into_address().expect("valid address");
+    let use_tls = address.is_tls() && tls.is_some();
+
     let endpoint = match &address {
         PeerNetAddress::Uds(_) => {
             // dummy endpoint required to specify an uds connector, it is not used anywhere
@@ -108,7 +120,6 @@ fn create_channel<P: ListenerPort + GrpcPort>(
         .initial_stream_window_size(options.stream_window_size())
         .initial_connection_window_size(options.connection_window_size())
         .keep_alive_while_idle(true)
-        // this true by default, but this is to guard against any change in defaults
         .tcp_nodelay(true);
 
     match address {
@@ -120,7 +131,27 @@ fn create_channel<P: ListenerPort + GrpcPort>(
                 }
             }))
         }
-        PeerNetAddress::Http(_) => endpoint.connect_lazy()
+        PeerNetAddress::Http(uri) if use_tls => {
+            let resolver = tls.as_ref().unwrap().clone();
+            endpoint.connect_with_connector_lazy(tower::service_fn(move |_: Uri| {
+                let resolver = resolver.clone();
+                let host = uri.host().unwrap_or("localhost").to_owned();
+                let port = uri.port_u16().unwrap_or(5122);
+                async move {
+                    let addr = format!("{host}:{port}");
+                    let tcp_stream = tokio::net::TcpStream::connect(&addr).await?;
+                    tcp_stream.set_nodelay(true)?;
+
+                    let client_config = resolver.client_config();
+                    let connector = TlsConnector::from(client_config);
+                    let server_name = ServerName::try_from(host)
+                        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
+                    let tls_stream = connector.connect(server_name, tcp_stream).await?;
+                    Ok::<_, io::Error>(TokioIo::new(tls_stream))
+                }
+            }))
+        }
+        PeerNetAddress::Http(_) => endpoint.connect_lazy(),
     }
 }
 

crates/core/src/network/mod.rs

@@ -23,6 +23,7 @@ mod network_sender;
 mod networking;
 pub mod protobuf;
 mod server_builder;
+pub mod tls;
 pub mod tonic_service_filter;
 mod tracking;
 pub mod transport_connector;

crates/core/src/network/net_util.rs

@@ -26,13 +26,14 @@ use tokio_util::either::Either;
 use tonic::transport::{Channel, Endpoint};
 use tracing::{Instrument, Span, debug, error_span, info, instrument, trace};
 
-use restate_types::config::Configuration;
+use restate_types::config::{Configuration, TlsMode};
 use restate_types::errors::GenericError;
 use restate_types::net::address::{AdvertisedAddress, GrpcPort};
 use restate_types::net::address::{ListenerPort, PeerNetAddress};
 use restate_types::net::connect_opts::CommonClientConnectionOptions;
 use restate_types::net::listener::Listeners;
 
+use crate::network::tls::TlsCertResolver;
 use crate::{ShutdownError, TaskCenter, TaskKind, cancellation_watcher};
 
 pub enum DNSResolution {
@@ -129,6 +130,7 @@ pub async fn run_hyper_server<P: ListenerPort, S, B>(
     listeners: Listeners<P>,
     service: S,
     on_stop: impl Fn(),
+    tls: Option<TlsCertResolver>,
 ) -> Result<(), Error>
 where
     S: hyper::service::Service<http::Request<Incoming>, Response = hyper::Response<B>>
@@ -150,8 +152,12 @@ where
         Span::current().record("server.port", socket_addr.port());
     }
 
-    info!("Server listening");
-    run_listener_loop(listeners, service, P::NAME).await?;
+    if tls.is_some() {
+        info!("Server listening with TLS enabled");
+    } else {
+        info!("Server listening");
+    }
+    run_listener_loop(listeners, service, P::NAME, tls).await?;
     on_stop();
 
     info!("Stopped listening");
@@ -163,6 +169,7 @@ async fn run_listener_loop<P: ListenerPort, S, B>(
     mut listeners: Listeners<P>,
     service: S,
     server_name: &'static str,
+    tls: Option<TlsCertResolver>,
 ) -> Result<(), Error>
 where
     S: hyper::service::Service<http::Request<Incoming>, Response = hyper::Response<B>>
@@ -180,6 +187,16 @@ where
     let graceful_shutdown = GracefulShutdown::new();
     let task_name: Arc<str> = Arc::from(format!("{server_name}-socket"));
 
+    let tls_mode = tls.as_ref().map(|_| {
+        configuration
+            .live_load()
+            .networking
+            .tls
+            .as_ref()
+            .map(|t| t.mode.clone())
+            .unwrap_or(TlsMode::Strict)
+    });
+
     loop {
         tokio::select! {
             biased;
@@ -205,29 +222,123 @@ where
 
                 match stream {
                     Either::Left(tcp_stream) => {
-                        // TCP SOCKET
-                        let io = TokioIo::new(tcp_stream);
-                        let connection = graceful_shutdown.watch(builder
-                            .serve_connection(io, service.clone()).into_owned());
-                        TaskCenter::spawn(TaskKind::SocketHandler, task_name.clone(), async move {
-                            trace!("New tcp connection accepted");
-                            if let Err(e) = connection.await {
-                                if let Some(hyper_error) = e.downcast_ref::<hyper::Error>() {
-                                    if hyper_error.is_incomplete_message() {
-                                        debug!("Connection closed before request completed");
+                        let tls_resolver = tls.clone();
+                        let tls_mode = tls_mode.clone();
+                        let service = service.clone();
+                        let graceful_shutdown = &graceful_shutdown;
+                        let task_name = task_name.clone();
+
+                        match (&tls_resolver, &tls_mode) {
+                            (Some(resolver), Some(TlsMode::Strict)) => {
+                                // TLS strict: all connections must be TLS
+                                let acceptor = resolver.tls_acceptor();
+                                let connection = match acceptor.accept(tcp_stream).await {
+                                    Ok(tls_stream) => {
+                                        let io = TokioIo::new(tls_stream);
+                                        graceful_shutdown.watch(builder.serve_connection(io, service).into_owned())
+                                    }
+                                    Err(e) => {
+                                        debug!("TLS handshake failed: {e}");
+                                        continue;
+                                    }
+                                };
+                                TaskCenter::spawn(TaskKind::SocketHandler, task_name, async move {
+                                    trace!("New TLS tcp connection accepted");
+                                    if let Err(e) = connection.await {
+                                        if let Some(hyper_error) = e.downcast_ref::<hyper::Error>() {
+                                            if hyper_error.is_incomplete_message() {
+                                                debug!("Connection closed before request completed");
+                                            }
+                                        } else {
+                                            debug!("Connection terminated due to error: {e}");
+                                        }
+                                    } else {
+                                        trace!("Connection completed cleanly");
+                                    }
+                                    Ok(())
+                                }.instrument(socket_span))?;
+                            }
+                            (Some(resolver), Some(TlsMode::Optional)) => {
+                                // TLS optional: peek first byte to detect TLS ClientHello
+                                let tcp_stream = tcp_stream;
+                                let mut peek_buf = [0u8; 1];
+                                match tcp_stream.peek(&mut peek_buf).await {
+                                    Ok(1) if peek_buf[0] == 0x16 => {
+                                        // TLS ClientHello detected
+                                        let acceptor = resolver.tls_acceptor();
+                                        let connection = match acceptor.accept(tcp_stream).await {
+                                            Ok(tls_stream) => {
+                                                let io = TokioIo::new(tls_stream);
+                                                graceful_shutdown.watch(builder.serve_connection(io, service).into_owned())
+                                            }
+                                            Err(e) => {
+                                                debug!("TLS handshake failed: {e}");
+                                                continue;
+                                            }
+                                        };
+                                        TaskCenter::spawn(TaskKind::SocketHandler, task_name, async move {
+                                            trace!("New TLS tcp connection accepted (optional mode)");
+                                            if let Err(e) = connection.await {
+                                                if let Some(hyper_error) = e.downcast_ref::<hyper::Error>() {
+                                                    if hyper_error.is_incomplete_message() {
+                                                        debug!("Connection closed before request completed");
+                                                    }
+                                                } else {
+                                                    debug!("Connection terminated due to error: {e}");
+                                                }
+                                            } else {
+                                                trace!("Connection completed cleanly");
+                                            }
+                                            Ok(())
+                                        }.instrument(socket_span))?;
+                                    }
+                                    _ => {
+                                        // Plaintext connection
+                                        let io = TokioIo::new(tcp_stream);
+                                        let connection = graceful_shutdown.watch(builder.serve_connection(io, service).into_owned());
+                                        TaskCenter::spawn(TaskKind::SocketHandler, task_name, async move {
+                                            trace!("New plaintext tcp connection accepted (optional mode)");
+                                            if let Err(e) = connection.await {
+                                                if let Some(hyper_error) = e.downcast_ref::<hyper::Error>() {
+                                                    if hyper_error.is_incomplete_message() {
+                                                        debug!("Connection closed before request completed");
+                                                    }
+                                                } else {
+                                                    debug!("Connection terminated due to error: {e}");
+                                                }
+                                            } else {
+                                                trace!("Connection completed cleanly");
+                                            }
+                                            Ok(())
+                                        }.instrument(socket_span))?;
                                     }
-                                } else {
-                                    debug!("Connection terminated due to error: {e}");
                                 }
-                            } else {
-                                trace!("Connection completed cleanly");
                             }
-                            Ok(())
-                        }.instrument(socket_span))?;
-
+                            _ => {
+                                // No TLS: plaintext (current behavior)
+                                let io = TokioIo::new(tcp_stream);
+                                let connection = graceful_shutdown.watch(builder
+                                    .serve_connection(io, service).into_owned());
+                                TaskCenter::spawn(TaskKind::SocketHandler, task_name, async move {
+                                    trace!("New tcp connection accepted");
+                                    if let Err(e) = connection.await {
+                                        if let Some(hyper_error) = e.downcast_ref::<hyper::Error>() {
+                                            if hyper_error.is_incomplete_message() {
+                                                debug!("Connection closed before request completed");
+                                            }
+                                        } else {
+                                            debug!("Connection terminated due to error: {e}");
+                                        }
+                                    } else {
+                                        trace!("Connection completed cleanly");
+                                    }
+                                    Ok(())
+                                }.instrument(socket_span))?;
+                            }
+                        }
                     },
                     Either::Right(unix_stream) => {
-                        // UNIX SOCKET
+                        // UNIX SOCKET — TLS never applies to UDS
                         let io = TokioIo::new(unix_stream);
                         let connection = graceful_shutdown.watch(builder
                             .serve_connection(io, service.clone()).into_owned());

crates/core/src/network/networking.rs

@@ -38,10 +38,10 @@ impl<T: Clone> Clone for Networking<T> {
 }
 
 impl Networking<GrpcConnector> {
-    pub fn with_grpc_connector() -> Self {
+    pub fn with_grpc_connector(tls: Option<super::tls::TlsCertResolver>) -> Self {
         Self {
             connections: ConnectionManager::default(),
-            connector: GrpcConnector::default(),
+            connector: GrpcConnector::new(tls),
         }
     }
 }