feat(heal): gate control capability by cluster topology (#4994)

* fix(rpc): bind internode auth to exact targets

* fix(heal): initialize the runtime atomically

* fix(heal): aggregate status across cluster nodes

* fix(heal): return canonical tokens for duplicate starts

* feat(heal): add authenticated control RPC contract

* feat(heal): gate control capability by cluster topology

* fix(heal): return canonical tokens for duplicate starts (#4992)

---------

Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

crates/ecstore/src/api/mod.rs

@@ -385,7 +385,8 @@ pub mod rpc {
         SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, gen_signature_headers,
         gen_tonic_signature_headers, gen_tonic_signature_interceptor, node_service_time_out_client,
         node_service_time_out_client_no_auth, normalize_tonic_rpc_audience, set_tonic_canonical_body_digest,
-        verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_rpc_signature,
+        sign_tonic_rpc_response_proof, verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_rpc_response_proof,
+        verify_tonic_rpc_signature,
     };
 }
 

crates/ecstore/src/cluster/rpc/http_auth.rs

@@ -53,6 +53,7 @@ const RPC_SIGNATURE_V2_HEADER: &str = "x-rustfs-rpc-signature-v2";
 const RPC_NONCE_HEADER: &str = "x-rustfs-rpc-nonce";
 pub(crate) const RPC_CONTENT_SHA256_HEADER: &str = "x-rustfs-content-sha256";
 const RPC_AUTH_VERSION_V2: &str = "2";
+const RPC_RESPONSE_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-response-proof-v1\0";
 const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
 const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned";
 const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes
@@ -138,6 +139,30 @@ fn get_shared_secret() -> std::io::Result<String> {
     })
 }
 
+fn rpc_response_proof_mac(canonical_body: &[u8]) -> std::io::Result<HmacSha256> {
+    let secret = get_shared_secret()?;
+    let mut mac =
+        <HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?;
+    mac.update(RPC_RESPONSE_PROOF_DOMAIN);
+    mac.update(
+        &u64::try_from(canonical_body.len())
+            .map_err(|_| std::io::Error::other("RPC response proof length cannot be represented"))?
+            .to_be_bytes(),
+    );
+    mac.update(canonical_body);
+    Ok(mac)
+}
+
+pub fn sign_tonic_rpc_response_proof(canonical_body: &[u8]) -> std::io::Result<Vec<u8>> {
+    Ok(rpc_response_proof_mac(canonical_body)?.finalize().into_bytes().to_vec())
+}
+
+pub fn verify_tonic_rpc_response_proof(canonical_body: &[u8], proof: &[u8]) -> std::io::Result<()> {
+    rpc_response_proof_mac(canonical_body)?
+        .verify_slice(proof)
+        .map_err(|_| std::io::Error::other("Invalid RPC response proof"))
+}
+
 /// Build the canonical payload covered by the RPC HMAC.
 fn signature_payload(url: &str, method: &Method, timestamp: i64) -> String {
     let uri: Uri = url.parse().expect("Invalid URL");

crates/ecstore/src/cluster/rpc/mod.rs

@@ -30,7 +30,8 @@ pub use client::{
 };
 pub use http_auth::{
     TONIC_RPC_PREFIX, build_auth_headers, gen_signature_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience,
-    set_tonic_canonical_body_digest, verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_rpc_signature,
+    set_tonic_canonical_body_digest, sign_tonic_rpc_response_proof, verify_rpc_signature, verify_tonic_canonical_body_digest,
+    verify_tonic_rpc_response_proof, verify_tonic_rpc_signature,
 };
 #[cfg(test)]
 pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport;

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -15,7 +15,7 @@
 use crate::cluster::rpc::client::{
     TonicInterceptor, gen_tonic_signature_interceptor, heal_control_time_out_client, node_service_time_out_client,
 };
-use crate::cluster::rpc::set_tonic_canonical_body_digest;
+use crate::cluster::rpc::{set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof};
 use crate::error::{Error, Result};
 use crate::{
     disk::disk_store::{get_drive_active_check_interval, get_drive_active_check_timeout},
@@ -94,6 +94,11 @@ fn decode_scanner_activity(response: ScannerActivityResponse) -> Result<ScannerP
     })
 }
 
+fn validate_heal_control_capability_proof(canonical_ack: &[u8], proof: &[u8]) -> Result<()> {
+    verify_tonic_rpc_response_proof(canonical_ack, proof)
+        .map_err(|_| Error::other("peer returned an invalid heal control capability proof"))
+}
+
 #[derive(Clone, Debug)]
 pub struct PeerLiveEventsBatch {
     pub events: Vec<u8>,
@@ -762,6 +767,24 @@ impl PeerRestClient {
         .await
     }
 
+    /// Confirms that a peer supports heal-control v1 and has the same storage
+    /// topology. Every non-success response is an error so old or divergent
+    /// peers cannot be mistaken for compatible ones.
+    pub async fn probe_heal_control(&self, topology_fingerprint: String) -> Result<()> {
+        let nonce = uuid::Uuid::new_v4();
+        let probe = rustfs_protos::heal_control_capability_probe(nonce.as_bytes());
+        let canonical_ack = rustfs_protos::canonical_heal_control_capability_ack(
+            rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION,
+            &topology_fingerprint,
+            &probe,
+        )
+        .map_err(|_| Error::other("heal control capability acknowledgement length cannot be represented"))?;
+        let proof = self
+            .heal_control(rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, topology_fingerprint, probe)
+            .await?;
+        validate_heal_control_capability_proof(&canonical_ack, &proof)
+    }
+
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
         self.finalize_result(
             async {
@@ -1374,6 +1397,16 @@ mod tests {
         assert!(!client.offline.load(Ordering::Acquire));
     }
 
+    #[test]
+    fn heal_control_capability_proof_must_authenticate_exact_ack() {
+        runtime_sources::ensure_test_rpc_secret();
+        let proof = crate::cluster::rpc::sign_tonic_rpc_response_proof(b"expected").expect("test proof should sign");
+        assert!(validate_heal_control_capability_proof(b"expected", &proof).is_ok());
+        let err = validate_heal_control_capability_proof(b"different", &proof)
+            .expect_err("a proof for a different acknowledgement must fail closed");
+        assert!(err.to_string().contains("invalid heal control capability proof"));
+    }
+
     #[tokio::test]
     async fn peer_rest_client_prepare_retry_clears_offline_gate() {
         // finalize_result sets the offline gate on a network error; without

crates/ecstore/src/store/mod.rs

@@ -257,6 +257,11 @@ impl ECStore {
         runtime_sources::endpoint_pools().unwrap_or_else(|| Vec::new().into())
     }
 
+    /// Get this store instance's endpoint topology without consulting process globals.
+    pub fn instance_endpoints(&self) -> Option<EndpointServerPools> {
+        self.ctx.endpoints()
+    }
+
     /// Get the global region
     pub fn region(&self) -> Option<s3s::region::Region> {
         runtime_sources::region()
@@ -897,6 +902,22 @@ mod tests {
         let ctx_b = Arc::new(InstanceContext::new());
         ctx_a.update_erasure_type(SetupType::DistErasure).await;
         ctx_b.update_erasure_type(SetupType::ErasureSD).await;
+        ctx_a.set_endpoints(EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 1,
+            endpoints: Endpoints::default(),
+            cmd_line: "instance-a".to_string(),
+            platform: String::new(),
+        }]));
+        ctx_b.set_endpoints(EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: true,
+            set_count: 2,
+            drives_per_set: 2,
+            endpoints: Endpoints::default(),
+            cmd_line: "instance-b".to_string(),
+            platform: String::new(),
+        }]));
 
         let store_a = build_store_with_ctx(ctx_a);
         let store_b = build_store_with_ctx(ctx_b);
@@ -910,6 +931,12 @@ mod tests {
         assert!(store_b.setup_is_erasure_sd().await);
         assert!(!store_b.setup_is_erasure().await);
         assert!(!store_b.setup_is_dist_erasure().await);
+        let endpoints_a = store_a.instance_endpoints().expect("instance A endpoints");
+        let endpoints_b = store_b.instance_endpoints().expect("instance B endpoints");
+        assert_eq!(endpoints_a.as_ref()[0].set_count, 1);
+        assert_eq!(endpoints_b.as_ref()[0].set_count, 2);
+        assert!(!endpoints_a.as_ref()[0].legacy);
+        assert!(endpoints_b.as_ref()[0].legacy);
     }
 
     // The production/test constructors ADOPT the process bootstrap context

crates/protos/src/lib.rs

@@ -141,6 +141,19 @@ pub fn internode_rpc_max_message_size() -> usize {
 }
 
 pub const HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE: usize = 65 * 1024;
+pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 1;
+pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v1\0";
+
+pub fn heal_control_capability_probe(nonce: &[u8; 16]) -> Vec<u8> {
+    let mut probe = Vec::with_capacity(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX.len() + nonce.len());
+    probe.extend_from_slice(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX);
+    probe.extend_from_slice(nonce);
+    probe
+}
+
+pub fn is_heal_control_capability_probe(command: &[u8]) -> bool {
+    command.len() == HEAL_CONTROL_CAPABILITY_PROBE_PREFIX.len() + 16 && command.starts_with(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX)
+}
 
 /// Builds the stable byte representation authenticated for a heal-control request.
 ///
@@ -164,9 +177,32 @@ pub fn canonical_heal_control_request_body(
     Ok(body)
 }
 
+/// Builds the exact acknowledgement expected from a peer that supports the
+/// heal-control protocol and agrees with the caller's storage topology.
+pub fn canonical_heal_control_capability_ack(
+    version: u32,
+    topology_fingerprint: &str,
+    probe: &[u8],
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    const DOMAIN: &[u8] = b"rustfs-heal-control-capability-ack-v1\0";
+
+    let fingerprint = topology_fingerprint.as_bytes();
+    let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + fingerprint.len() + 8 + probe.len());
+    body.extend_from_slice(DOMAIN);
+    body.extend_from_slice(&version.to_be_bytes());
+    body.extend_from_slice(&u64::try_from(fingerprint.len())?.to_be_bytes());
+    body.extend_from_slice(fingerprint);
+    body.extend_from_slice(&u64::try_from(probe.len())?.to_be_bytes());
+    body.extend_from_slice(probe);
+    Ok(body)
+}
+
 #[cfg(test)]
 mod heal_control_tests {
-    use super::canonical_heal_control_request_body;
+    use super::{
+        HEAL_CONTROL_CAPABILITY_PROBE_PREFIX, canonical_heal_control_capability_ack, canonical_heal_control_request_body,
+        heal_control_capability_probe, is_heal_control_capability_probe,
+    };
 
     #[test]
     fn canonical_heal_control_body_binds_every_field_and_boundary() {
@@ -196,6 +232,27 @@ mod heal_control_tests {
             canonical_heal_control_request_body(1, "a", b"bc").expect("small request should encode")
         );
     }
+
+    #[test]
+    fn canonical_capability_ack_binds_version_and_topology() {
+        let probe = heal_control_capability_probe(&[7; 16]);
+        let ack = canonical_heal_control_capability_ack(1, "ab", &probe).expect("small acknowledgement should encode");
+        let mut golden = b"rustfs-heal-control-capability-ack-v1\0".to_vec();
+        golden.extend_from_slice(&1_u32.to_be_bytes());
+        golden.extend_from_slice(&2_u64.to_be_bytes());
+        golden.extend_from_slice(b"ab");
+        golden.extend_from_slice(&u64::try_from(probe.len()).unwrap().to_be_bytes());
+        golden.extend_from_slice(&probe);
+        assert_eq!(ack, golden);
+        assert_ne!(ack, canonical_heal_control_capability_ack(2, "ab", &probe).unwrap());
+        assert_ne!(ack, canonical_heal_control_capability_ack(1, "ac", &probe).unwrap());
+        assert_ne!(
+            ack,
+            canonical_heal_control_capability_ack(1, "ab", &heal_control_capability_probe(&[8; 16])).unwrap()
+        );
+        assert!(is_heal_control_capability_probe(&probe));
+        assert!(!is_heal_control_capability_probe(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX));
+    }
 }
 
 /// Whether internode metadata RPCs should send only the msgpack `_bin` payloads and leave the JSON