feat(runtime): encrypt TCP request dispatch (frontend → worker)

docs/fern/pages/reference/components/tls-configuration.mdx

@@ -2,17 +2,23 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 title: TCP TLS
-subtitle: Encrypt TCP streaming connections between frontend and workers
+subtitle: Encrypt the TCP request and response streams between frontend and workers
 ---
 
-Dynamo supports opt-in TLS encryption on the TCP call-home streaming transport
-(implemented by `TcpStreamServer` and `TcpClient`, handling both response
-streams and request streams between frontends and workers). When enabled, all TCP
-connections on this path are upgraded to TLS using
+Dynamo supports opt-in TLS encryption on the TCP request and response streams
+between frontends and workers. When enabled, both the **request plane**
+(frontend → worker inference requests) and the **response stream** (worker →
+frontend inference output) are encrypted using
 [rustls](https://github.com/rustls/rustls) with the `ring` cryptographic
-provider. When no TLS configuration is provided, the transport operates in
+provider. When no TLS configuration is provided, these streams operate in
 plaintext exactly as before.
 
+The same `DYN_TCP_TLS_*` environment variables apply to both transports — a
+single set of certificates encrypts the frontend↔worker TCP request and
+response streams. This does **not** cover the KV event plane, which is carried
+over ZMQ or NATS Core depending on setup and is a separate transport; those
+paths are not encrypted by this configuration.
+
 ## Environment variables
 
 All TLS configuration is driven by environment variables. The Rust runtime
@@ -94,7 +100,7 @@ python -m dynamo.frontend \
 ## Kubernetes deployment
 
 In Kubernetes, TLS certificates are typically delivered by a certificate
-management system (e.g., cert-manager) and mounted into pods. Set the
+management system and mounted into pods. Set the
 environment variables on each component's pod template in the
 `DynamoGraphDeployment` spec:
 
@@ -135,12 +141,32 @@ server and client depending on the stream direction.
 > allowing TLS to be configured once at the platform level and auto-injected
 > into all DGD pods without per-component env var setup.
 
+## Encrypted paths
+
+When TLS is configured, the following frontend↔worker streams are encrypted:
+
+| Path | Direction | Data | Transport |
+|---|---|---|---|
+| Request plane | Frontend → Worker | User prompts, request metadata | `egress/tcp_client` → `ingress/shared_tcp_endpoint` |
+| Response stream | Worker → Frontend | Inference output tokens | `tcp/client` → `tcp/server` |
+| Request stream | Frontend → Worker | Streaming input (bidirectional) | `tcp/client` → `tcp/server` |
+
 ## Design notes
 
-- TLS configuration is cached after the first TCP connection via `OnceCell`.
-  Certificate rotation requires a process restart.
-- The TLS handshake is spawned per-connection on the server side so the accept
-  loop is never blocked by a slow handshake.
+- Server certificates hot-reload: the request-plane and response-stream servers
+  serve their leaf cert/key through a resolver that re-reads the files from disk
+  when their contents change (detected by a content hash, so rotations done by
+  an atomic symlink swap are handled too), so certificate rotation
+  takes effect **without a process restart**. The check is rate-limited (at most
+  once every 30s, sooner after a failed reload) and never blocks a handshake; a
+  failed reload keeps serving the last valid certificate. Client trust anchors
+  (the CA) are still loaded once, so rotating the CA itself requires a restart.
+- Client TLS connectors are built once and cached via `OnceCell` on the first
+  outbound connection.
+- The TLS handshake is spawned per-connection on both the request plane and
+  response stream servers so the accept loop is never blocked.
+- Invalid TLS configuration on the request plane (e.g. bad cert path) prevents
+  server startup rather than silently falling back to plaintext.
 - When server and client TLS configurations are mismatched (e.g., server has TLS
   but client does not), a warning is logged at startup.
 - An empty CA certificate file is detected at load time and rejected with a

lib/runtime/src/pipeline/network/egress/tcp_client.rs

@@ -31,8 +31,12 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::Duration;
-use tokio::io::{AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 use tokio::net::TcpStream;
+use tokio_rustls::TlsConnector;
+
+type BoxRead = Box<dyn AsyncRead + Unpin + Send>;
+type BoxWrite = Box<dyn AsyncWrite + Unpin + Send>;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
 use tokio_util::codec::FramedRead;
@@ -376,17 +380,82 @@ struct TcpConnection {
     post_enqueue_barrier: Option<Arc<tokio::sync::Barrier>>,
 }
 
+/// Cached TLS connector for the request plane. Built once from env vars.
+static REQUEST_PLANE_TLS_CONNECTOR: once_cell::sync::OnceCell<Option<TlsConnector>> =
+    once_cell::sync::OnceCell::new();
+
+fn get_request_plane_tls_connector() -> anyhow::Result<&'static Option<TlsConnector>> {
+    REQUEST_PLANE_TLS_CONNECTOR.get_or_try_init(build_request_plane_tls_connector_from_env)
+}
+
+/// Build the request-plane client TLS connector from the `DYN_TCP_TLS_*`
+/// environment. Returns `None` when no TLS is requested (no CA and not
+/// insecure). Split out from the `OnceCell` init so the env parsing can be
+/// unit-tested directly.
+fn build_request_plane_tls_connector_from_env() -> anyhow::Result<Option<TlsConnector>> {
+    use crate::config::environment_names::tcp_response_stream::tls as env;
+    let ca_cert_path = std::env::var(env::DYN_TCP_TLS_CA_CERT_PATH).ok();
+    let insecure = crate::config::env_is_truthy(env::DYN_TCP_TLS_INSECURE);
+    let tls_requested = ca_cert_path.is_some() || insecure;
+    if !tls_requested {
+        return Ok(None);
+    }
+    let tls_config = crate::tls_utils::client_tls_config(
+        ca_cert_path.as_deref().map(std::path::Path::new),
+        insecure,
+    )?;
+    Ok(Some(TlsConnector::from(std::sync::Arc::new(tls_config))))
+}
+
 impl TcpConnection {
     /// Create a new connection with lock-free submit and batched write/read tasks
     async fn connect(addr: SocketAddr, timeout: Duration, channel_buffer: usize) -> Result<Self> {
+        Self::connect_with_connector(
+            addr,
+            timeout,
+            channel_buffer,
+            get_request_plane_tls_connector()?.as_ref(),
+        )
+        .await
+    }
+
+    /// Like [`TcpConnection::connect`], but with an explicitly supplied TLS
+    /// connector instead of the process-global `REQUEST_PLANE_TLS_CONNECTOR`.
+    /// This lets tests drive the real connect/handshake/reader/writer path with a
+    /// per-test connector, without initializing (and poisoning) the cached one.
+    async fn connect_with_connector(
+        addr: SocketAddr,
+        timeout: Duration,
+        channel_buffer: usize,
+        connector: Option<&TlsConnector>,
+    ) -> Result<Self> {
         let stream = tokio::time::timeout(timeout, TcpStream::connect(addr))
             .await
             .map_err(|_| anyhow::anyhow!("TCP connect timeout to {}", addr))??;
 
         // Configure socket for lower latency
         Self::configure_socket(&stream)?;
 
-        let (read_half, write_half) = tokio::io::split(stream);
+        let (read_half, write_half): (BoxRead, BoxWrite) = if let Some(connector) = connector {
+            use crate::config::environment_names::tcp_response_stream::tls as env;
+            let server_name = match std::env::var(env::DYN_TCP_TLS_SERVER_NAME) {
+                Ok(name) => rustls::pki_types::ServerName::try_from(name)
+                    .map_err(|e| anyhow::anyhow!("invalid TLS server name: {e}"))?,
+                Err(_) => rustls::pki_types::ServerName::IpAddress(addr.ip().into()),
+            };
+            let tls_stream = tokio::time::timeout(
+                crate::tls_utils::handshake_timeout(),
+                connector.connect(server_name, stream),
+            )
+            .await
+            .map_err(|_| anyhow::anyhow!("Request plane TLS handshake timed out to {}", addr))?
+            .map_err(|e| anyhow::anyhow!("Request plane TLS handshake failed to {}: {e}", addr))?;
+            let (r, w) = tokio::io::split(tls_stream);
+            (Box::new(r), Box::new(w))
+        } else {
+            let (r, w) = tokio::io::split(stream);
+            (Box::new(r), Box::new(w))
+        };
 
         let submit_queue = Arc::new(SegQueue::new());
         let response_queue = Arc::new(SegQueue::new());
@@ -557,7 +626,7 @@ impl TcpConnection {
     /// - If a response races back before waiters are queued, the reader's
     ///   existing spin-wait covers that small handoff window
     async fn writer_task(
-        mut write_half: tokio::io::WriteHalf<TcpStream>,
+        mut write_half: BoxWrite,
         submit_queue: Arc<SegQueue<PendingRequest>>,
         response_queue: Arc<SegQueue<oneshot::Sender<Result<Bytes>>>>,
         notify: Arc<tokio::sync::Notify>,
@@ -632,11 +701,26 @@ impl TcpConnection {
                     }
                     return Err(e.into());
                 }
+                // Flush after the batch write. `write_half` may be a tokio-rustls
+                // TLS stream, whose `poll_write` copies plaintext into the session
+                // buffer and can return Ready(Ok(n)) with encrypted records still
+                // buffered when the socket would block; without this flush the batch
+                // can stall under write back-pressure until the next request is
+                // written, hanging its callers until the request timeout. For a
+                // plaintext TCP write half this is a no-op.
+                if let Err(e) = write_half.flush().await {
+                    write_buf.clear();
+                    let err_msg = format!("Flush failed: {}", e);
+                    for tx in response_batch.drain(..) {
+                        let _ = tx.send(Err(anyhow::anyhow!("{}", err_msg)));
+                    }
+                    return Err(e.into());
+                }
                 TCP_BYTES_SENT_TOTAL.inc_by(bytes_to_write as f64);
                 debug_assert!(write_buf.is_empty());
 
-                // Phase 3: write_all succeeded — data is committed to the wire.
-                // NOW push response_txs to response_queue so the reader can
+                // Phase 3: write_all + flush succeeded — data is committed to the
+                // wire. NOW push response_txs to response_queue so the reader can
                 // match them with incoming responses.
                 for tx in response_batch.drain(..) {
                     response_queue.push(tx);
@@ -698,7 +782,7 @@ impl TcpConnection {
     /// On exit (clean close or error), sets `healthy=false` and wakes the writer
     /// via `writer_notify` so it can detect reader death and drain pending callers.
     async fn reader_task(
-        read_half: tokio::io::ReadHalf<TcpStream>,
+        read_half: BoxRead,
         response_queue: Arc<SegQueue<oneshot::Sender<Result<Bytes>>>>,
         healthy: Arc<AtomicBool>,
         writer_notify: Arc<tokio::sync::Notify>,
@@ -1590,6 +1674,22 @@ mod tests {
     use tokio::io::{AsyncReadExt, AsyncWrite};
     use tokio::net::TcpListener;
 
+    fn make_cert_files() -> (tempfile::NamedTempFile, tempfile::NamedTempFile) {
+        use std::io::Write as _;
+        let key_pair = rcgen::KeyPair::generate().unwrap();
+        let cert = rcgen::CertificateParams::new(vec!["localhost".to_string()])
+            .unwrap()
+            .self_signed(&key_pair)
+            .unwrap();
+        let mut cert_file = tempfile::NamedTempFile::new().unwrap();
+        cert_file.write_all(cert.pem().as_bytes()).unwrap();
+        let mut key_file = tempfile::NamedTempFile::new().unwrap();
+        key_file
+            .write_all(key_pair.serialize_pem().as_bytes())
+            .unwrap();
+        (cert_file, key_file)
+    }
+
     #[test]
     fn test_tcp_config_default() {
         let config = TcpRequestConfig::default();
@@ -1713,6 +1813,127 @@ mod tests {
         (addr, conn_count)
     }
 
+    /// Env parsing for the request-plane client connector (independent of the
+    /// process-global `OnceCell`): no TLS env → no connector; CA or insecure →
+    /// connector built.
+    #[test]
+    fn request_plane_tls_connector_from_env_parses() {
+        let (cert, _key) = make_cert_files();
+        temp_env::with_vars_unset(["DYN_TCP_TLS_CA_CERT_PATH", "DYN_TCP_TLS_INSECURE"], || {
+            assert!(
+                build_request_plane_tls_connector_from_env()
+                    .unwrap()
+                    .is_none()
+            );
+        });
+        temp_env::with_vars(
+            [
+                (
+                    "DYN_TCP_TLS_CA_CERT_PATH",
+                    Some(cert.path().to_str().unwrap()),
+                ),
+                ("DYN_TCP_TLS_INSECURE", None),
+            ],
+            || {
+                assert!(
+                    build_request_plane_tls_connector_from_env()
+                        .unwrap()
+                        .is_some()
+                );
+            },
+        );
+        temp_env::with_vars(
+            [
+                ("DYN_TCP_TLS_CA_CERT_PATH", None),
+                ("DYN_TCP_TLS_INSECURE", Some("1")),
+            ],
+            || {
+                assert!(
+                    build_request_plane_tls_connector_from_env()
+                        .unwrap()
+                        .is_some()
+                );
+            },
+        );
+    }
+
+    /// End-to-end encrypted request through the **real** request-plane client
+    /// path: `TcpConnection::connect_with_connector` performs the TLS handshake
+    /// (SNI from `DYN_TCP_TLS_SERVER_NAME`), spawns the reader/writer tasks over
+    /// boxed TLS I/O, and `send_request` frames a request that a TLS-wrapped echo
+    /// server (built from the production `server_tls_config`) reads and replies
+    /// to. This exercises handshake + SNI + boxed I/O + reader/writer + framing,
+    /// not just a `tls_utils` round-trip.
+    #[tokio::test]
+    async fn request_plane_tls_end_to_end() {
+        use crate::pipeline::network::codec::TcpResponseMessage;
+        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+
+        // Self-signed cert (SAN=localhost), trusted as the CA by the client.
+        let (cert, key) = make_cert_files();
+        let server_config = crate::tls_utils::server_tls_config(cert.path(), key.path()).unwrap();
+        let acceptor = tokio_rustls::TlsAcceptor::from(std::sync::Arc::new(server_config));
+        let client_config = crate::tls_utils::client_tls_config(Some(cert.path()), false).unwrap();
+        let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(client_config));
+
+        // TLS-wrapped echo server speaking the request-plane wire framing.
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            let (tcp, _) = listener.accept().await.unwrap();
+            let tls = acceptor.accept(tcp).await.expect("server TLS handshake");
+            let (mut r, mut w) = tokio::io::split(tls);
+            // path
+            let mut l2 = [0u8; 2];
+            r.read_exact(&mut l2).await.unwrap();
+            let mut path = vec![0u8; u16::from_be_bytes(l2) as usize];
+            r.read_exact(&mut path).await.unwrap();
+            // headers
+            let mut hl = [0u8; 2];
+            r.read_exact(&mut hl).await.unwrap();
+            let mut headers = vec![0u8; u16::from_be_bytes(hl) as usize];
+            r.read_exact(&mut headers).await.unwrap();
+            // payload
+            let mut l4 = [0u8; 4];
+            r.read_exact(&mut l4).await.unwrap();
+            let mut payload = vec![0u8; u32::from_be_bytes(l4) as usize];
+            r.read_exact(&mut payload).await.unwrap();
+            // echo the payload back as a response frame
+            let resp = TcpResponseMessage::new(Bytes::from(payload));
+            w.write_all(&resp.encode().unwrap()).await.unwrap();
+            w.flush().await.unwrap();
+        });
+
+        // Drive the real client path with an explicit connector (no OnceCell) and
+        // SNI supplied via env so it matches the cert's `localhost` SAN.
+        let payload = Bytes::from_static(b"encrypted-request-plane-payload");
+        let expected = payload.clone();
+        let response = temp_env::async_with_vars(
+            [("DYN_TCP_TLS_SERVER_NAME", Some("localhost"))],
+            async move {
+                let conn = TcpConnection::connect_with_connector(
+                    addr,
+                    Duration::from_secs(5),
+                    10,
+                    Some(&connector),
+                )
+                .await
+                .expect("client connect + TLS handshake");
+                let mut headers = Headers::new();
+                headers.insert("x-endpoint-path".to_string(), "test".to_string());
+                conn.send_request(payload, &headers)
+                    .await
+                    .expect("send_request over TLS")
+            },
+        )
+        .await;
+
+        assert_eq!(
+            response, expected,
+            "payload should round-trip through the encrypted request plane"
+        );
+    }
+
     struct RecordingWriter {
         written: Vec<u8>,
         max_per_write: Option<usize>,