feat(cluster): add TLS 1.3 to replica plane with PSK channel binding

Cargo.toml

@@ -162,6 +162,12 @@ flume = "0.12.0"
 fs2 = "0.4.3"
 futures = "0.3.32"
 futures-core = { version = "0.3.32", default-features = false }
+# Pinned to the minor compio-tls 0.10.0 resolves to: the replica plane
+# handshakes with futures-rustls directly (TLS exporter access) and hands
+# the stream to compio-tls via its From impls, so both crates must agree
+# on the futures-rustls types. Cargo unifies within 0.26.x; drift to 0.27
+# fails compilation loudly (see the tripwire test in message_bus).
+futures-rustls = "0.26.0"
 futures-util = "0.3.32"
 getrandom = { version = "0.4", features = ["wasm_js"] }
 git2 = { version = "0.21.0", default-features = false, features = ["vendored-libgit2"] }

core/configs/src/server_config/cluster.rs

@@ -37,6 +37,9 @@ pub struct ClusterConfig {
     /// Replica-to-replica authentication settings (PSK + BLAKE3 handshake).
     #[serde(default)]
     pub auth: ClusterAuthConfig,
+    /// Replica-to-replica TLS settings for the consensus (`tcp_replica`) port.
+    #[serde(default)]
+    pub tls: ClusterTlsConfig,
 }
 
 /// Replica-to-replica authentication for the consensus (`tcp_replica`) port.
@@ -70,6 +73,42 @@ pub struct ClusterAuthConfig {
     pub shared_secret: String,
 }
 
+/// Replica-to-replica TLS for the consensus (`tcp_replica`) port.
+///
+/// Mirrors the legacy [`super::tcp::TcpTlsConfig`] shape plus `ca_file`:
+/// the replica plane DIALS its peers (a TLS client role the
+/// client-facing server plane never has), so the dialer needs a trust
+/// anchor to verify the acceptor's certificate against.
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
+pub struct ClusterTlsConfig {
+    /// When true every replica connection is wrapped in TLS (1.3 only)
+    /// before the replica handshake runs. Requires `cluster.auth.enabled`:
+    /// TLS carries no client certificates, so it authenticates the
+    /// acceptor only; the PSK handshake authenticates the peer while TLS
+    /// supplies confidentiality. Enabling is a coordinated-restart
+    /// change: a TLS dialer cannot talk to a plaintext acceptor or vice
+    /// versa. Flip every node in one restart.
+    #[serde(default)]
+    pub enabled: bool,
+    /// When true the node auto-generates a self-signed certificate at
+    /// boot and the dialer accepts ANY peer certificate. With the
+    /// default `false`, `cert_file` / `key_file` / `ca_file` are all
+    /// required.
+    #[serde(default)]
+    pub self_signed: bool,
+    /// PEM certificate chain presented by this node's acceptor side.
+    #[serde(default)]
+    pub cert_file: String,
+    /// PEM private key matching `cert_file`.
+    #[serde(default)]
+    pub key_file: String,
+    /// PEM trust anchor(s) the dialer verifies peer certificates
+    /// against. Unused when `self_signed` is true.
+    #[serde(default)]
+    pub ca_file: String,
+}
+
 #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
 pub struct ClusterNodeConfig {
     pub name: String,
@@ -111,6 +150,7 @@ mod tests {
                 enabled: true,
                 shared_secret: "current-psk-MUST-NOT-be-persisted".to_owned(),
             },
+            tls: ClusterTlsConfig::default(),
         };
         let serialized = serde_json::to_string(&config).expect("serialize cluster config");
         assert!(

core/configs/src/server_config/defaults.rs

@@ -15,7 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use super::cluster::{ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, TransportPorts};
+use super::cluster::{
+    ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, ClusterTlsConfig, TransportPorts,
+};
 use super::http::{HttpConfig, HttpCorsConfig, HttpJwtConfig, HttpMetricsConfig, HttpTlsConfig};
 use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig};
 use super::server::{
@@ -606,6 +608,7 @@ impl Default for ClusterConfig {
                 })
                 .collect(),
             auth: ClusterAuthConfig::default(),
+            tls: ClusterTlsConfig::default(),
         }
     }
 }

core/configs/src/server_config/validators.rs

@@ -627,6 +627,36 @@ impl Validatable<ConfigurationError> for ClusterConfig {
             return Err(ConfigurationError::InvalidConfigurationValue);
         }
 
+        // Replica TLS. Both cert modes run one-directional TLS (no client
+        // certificate anywhere), so TLS only authenticates the acceptor to
+        // the dialer; peer authentication comes solely from the PSK
+        // handshake. Without it any TLS-capable host could register as a
+        // replica - require auth in both modes. CA mode (the default)
+        // additionally needs all three PEM paths: cert/key for this node's
+        // acceptor side, ca_file as the dialer's trust anchor.
+        if self.tls.enabled {
+            if !self.auth.enabled {
+                eprintln!(
+                    "Invalid cluster configuration: cluster.tls.enabled = true requires cluster.auth.enabled = true (TLS authenticates the acceptor only; the PSK handshake authenticates the peer)"
+                );
+                return Err(ConfigurationError::InvalidConfigurationValue);
+            }
+            if !self.tls.self_signed {
+                for (field, value) in [
+                    ("cert_file", &self.tls.cert_file),
+                    ("key_file", &self.tls.key_file),
+                    ("ca_file", &self.tls.ca_file),
+                ] {
+                    if value.trim().is_empty() {
+                        eprintln!(
+                            "Invalid cluster configuration: cluster.tls.{field} must be set when cluster.tls.enabled = true and self_signed = false"
+                        );
+                        return Err(ConfigurationError::InvalidConfigurationValue);
+                    }
+                }
+            }
+        }
+
         Ok(())
     }
 }
@@ -635,7 +665,7 @@ impl Validatable<ConfigurationError> for ClusterConfig {
 mod cluster_validate_tests {
     use super::*;
     use crate::server_config::cluster::{
-        ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, TransportPorts,
+        ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, ClusterTlsConfig, TransportPorts,
     };
 
     fn node(name: &str, id: u8) -> ClusterNodeConfig {
@@ -653,6 +683,7 @@ mod cluster_validate_tests {
             name: "iggy-cluster".to_string(),
             nodes,
             auth: ClusterAuthConfig::default(),
+            tls: ClusterTlsConfig::default(),
         }
     }
 
@@ -804,6 +835,76 @@ mod cluster_validate_tests {
         c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN);
         assert!(c.validate().is_ok());
     }
+
+    fn tls_files() -> ClusterTlsConfig {
+        ClusterTlsConfig {
+            enabled: true,
+            self_signed: false,
+            cert_file: "cert.pem".to_string(),
+            key_file: "key.pem".to_string(),
+            ca_file: "ca.pem".to_string(),
+        }
+    }
+
+    #[test]
+    fn validate_rejects_tls_ca_mode_with_missing_files() {
+        // Auth on so the failure exercises the file check, not the auth gate.
+        for missing in ["cert_file", "key_file", "ca_file"] {
+            let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+            c.auth.enabled = true;
+            c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN);
+            c.tls = tls_files();
+            match missing {
+                "cert_file" => c.tls.cert_file.clear(),
+                "key_file" => c.tls.key_file.clear(),
+                _ => c.tls.ca_file.clear(),
+            }
+            assert!(c.validate().is_err(), "missing {missing} must be rejected");
+        }
+    }
+
+    #[test]
+    fn validate_rejects_tls_self_signed_without_auth() {
+        // Accept-any certificate without the PSK handshake = MITM-able.
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.tls = ClusterTlsConfig {
+            enabled: true,
+            self_signed: true,
+            ..ClusterTlsConfig::default()
+        };
+        assert!(c.validate().is_err());
+    }
+
+    #[test]
+    fn validate_accepts_tls_self_signed_with_auth() {
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.auth.enabled = true;
+        c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN);
+        c.tls = ClusterTlsConfig {
+            enabled: true,
+            self_signed: true,
+            ..ClusterTlsConfig::default()
+        };
+        assert!(c.validate().is_ok());
+    }
+
+    #[test]
+    fn validate_rejects_tls_ca_mode_without_auth() {
+        // TLS never authenticates the dialer (no client certificates);
+        // only the PSK handshake does, so it is mandatory with TLS on.
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.tls = tls_files();
+        assert!(c.validate().is_err());
+    }
+
+    #[test]
+    fn validate_accepts_tls_ca_mode_with_auth() {
+        let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+        c.auth.enabled = true;
+        c.auth.shared_secret = "a".repeat(MIN_SHARED_SECRET_LEN);
+        c.tls = tls_files();
+        assert!(c.validate().is_ok());
+    }
 }
 
 #[cfg(test)]

core/message_bus/Cargo.toml

@@ -38,6 +38,7 @@ compio-quic = { workspace = true }
 compio-ws = { workspace = true }
 configs = { workspace = true }
 futures = { workspace = true }
+futures-rustls = { workspace = true }
 iggy_binary_protocol = { workspace = true }
 iggy_common = { workspace = true }
 libc = { workspace = true }