feat(cluster): authenticate replica peers with PSK + BLAKE3 handshake

Cargo.toml

@@ -336,6 +336,7 @@ wiremock = "0.6"
 yew = { version = "0.23", features = ["csr"] }
 yew-router = "0.20"
 zbus-secret-service-keyring-store = { version = "1.0.0", features = ["rt-async-io-crypto-rust"] }
+zeroize = "1"
 zip = { version = "8.6.0", default-features = false, features = ["deflate"] }
 
 [profile.release]

core/binary_protocol/src/consensus/command.rs

@@ -40,6 +40,11 @@ pub enum Command2 {
     DoViewChange = 11,
     StartView = 12,
     Eviction = 13,
+
+    // Replica-to-replica auth handshake (server-ng consensus plane).
+    ReplicaHello = 14,
+    ReplicaChallenge = 15,
+    ReplicaFinish = 16,
 }
 
 // SAFETY: Command2 is #[repr(u8)] with no padding bytes.
@@ -50,7 +55,7 @@ unsafe impl CheckedBitPattern for Command2 {
     type Bits = u8;
 
     fn is_valid_bit_pattern(bits: &u8) -> bool {
-        *bits <= 13
+        *bits <= Self::ReplicaFinish as u8
     }
 }
 
@@ -68,4 +73,19 @@ mod tests {
         let result = bytemuck::checked::try_from_bytes::<GenericHeader>(&buf);
         assert!(result.is_err());
     }
+
+    #[test]
+    fn replica_auth_commands_are_valid_bit_patterns() {
+        // Locks the is_valid_bit_pattern bump: 14/15/16 parse, 17 still rejects.
+        for command in 14u8..=16 {
+            let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
+            buf.resize(256, 0);
+            buf[60] = command;
+            assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_ok());
+        }
+        let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
+        buf.resize(256, 0);
+        buf[60] = 17;
+        assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_err());
+    }
 }

core/binary_protocol/src/consensus/header.rs

@@ -25,6 +25,10 @@ use std::mem::offset_of;
 
 pub const HEADER_SIZE: usize = 256;
 
+/// Length of [`GenericHeader::reserved_command`], the per-command scratch area
+/// the replica-auth handshake writes its nonce / MAC / reject-reason into.
+pub const RESERVED_COMMAND_LEN: usize = 128;
+
 /// Byte offset of [`GenericHeader::size`] within the on-wire header.
 ///
 /// Single source of truth for transports that decode the size field
@@ -89,7 +93,7 @@ pub struct GenericHeader {
     pub command: Command2,
     pub replica: u8,
     pub reserved_frame: [u8; 66],
-    pub reserved_command: [u8; 128],
+    pub reserved_command: [u8; RESERVED_COMMAND_LEN],
 }
 const _: () = {
     assert!(size_of::<GenericHeader>() == HEADER_SIZE);
@@ -101,7 +105,10 @@ const _: () = {
         offset_of!(GenericHeader, reserved_command)
             == offset_of!(GenericHeader, reserved_frame) + size_of::<[u8; 66]>()
     );
-    assert!(offset_of!(GenericHeader, reserved_command) + size_of::<[u8; 128]>() == HEADER_SIZE);
+    assert!(
+        offset_of!(GenericHeader, reserved_command) + size_of::<[u8; RESERVED_COMMAND_LEN]>()
+            == HEADER_SIZE
+    );
 };
 
 impl ConsensusHeader for GenericHeader {

core/binary_protocol/src/consensus/mod.rs

@@ -45,7 +45,7 @@ pub use command::Command2;
 pub use error::ConsensusError;
 pub use header::{
     CommitHeader, ConsensusHeader, DoViewChangeHeader, EvictionHeader, EvictionReason,
-    GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader, ReplyHeader, RequestHeader,
-    SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader, read_size_field,
+    GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader, RESERVED_COMMAND_LEN, ReplyHeader,
+    RequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader, read_size_field,
 };
 pub use operation::Operation;

core/binary_protocol/src/lib.rs

@@ -72,8 +72,8 @@ pub use codec::{WireDecode, WireEncode};
 pub use consensus::{
     Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, EvictionHeader,
     EvictionReason, GenericHeader, HEADER_SIZE, Operation, PrepareHeader, PrepareOkHeader,
-    ReplyHeader, RequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
-    read_size_field,
+    RESERVED_COMMAND_LEN, ReplyHeader, RequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
+    StartViewHeader, read_size_field,
 };
 pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};
 pub use error::WireError;

core/configs/src/server_config/cluster.rs

@@ -27,8 +27,47 @@ pub struct ClusterConfig {
     /// every node so operators ship one config. The running node's identity
     /// is supplied out-of-band via the `--replica-id` CLI flag, which
     /// selects the entry in this list that describes the current node.
+    //
+    // TODO(hubcio): IGGY-155 `register-replica` CLI (a validated roster
+    // append) is deferred - it is convenience only over a manual TOML edit,
+    // and `ClusterConfig::validate` already rejects a malformed roster at
+    // boot. Add it only if scripted/automated roster edits become a need.
     #[serde(default)]
     pub nodes: Vec<ClusterNodeConfig>,
+    /// Replica-to-replica authentication settings (PSK + BLAKE3 handshake).
+    #[serde(default)]
+    pub auth: ClusterAuthConfig,
+}
+
+/// Replica-to-replica authentication for the consensus (`tcp_replica`) port.
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
+pub struct ClusterAuthConfig {
+    /// When true, every replica peer must complete the authenticated handshake
+    /// or be rejected, and [`Self::shared_secret`] is mandatory. When false
+    /// (default) the replica handshake stays in legacy unauthenticated mode and
+    /// `shared_secret` is not used for authentication. A configured non-empty
+    /// `shared_secret` must still meet the 32-byte minimum whenever the cluster
+    /// is enabled (a short value fails boot even with auth off).
+    ///
+    /// Enabling auth is a coordinated-restart change, and not the only one: the
+    /// consensus `cluster_id` is derived from `ClusterConfig::name`
+    /// unconditionally, so a mixed-version roster fails to connect regardless of
+    /// this flag. Flip every node in one restart.
+    #[serde(default)]
+    pub enabled: bool,
+    /// Cluster-wide pre-shared key for replica-to-replica authentication.
+    ///
+    /// At least 32 bytes of CSPRNG output, byte-identical across every node.
+    /// Provisioned out-of-band, normally via `IGGY_CLUSTER_AUTH_SHARED_SECRET`
+    /// rather than the on-disk config.
+    // skip_serializing keeps the PSK out of the runtime `current_config.toml`
+    // (and the `ServerConfig` diagnostic snapshot that cats it). The live
+    // secret is read from env / on-disk config at boot, never from the
+    // snapshot, so it must never be persisted there. Deserialize is retained.
+    #[serde(default, skip_serializing)]
+    #[config_env(secret)]
+    pub shared_secret: String,
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
@@ -52,3 +91,35 @@ pub struct TransportPorts {
     /// Dedicated port for replica-to-replica consensus traffic.
     pub tcp_replica: Option<u16>,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn shared_secret_is_never_serialized() {
+        // Regression guard: the runtime current_config.toml (and the
+        // ServerConfig diagnostic snapshot that cats it) are produced by
+        // serializing this struct, so the PSK must not survive serialize.
+        // skip_serializing is format-agnostic, so a JSON dump proves the toml
+        // path too.
+        let config = ClusterConfig {
+            enabled: true,
+            name: "iggy-cluster".to_owned(),
+            nodes: Vec::new(),
+            auth: ClusterAuthConfig {
+                enabled: true,
+                shared_secret: "current-psk-MUST-NOT-be-persisted".to_owned(),
+            },
+        };
+        let serialized = serde_json::to_string(&config).expect("serialize cluster config");
+        assert!(
+            !serialized.contains("MUST-NOT-be-persisted"),
+            "PSK leaked into serialized config: {serialized}"
+        );
+        assert!(
+            !serialized.contains("shared_secret"),
+            "shared_secret field present in serialized config: {serialized}"
+        );
+    }
+}

core/configs/src/server_config/defaults.rs

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

core/configs/src/server_config/http.rs

@@ -75,8 +75,13 @@ pub struct HttpJwtConfig {
     #[config_env(leaf)]
     #[serde_as(as = "DisplayFromStr")]
     pub not_before: IggyDuration,
+    // skip_serializing keeps the secrets out of the runtime current_config.toml
+    // (and the diagnostic snapshot that cats it). The live secrets are read from
+    // env / on-disk config at boot, never from the snapshot.
+    #[serde(default, skip_serializing)]
     #[config_env(secret)]
     pub encoding_secret: String,
+    #[serde(default, skip_serializing)]
     #[config_env(secret)]
     pub decoding_secret: String,
     pub use_base64_secret: bool,
@@ -158,3 +163,36 @@ impl HttpJwtConfig {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn jwt_secrets_are_never_serialized() {
+        // current_config.toml (and the diagnostic snapshot that cats it) is
+        // produced by serializing this struct, so the JWT secrets must not
+        // survive a serialize. Built via deserialize to skip enumerating the
+        // duration fields; skip_serializing is format-agnostic, so a JSON dump
+        // proves the toml path too.
+        let json = r#"{
+            "algorithm": "HS256",
+            "issuer": "iggy.apache.org",
+            "audience": "iggy.apache.org",
+            "valid_issuers": ["iggy.apache.org"],
+            "valid_audiences": ["iggy.apache.org"],
+            "access_token_expiry": "1 h",
+            "clock_skew": "5 s",
+            "not_before": "0 s",
+            "encoding_secret": "encoding-MUST-NOT-be-persisted",
+            "decoding_secret": "decoding-MUST-NOT-be-persisted",
+            "use_base64_secret": false
+        }"#;
+        let config: HttpJwtConfig = serde_json::from_str(json).expect("deserialize jwt config");
+        let serialized = serde_json::to_string(&config).expect("serialize jwt config");
+        assert!(
+            !serialized.contains("MUST-NOT-be-persisted"),
+            "JWT secret leaked into serialized config: {serialized}"
+        );
+    }
+}

core/configs/src/server_config/system.rs

@@ -103,6 +103,10 @@ pub struct LoggingConfig {
 #[derive(Debug, Deserialize, Serialize, ConfigEnv)]
 pub struct EncryptionConfig {
     pub enabled: bool,
+    // skip_serializing keeps the key out of the runtime current_config.toml (and
+    // the diagnostic snapshot that cats it). The live key is read from env /
+    // on-disk config at boot, never from the snapshot.
+    #[serde(default, skip_serializing)]
     #[config_env(secret)]
     pub key: String,
 }
@@ -354,3 +358,25 @@ impl SystemPaths for SystemConfig {
         SystemConfig::get_runtime_path(self)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn encryption_key_is_never_serialized() {
+        // current_config.toml (and the diagnostic snapshot that cats it) is
+        // produced by serializing this struct, so the key must not survive a
+        // serialize. skip_serializing is format-agnostic, so a JSON dump proves
+        // the toml path too.
+        let config = EncryptionConfig {
+            enabled: true,
+            key: "encryption-key-MUST-NOT-be-persisted".to_owned(),
+        };
+        let serialized = serde_json::to_string(&config).expect("serialize encryption config");
+        assert!(
+            !serialized.contains("MUST-NOT-be-persisted"),
+            "encryption key leaked into serialized config: {serialized}"
+        );
+    }
+}