From 908ca548bbda14ac6c21a9cd0765576f6b457948 Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 20 Jul 2026 13:04:30 +0800 Subject: [PATCH] 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 --- crates/ecstore/src/api/mod.rs | 3 +- crates/ecstore/src/cluster/rpc/http_auth.rs | 25 ++ crates/ecstore/src/cluster/rpc/mod.rs | 3 +- .../src/cluster/rpc/peer_rest_client.rs | 35 ++- crates/ecstore/src/store/mod.rs | 27 ++ crates/protos/src/lib.rs | 59 ++++- rustfs/src/app/context/server_slot.rs | 16 ++ rustfs/src/server/http.rs | 70 +++++- rustfs/src/startup_iam.rs | 113 +++++++-- rustfs/src/storage/rpc/node_service.rs | 238 +++++++++++++++++- rustfs/src/storage/rpc/node_service/heal.rs | 212 +++++++++++++++- rustfs/src/storage/storage_api.rs | 22 +- rustfs/src/storage/tonic_service.rs | 3 + rustfs/src/storage_api.rs | 18 +- 14 files changed, 805 insertions(+), 39 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index dca357741..4f85333ff 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, }; } diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 2dd609443..4f479101f 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/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 { }) } +fn rpc_response_proof_mac(canonical_body: &[u8]) -> std::io::Result { + let secret = get_shared_secret()?; + let mut mac = + ::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> { + 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"); diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index a4a1539c6..67616dec4 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/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; diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index c5b35079f..c8820cf2a 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/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 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, @@ -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 diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 9e3b2b244..90e851c17 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/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 { + self.ctx.endpoints() + } + /// Get the global region pub fn region(&self) -> Option { 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 diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 2ae7fb03e..50da7d209 100644 --- a/crates/protos/src/lib.rs +++ b/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 { + 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, 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 diff --git a/rustfs/src/app/context/server_slot.rs b/rustfs/src/app/context/server_slot.rs index 34f6f1ee4..1d204be34 100644 --- a/rustfs/src/app/context/server_slot.rs +++ b/rustfs/src/app/context/server_slot.rs @@ -34,6 +34,7 @@ use std::sync::{Arc, OnceLock}; #[derive(Default)] pub struct ServerContextSlot { app_context: OnceLock>, + heal_topology_fingerprint: Arc>, } // Manual Debug: AppContext is not Debug, so summarize installation state. @@ -41,6 +42,7 @@ impl std::fmt::Debug for ServerContextSlot { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ServerContextSlot") .field("app_context_installed", &self.app_context.get().is_some()) + .field("heal_topology_fingerprint_initialized", &self.heal_topology_fingerprint.get().is_some()) .finish() } } @@ -49,6 +51,7 @@ impl ServerContextSlot { pub fn new() -> Arc { Arc::new(Self { app_context: OnceLock::new(), + heal_topology_fingerprint: Arc::new(tokio::sync::OnceCell::new()), }) } @@ -69,11 +72,16 @@ impl ServerContextSlot { pub fn object_store(&self) -> Option> { self.app_context().map(|context| context.object_store()) } + + pub fn heal_topology_fingerprint(&self) -> Arc> { + Arc::clone(&self.heal_topology_fingerprint) + } } #[cfg(test)] mod tests { use super::{ServerContextSlot, get_global_app_context}; + use std::sync::Arc; // Before anything is installed — and with no global AppContext in this // process (nextest runs each test in its own process) — resolution yields @@ -85,4 +93,12 @@ mod tests { assert_eq!(slot.app_context().is_some(), get_global_app_context().is_some()); assert_eq!(slot.object_store().is_some(), get_global_app_context().is_some()); } + + #[test] + fn heal_topology_cache_is_owned_by_each_server_slot() { + let first = ServerContextSlot::new(); + let second = ServerContextSlot::new(); + + assert!(!Arc::ptr_eq(&first.heal_topology_fingerprint(), &second.heal_topology_fingerprint())); + } } diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 3df55d6e7..1654efcf2 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -35,7 +35,7 @@ use crate::server::{ use crate::storage_api::server::http as storage; use crate::storage_api::server::http::request_context::{RequestContext, extract_request_id_from_headers}; use crate::storage_api::server::http::rpc::InternodeRpcService; -use crate::storage_api::server::http::tonic_service::{make_heal_control_server, make_server}; +use crate::storage_api::server::http::tonic_service::make_server; use crate::storage_api::server::http::{ ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, verify_tonic_rpc_signature, }; @@ -685,7 +685,7 @@ pub async fn start_http_server( // Bind the S3 service to this server's context slot (backlog#1052 S2) // so request dispatch resolves the server's own store. let admin_server_ctx = server_ctx.clone(); - let store = storage::ecfs::FS::with_server_ctx(server_ctx); + let store = storage::ecfs::FS::with_server_ctx(server_ctx.clone()); let mut b = S3ServiceBuilder::new(store.clone()); let access_key = config.access_key.clone(); @@ -1044,6 +1044,7 @@ pub async fn start_http_server( keystone_auth: auth_keystone::get_keystone_auth(), trusted_proxy_layer: rustfs_trusted_proxies::is_enabled().then(|| rustfs_trusted_proxies::layer().clone()), rate_limit_layer: api_rate_limit_layer.clone(), + server_ctx: Arc::clone(&server_ctx), }; process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.watcher(), connection_permit); @@ -1103,6 +1104,7 @@ struct ConnectionContext { /// Per-client API rate limit layer; `None` when disabled (the default). /// All clones share one limiter, keeping budgets global across connections. rate_limit_layer: Option, + server_ctx: Arc, } #[derive(Clone)] @@ -1292,6 +1294,7 @@ fn process_connection( keystone_auth, trusted_proxy_layer, rate_limit_layer, + server_ctx, } = context; // Build the hybrid service per-connection. @@ -1312,9 +1315,11 @@ fn process_connection( ); let heal_control_max_message_size = rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE; let heal_control_service = InterceptedService::new( - HealControlServiceServer::new(make_heal_control_server()) - .max_decoding_message_size(heal_control_max_message_size) - .max_encoding_message_size(heal_control_max_message_size), + HealControlServiceServer::new(storage::tonic_service::make_heal_control_server_with_cache( + server_ctx.heal_topology_fingerprint(), + )) + .max_decoding_message_size(heal_control_max_message_size) + .max_encoding_message_size(heal_control_max_message_size), check_auth, ); let rpc_service = RpcRequestPathService::new(Routes::new(node_service).add_service(heal_control_service).prepare()); @@ -1980,6 +1985,8 @@ mod tests { use std::future::Ready; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; + use storage::tonic_service::{heal_topology_fingerprint, make_heal_control_server_for_source}; + use storage::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; use tower::{Layer, Service, ServiceBuilder}; /// Baseline constants — reference the authoritative config defaults. @@ -2323,9 +2330,25 @@ mod tests { let previous_node_name = rustfs_common::get_global_local_node_name().await; rustfs_common::set_global_local_node_name(&addr.to_string()).await; + let endpoint_pools = EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 1, + endpoints: Endpoints::from(vec![{ + let mut endpoint = Endpoint::try_from("http://node-a:9000/disk1").expect("test endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + endpoint + }]), + cmd_line: String::new(), + platform: String::new(), + }]); + let fingerprint = heal_topology_fingerprint(&endpoint_pools).expect("test topology should hash"); + let (heal_control_server, endpoint_pools_source) = make_heal_control_server_for_source(); let node_service = InterceptedService::new(NodeServiceServer::new(make_server()), check_auth); let heal_control_service = InterceptedService::new( - HealControlServiceServer::new(make_heal_control_server()) + HealControlServiceServer::new(heal_control_server) .max_decoding_message_size(rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE) .max_encoding_message_size(rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE), check_auth, @@ -2343,6 +2366,14 @@ mod tests { let grid_host = format!("http://{addr}"); let host = rustfs_utils::XHost::try_from(addr.to_string()).expect("test address should resolve"); let client = storage::PeerRestClient::new(host, grid_host); + let not_ready = client + .probe_heal_control(fingerprint.clone()) + .await + .expect_err("an uninitialized topology must fail closed"); + assert!(not_ready.to_string().contains("topology is not initialized")); + assert!(!not_ready.to_string().contains("temporarily offline")); + *endpoint_pools_source.write().await = Some(endpoint_pools); + for _ in 0..2 { let error = client .heal_control(1, "fingerprint".to_string(), b"query".to_vec()) @@ -2352,6 +2383,33 @@ mod tests { assert!(message.contains("routing is not enabled")); assert!(!message.contains("temporarily offline")); } + client + .probe_heal_control(fingerprint.clone()) + .await + .expect("production client should accept an exact capability acknowledgement"); + + let mismatch = client + .probe_heal_control("different-topology".to_string()) + .await + .expect_err("a divergent topology must fail closed"); + assert!(mismatch.to_string().contains("topology does not match")); + assert!(!mismatch.to_string().contains("temporarily offline")); + + let unsupported = client + .heal_control( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION + 1, + fingerprint.clone(), + rustfs_protos::heal_control_capability_probe(&[9; 16]), + ) + .await + .expect_err("an unsupported protocol version must fail closed"); + assert!(unsupported.to_string().contains("unsupported heal control protocol version")); + assert!(!unsupported.to_string().contains("temporarily offline")); + + client + .probe_heal_control(fingerprint) + .await + .expect("semantic probe failures must not mark a reachable peer offline"); client.evict_connection().await; server.abort(); diff --git a/rustfs/src/startup_iam.rs b/rustfs/src/startup_iam.rs index 53737bee8..4dffb7f71 100644 --- a/rustfs/src/startup_iam.rs +++ b/rustfs/src/startup_iam.rs @@ -38,9 +38,12 @@ const IAM_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30); /// After this many retries (~5 min at initial interval), escalate log level to ERROR. const IAM_RETRY_ESCALATION_THRESHOLD: u64 = 12; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct HealTopologyReady(()); + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum IamBootstrapDisposition { - ReadyInline, + ReadyInline(HealTopologyReady), Deferred, } @@ -63,12 +66,23 @@ where PublishFn: FnOnce() -> PublishFuture, PublishFuture: Future>, { - if disposition == IamBootstrapDisposition::ReadyInline { - publish_ready().await?; - return Ok(true); + if let IamBootstrapDisposition::ReadyInline(topology_ready) = disposition { + publish_runtime_ready_after_heal_topology(topology_ready, publish_ready).await?; + Ok(true) + } else { + Ok(false) } +} - Ok(false) +async fn publish_runtime_ready_after_heal_topology( + _topology_ready: HealTopologyReady, + publish_ready: PublishFn, +) -> Result<()> +where + PublishFn: FnOnce() -> PublishFuture, + PublishFuture: Future>, +{ + publish_ready().await } async fn finalize_iam_recovery( @@ -81,10 +95,32 @@ async fn finalize_iam_recovery( // Resolve the globally published system directly (not context-first — // the AppContext does not exist yet; creating it is this function's job). let iam = rustfs_iam::get().map_err(|_| std::io::Error::other("IAM recovered but unavailable"))?; + let endpoint_pools = heal_control_endpoint_pools(store.as_ref())?; AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?; + let topology_ready = + publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools).await?; + publish_runtime_ready_after_heal_topology(topology_ready, || async { + publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await + }) + .await +} +async fn publish_iam_ready_after_heal_topology( + readiness: &GlobalReadiness, + cache: Arc>, + endpoint_pools: crate::storage_api::cluster::EndpointServerPools, +) -> Result { + crate::storage_api::startup::heal_control::initialize_heal_topology_fingerprint(cache, endpoint_pools) + .await + .map_err(std::io::Error::other)?; readiness.mark_stage(SystemStage::IamReady); - publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await + Ok(HealTopologyReady(())) +} + +fn heal_control_endpoint_pools(store: &ECStore) -> Result { + store + .instance_endpoints() + .ok_or_else(|| std::io::Error::other("heal control topology is unavailable")) } fn compute_backoff_interval(attempt: u64, initial: Duration, max: Duration) -> Duration { @@ -336,7 +372,7 @@ async fn attempt_init_iam_sys( /// /// Returns `Ok(ReadyInline)` if IAM initialized immediately, `Ok(Deferred)` if /// recovery is happening in the background, or `Err` if IAM succeeded but -/// app context initialization failed (unexpected, indicates a bug). +/// the instance context or heal topology could not be initialized. pub(crate) async fn bootstrap_or_defer_iam_init( store: Arc, kms_interface: Arc, @@ -347,9 +383,12 @@ pub(crate) async fn bootstrap_or_defer_iam_init( ) -> Result { match attempt_init_iam_sys(store.clone()).await { Ok(iam) => { + let endpoint_pools = heal_control_endpoint_pools(store.as_ref())?; AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?; - readiness.mark_stage(SystemStage::IamReady); - return Ok(IamBootstrapDisposition::ReadyInline); + let topology_ready = + publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools) + .await?; + return Ok(IamBootstrapDisposition::ReadyInline(topology_ready)); } Err(err) => { let interval = initial_retry_interval(); @@ -445,6 +484,7 @@ mod hint_tests { mod tests { use super::*; use super::{IAM_RETRY_ESCALATION_THRESHOLD, IAM_RETRY_INITIAL_INTERVAL, IAM_RETRY_MAX_INTERVAL, compute_backoff_interval}; + use crate::storage_api::cluster::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; use rustfs_common::{GlobalReadiness, SystemStage}; use std::io::Error; use std::sync::{ @@ -483,10 +523,13 @@ mod tests { let publish_calls = Arc::new(AtomicUsize::new(0)); let publish_calls_for_assert = publish_calls.clone(); - let published = publish_ready_for_iam_bootstrap_with(IamBootstrapDisposition::ReadyInline, move || async move { - publish_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - }) + let published = publish_ready_for_iam_bootstrap_with( + IamBootstrapDisposition::ReadyInline(HealTopologyReady(())), + move || async move { + publish_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + }, + ) .await; assert!(published.is_ok(), "ready inline publication should succeed"); @@ -514,7 +557,7 @@ mod tests { #[tokio::test] async fn ready_inline_bootstrap_propagates_runtime_readiness_failure() { - let err = publish_ready_for_iam_bootstrap_with(IamBootstrapDisposition::ReadyInline, || async { + let err = publish_ready_for_iam_bootstrap_with(IamBootstrapDisposition::ReadyInline(HealTopologyReady(())), || async { Err(Error::other("runtime readiness failed")) }) .await @@ -523,6 +566,48 @@ mod tests { assert_eq!(err.to_string(), "runtime readiness failed"); } + #[tokio::test] + async fn iam_readiness_is_published_only_after_heal_topology_initialization() { + let readiness = GlobalReadiness::new(); + let cache = Arc::new(tokio::sync::OnceCell::new()); + let mut endpoint = Endpoint::try_from("http://node-a:9000/disk").expect("valid test endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + let endpoint_pools = EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 1, + endpoints: Endpoints::from(vec![endpoint]), + cmd_line: "test".to_string(), + platform: String::new(), + }]); + let expected = crate::storage_api::startup::heal_control::heal_topology_fingerprint(&endpoint_pools) + .expect("valid topology should hash"); + + publish_iam_ready_after_heal_topology(&readiness, Arc::clone(&cache), endpoint_pools) + .await + .expect("topology initialization should publish IAM readiness"); + + assert_eq!(cache.get(), Some(&expected)); + assert!(matches!(readiness.current_stage(), SystemStage::IamReady)); + + let failed_readiness = GlobalReadiness::new(); + let invalid_cache = Arc::new(tokio::sync::OnceCell::new()); + publish_iam_ready_after_heal_topology( + &failed_readiness, + Arc::clone(&invalid_cache), + EndpointServerPools::from(Vec::::new()), + ) + .await + .expect_err("invalid topology must block IAM readiness"); + assert!(invalid_cache.get().is_none()); + assert!(!matches!( + failed_readiness.current_stage(), + SystemStage::IamReady | SystemStage::FullReady + )); + } + #[tokio::test(start_paused = true)] async fn recovery_loop_retries_finalize_until_success() { let init_calls = Arc::new(AtomicUsize::new(0)); diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 45cefd955..745f04933 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -25,8 +25,8 @@ use crate::storage::storage_api::rpc_consumer::node_service::{ SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _, StorageResult, all_local_disk_path, find_local_disk_by_ref, reload_transition_tier_config, }; -use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; -use crate::storage::storage_api::verify_tonic_canonical_body_digest; +use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, runtime_sources}; +use crate::storage::storage_api::{sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest}; use bytes::Bytes; use futures::Stream; use futures_util::future::join_all; @@ -198,11 +198,94 @@ pub fn make_server_for_context(context: Option> NodeService { local_peer, context } } -#[derive(Clone, Copy, Debug, Default)] -pub struct HealControlRpcService; +#[derive(Clone, Debug, Default)] +pub struct HealControlRpcService { + #[cfg(test)] + endpoint_pools: Option, + topology_fingerprint: Arc>, + #[cfg(test)] + endpoint_pools_source: Option>>>, +} pub fn make_heal_control_server() -> HealControlRpcService { - HealControlRpcService + make_heal_control_server_with_cache(Arc::new(tokio::sync::OnceCell::new())) +} + +pub(crate) fn make_heal_control_server_with_cache( + topology_fingerprint: Arc>, +) -> HealControlRpcService { + HealControlRpcService { + topology_fingerprint, + #[cfg(test)] + endpoint_pools: None, + #[cfg(test)] + endpoint_pools_source: None, + } +} + +#[cfg(test)] +pub(crate) fn make_heal_control_server_for_source() +-> (HealControlRpcService, Arc>>) { + let source = Arc::new(tokio::sync::RwLock::new(None)); + ( + HealControlRpcService { + endpoint_pools: None, + topology_fingerprint: Arc::new(tokio::sync::OnceCell::new()), + endpoint_pools_source: Some(Arc::clone(&source)), + }, + source, + ) +} + +impl HealControlRpcService { + async fn capability_fingerprint(&self) -> Result<&str, Status> { + if let Some(fingerprint) = self.topology_fingerprint.get() { + return Ok(fingerprint); + } + + #[cfg(test)] + { + return self + .topology_fingerprint + .get_or_try_init(|| async { + let endpoint_pools = self + .endpoint_pools() + .await + .ok_or_else(|| Status::failed_precondition("heal control topology is not initialized"))?; + tokio::task::spawn_blocking(move || heal::heal_topology_fingerprint(&endpoint_pools)) + .await + .map_err(|_| Status::internal("heal control topology calculation failed"))? + .map_err(|_| Status::failed_precondition("heal control topology is invalid")) + }) + .await + .map(String::as_str); + } + + #[cfg(not(test))] + Err(Status::failed_precondition("heal control topology is not initialized")) + } + + #[cfg(test)] + async fn endpoint_pools(&self) -> Option { + if let Some(source) = self.endpoint_pools_source.as_ref() { + return source.read().await.clone(); + } + self.endpoint_pools.clone() + } +} + +pub(crate) async fn initialize_heal_topology_fingerprint( + cache: Arc>, + endpoint_pools: EndpointServerPools, +) -> Result<(), String> { + if cache.get().is_some() { + return Ok(()); + } + let fingerprint = tokio::task::spawn_blocking(move || heal::heal_topology_fingerprint(&endpoint_pools)) + .await + .map_err(|_| "heal control topology calculation task failed".to_string())??; + let _ = cache.set(fingerprint); + Ok(()) } #[tonic::async_trait] @@ -221,6 +304,28 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { .map_err(|_| Status::invalid_argument("heal control request length cannot be represented"))?; verify_tonic_canonical_body_digest(&request, &body) .map_err(|err| Status::permission_denied(format!("heal control authentication failed: {err}")))?; + if request.get_ref().version != rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION { + return Err(Status::failed_precondition("unsupported heal control protocol version")); + } + if rustfs_protos::is_heal_control_capability_probe(&request.get_ref().command) { + let fingerprint = self.capability_fingerprint().await?; + if request.get_ref().topology_fingerprint != *fingerprint { + return Err(Status::failed_precondition("heal control topology does not match")); + } + let canonical_ack = rustfs_protos::canonical_heal_control_capability_ack( + request.get_ref().version, + fingerprint, + &request.get_ref().command, + ) + .map_err(|_| Status::internal("heal control acknowledgement length cannot be represented"))?; + let result = sign_tonic_rpc_response_proof(&canonical_ack) + .map_err(|_| Status::internal("heal control response proof is unavailable"))?; + return Ok(Response::new(HealControlResponse { + success: true, + result: result.into(), + error_info: None, + })); + } Err(Status::unimplemented("heal control routing is not enabled")) } } @@ -1285,10 +1390,15 @@ mod tests { use super::{ CollectMetricsOpts, Error, HEAL_CONTROL_PAYLOAD_MAX_SIZE, MetricType, Node as _, NodeService, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, STORAGE_CLASS_SUB_SYS, - background_rebalance_start_error_message, make_heal_control_server, make_server, scanner_activity_response, - stop_rebalance_response, + background_rebalance_start_error_message, initialize_heal_topology_fingerprint, make_heal_control_server, + make_heal_control_server_with_cache, make_server, scanner_activity_response, stop_rebalance_response, }; + use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; use crate::storage::storage_api::set_tonic_canonical_body_digest; + use crate::storage::storage_api::{ + Endpoint, + ecstore_layout::{EndpointServerPools, Endpoints, PoolEndpoints}, + }; use bytes::Bytes; use rustfs_protos::models::PingBodyBuilder; use rustfs_protos::proto_gen::node_service::{ @@ -1311,7 +1421,7 @@ mod tests { node_service_client::NodeServiceClient, node_service_server::NodeServiceServer, }; - use std::collections::HashMap; + use std::{collections::HashMap, sync::Arc}; use tokio::net::TcpListener; use tokio_stream::wrappers::TcpListenerStream; use tonic::{Request, Response, Status}; @@ -1335,6 +1445,29 @@ mod tests { }) } + fn heal_control_test_endpoints(last_host: &str) -> EndpointServerPools { + let endpoints = ["node-a", "node-b", "node-c", last_host] + .into_iter() + .enumerate() + .map(|(index, host)| { + let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/disk{}", index + 1).as_str()) + .expect("test endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(index / 2); + endpoint.set_disk_index(index % 2); + endpoint + }) + .collect::>(); + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: String::new(), + platform: String::new(), + }]) + } + fn mark_v2_authenticated(request: &mut Request) { request .metadata_mut() @@ -1377,6 +1510,95 @@ mod tests { assert_eq!(oversized.code(), tonic::Code::InvalidArgument); } + #[tokio::test] + async fn heal_control_probe_requires_exact_topology_and_keeps_commands_disabled() { + let _ = rustfs_credentials::set_global_rpc_secret("heal-control-node-service-test-secret".to_string()); + let endpoints = heal_control_test_endpoints("node-d"); + let fingerprint = heal_topology_fingerprint(&endpoints).expect("test topology should hash"); + let cache = Arc::new(tokio::sync::OnceCell::new()); + initialize_heal_topology_fingerprint(Arc::clone(&cache), endpoints) + .await + .expect("valid topology should initialize"); + let service = make_heal_control_server_with_cache(cache); + + let probe_command = rustfs_protos::heal_control_capability_probe(&[7; 16]); + let mut probe = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: fingerprint.clone(), + command: Bytes::from(probe_command.clone()), + }); + let body = rustfs_protos::canonical_heal_control_request_body( + probe.get_ref().version, + &probe.get_ref().topology_fingerprint, + &probe.get_ref().command, + ) + .expect("probe should encode"); + set_tonic_canonical_body_digest(&mut probe, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut probe); + let response = service + .heal_control(probe) + .await + .expect("matching topology should be acknowledged"); + let canonical_ack = rustfs_protos::canonical_heal_control_capability_ack( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + &fingerprint, + &probe_command, + ) + .expect("acknowledgement should encode"); + crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical_ack, &response.into_inner().result) + .expect("response proof should authenticate the exact acknowledgement"); + + let divergent_probe = rustfs_protos::heal_control_capability_probe(&[8; 16]); + let mut divergent = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: heal_topology_fingerprint(&heal_control_test_endpoints("node-e")) + .expect("divergent topology should hash"), + command: Bytes::from(divergent_probe), + }); + let body = rustfs_protos::canonical_heal_control_request_body( + divergent.get_ref().version, + &divergent.get_ref().topology_fingerprint, + &divergent.get_ref().command, + ) + .expect("probe should encode"); + set_tonic_canonical_body_digest(&mut divergent, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut divergent); + let mismatch = service + .heal_control(divergent) + .await + .expect_err("divergent topology must fail closed"); + assert_eq!(mismatch.code(), tonic::Code::FailedPrecondition); + + let mut command = heal_control_request(b"start"); + let body = rustfs_protos::canonical_heal_control_request_body(1, "fingerprint", b"start").expect("command should encode"); + set_tonic_canonical_body_digest(&mut command, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut command); + let disabled = service + .heal_control(command) + .await + .expect_err("commands must remain disabled"); + assert_eq!(disabled.code(), tonic::Code::Unimplemented); + } + + #[tokio::test] + async fn server_owned_heal_topology_initialization_only_publishes_valid_layouts() { + let topology = heal_control_test_endpoints("node-d"); + let expected = heal_topology_fingerprint(&topology).expect("test topology should hash"); + let cache = Arc::new(tokio::sync::OnceCell::new()); + initialize_heal_topology_fingerprint(Arc::clone(&cache), topology) + .await + .expect("valid topology should initialize"); + assert_eq!(cache.get(), Some(&expected)); + + let mut invalid = heal_control_test_endpoints("node-d"); + invalid.as_mut()[0].endpoints.as_mut()[0].pool_idx = -1; + let invalid_cache = Arc::new(tokio::sync::OnceCell::new()); + initialize_heal_topology_fingerprint(Arc::clone(&invalid_cache), invalid) + .await + .expect_err("invalid topology must fail closed"); + assert!(invalid_cache.get().is_none()); + } + #[tokio::test] async fn test_ping_success() { let service = create_test_node_service(); diff --git a/rustfs/src/storage/rpc/node_service/heal.rs b/rustfs/src/storage/rpc/node_service/heal.rs index 23f4b8cf2..783ce8823 100644 --- a/rustfs/src/storage/rpc/node_service/heal.rs +++ b/rustfs/src/storage/rpc/node_service/heal.rs @@ -13,17 +13,118 @@ // limitations under the License. use crate::startup_background::{heal_enabled_from_env, scanner_enabled_from_env}; +use crate::storage::storage_api::runtime_sources_consumer::EndpointServerPools; use rmp_serde::Deserializer; use rustfs_common::heal_channel::HealScanMode; use rustfs_heal::HealOperationsSnapshot; use rustfs_scanner::scanner::BackgroundHealInfo; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use std::io::Cursor; use super::super::encode_msgpack_map; const NODE_HEAL_STATUS_VERSION: u8 = 1; const NODE_HEAL_STATUS_MAX_SIZE: usize = 64 * 1024; +const HEAL_TOPOLOGY_FINGERPRINT_DOMAIN: &[u8] = b"rustfs-heal-topology-v1\0"; + +fn hash_sized(hasher: &mut Sha256, value: &[u8]) -> Result<(), String> { + let len = u64::try_from(value.len()).map_err(|_| "heal topology field length cannot be represented".to_string())?; + hasher.update(len.to_be_bytes()); + hasher.update(value); + Ok(()) +} + +/// Returns a node-independent digest of the complete pool/set/disk layout. +/// Locality is deliberately excluded because it differs on every peer. +pub(crate) fn heal_topology_fingerprint(endpoint_pools: &EndpointServerPools) -> Result { + if endpoint_pools.as_ref().is_empty() { + return Err("heal topology has no storage pools".to_string()); + } + let mut hasher = Sha256::new(); + hasher.update(HEAL_TOPOLOGY_FINGERPRINT_DOMAIN); + hasher.update( + u64::try_from(endpoint_pools.as_ref().len()) + .map_err(|_| "heal topology pool count cannot be represented".to_string())? + .to_be_bytes(), + ); + + for (pool_index, pool) in endpoint_pools.as_ref().iter().enumerate() { + if pool.set_count == 0 || pool.drives_per_set == 0 { + return Err(format!("heal topology pool {pool_index} has an empty set or drive layout")); + } + let expected_endpoints = pool + .set_count + .checked_mul(pool.drives_per_set) + .ok_or_else(|| "heal topology endpoint count overflow".to_string())?; + if pool.endpoints.as_ref().len() != expected_endpoints { + return Err(format!( + "heal topology pool {pool_index} has {} endpoints, expected {expected_endpoints}", + pool.endpoints.as_ref().len() + )); + } + + hasher.update( + u64::try_from(pool_index) + .map_err(|_| "heal topology pool index cannot be represented")? + .to_be_bytes(), + ); + hasher.update( + u64::try_from(pool.set_count) + .map_err(|_| "heal topology set count cannot be represented")? + .to_be_bytes(), + ); + hasher.update( + u64::try_from(pool.drives_per_set) + .map_err(|_| "heal topology drive count cannot be represented")? + .to_be_bytes(), + ); + hasher.update([u8::from(pool.legacy)]); + + let mut entries = Vec::with_capacity(pool.endpoints.as_ref().len()); + for (endpoint_index, endpoint) in pool.endpoints.as_ref().iter().enumerate() { + let declared_pool = usize::try_from(endpoint.pool_idx) + .map_err(|_| format!("heal topology endpoint {endpoint_index} has an invalid pool index"))?; + if declared_pool != pool_index { + return Err(format!("heal topology endpoint declares pool {declared_pool}, expected {pool_index}")); + } + let set_index = usize::try_from(endpoint.set_idx) + .map_err(|_| format!("heal topology endpoint {endpoint_index} has an invalid set index"))?; + let disk_index = usize::try_from(endpoint.disk_idx) + .map_err(|_| format!("heal topology endpoint {endpoint_index} has an invalid disk index"))?; + if set_index >= pool.set_count || disk_index >= pool.drives_per_set { + return Err(format!( + "heal topology endpoint position {pool_index}/{set_index}/{disk_index} is out of range" + )); + } + entries.push((set_index, disk_index, endpoint.url.as_str())); + } + entries.sort_by(|left, right| left.0.cmp(&right.0).then(left.1.cmp(&right.1))); + for pair in entries.windows(2) { + if pair[0].0 == pair[1].0 && pair[0].1 == pair[1].1 { + return Err(format!( + "heal topology contains duplicate endpoint position {pool_index}/{}/{}", + pair[0].0, pair[0].1 + )); + } + } + for (set_index, disk_index, endpoint) in entries { + hasher.update( + u64::try_from(set_index) + .map_err(|_| "heal topology set index cannot be represented")? + .to_be_bytes(), + ); + hasher.update( + u64::try_from(disk_index) + .map_err(|_| "heal topology disk index cannot be represented")? + .to_be_bytes(), + ); + hash_sized(&mut hasher, endpoint.as_bytes())?; + } + } + + Ok(hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower)) +} #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] @@ -145,11 +246,120 @@ pub(crate) fn decode_node_heal_status(data: &[u8]) -> Result EndpointServerPools { + let endpoints = ["node-a", "node-b", "node-c", last_host] + .into_iter() + .enumerate() + .map(|(index, host)| { + let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/disk{}", index + 1).as_str()) + .expect("test endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(index / 2); + endpoint.set_disk_index(index % 2); + endpoint.is_local = index == 0; + endpoint + }) + .collect::>(); + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: String::new(), + platform: String::new(), + }]) + } + + #[test] + fn heal_topology_fingerprint_is_node_independent_and_layout_complete() { + let topology = topology_endpoints("node-d"); + let expected = heal_topology_fingerprint(&topology).expect("valid topology should hash"); + + let mut other_node_view = topology; + for endpoint in other_node_view.as_mut()[0].endpoints.as_mut() { + endpoint.is_local = endpoint.host_port() == "node-c:9000"; + } + assert_eq!( + heal_topology_fingerprint(&other_node_view).expect("locality must not affect the hash"), + expected + ); + + let mut reordered = topology_endpoints("node-d"); + reordered.as_mut()[0].endpoints.as_mut().reverse(); + assert_eq!( + heal_topology_fingerprint(&reordered).expect("explicit positions must normalize endpoint order"), + expected + ); + + let mut legacy = topology_endpoints("node-d"); + legacy.as_mut()[0].legacy = true; + assert_ne!(heal_topology_fingerprint(&legacy).expect("legacy topology should hash"), expected); + + let changed = topology_endpoints("node-e"); + assert_ne!(heal_topology_fingerprint(&changed).expect("changed topology should still hash"), expected); + } + + #[test] + fn heal_topology_fingerprint_rejects_duplicate_positions() { + let mut topology = topology_endpoints("node-d"); + topology.as_mut()[0].endpoints.as_mut()[1].set_set_index(0); + topology.as_mut()[0].endpoints.as_mut()[1].set_disk_index(0); + let err = heal_topology_fingerprint(&topology).expect_err("duplicate position must fail closed"); + assert!(err.contains("duplicate endpoint position")); + } + + #[test] + fn heal_topology_fingerprint_rejects_uninitialized_indices_and_empty_layouts() { + let mut uninitialized = topology_endpoints("node-d"); + uninitialized.as_mut()[0].endpoints.as_mut()[0].pool_idx = -1; + let err = heal_topology_fingerprint(&uninitialized).expect_err("negative index must fail closed"); + assert!(err.contains("invalid pool index")); + + let empty = EndpointServerPools::default(); + let err = heal_topology_fingerprint(&empty).expect_err("empty topology must fail closed"); + assert!(err.contains("no storage pools")); + + let mut cases = Vec::new(); + let mut zero_sets = topology_endpoints("node-d"); + zero_sets.as_mut()[0].set_count = 0; + cases.push(("zero sets", zero_sets, "empty set or drive layout")); + let mut zero_drives = topology_endpoints("node-d"); + zero_drives.as_mut()[0].drives_per_set = 0; + cases.push(("zero drives", zero_drives, "empty set or drive layout")); + let mut missing_endpoint = topology_endpoints("node-d"); + missing_endpoint.as_mut()[0].endpoints.as_mut().pop(); + cases.push(("missing endpoint", missing_endpoint, "has 3 endpoints, expected 4")); + let mut wrong_pool = topology_endpoints("node-d"); + wrong_pool.as_mut()[0].endpoints.as_mut()[0].set_pool_index(1); + cases.push(("wrong pool", wrong_pool, "declares pool 1, expected 0")); + let mut negative_set = topology_endpoints("node-d"); + negative_set.as_mut()[0].endpoints.as_mut()[0].set_idx = -1; + cases.push(("negative set", negative_set, "invalid set index")); + let mut negative_disk = topology_endpoints("node-d"); + negative_disk.as_mut()[0].endpoints.as_mut()[0].disk_idx = -1; + cases.push(("negative disk", negative_disk, "invalid disk index")); + let mut out_of_range_set = topology_endpoints("node-d"); + out_of_range_set.as_mut()[0].endpoints.as_mut()[0].set_set_index(2); + cases.push(("set out of range", out_of_range_set, "is out of range")); + let mut out_of_range_disk = topology_endpoints("node-d"); + out_of_range_disk.as_mut()[0].endpoints.as_mut()[0].set_disk_index(2); + cases.push(("disk out of range", out_of_range_disk, "is out of range")); + + for (name, topology, expected) in cases { + let err = heal_topology_fingerprint(&topology).expect_err(name); + assert!(err.contains(expected), "{name}: {err}"); + } + } + #[test] fn node_heal_status_round_trip_is_versioned() { let snapshot = NodeHealStatusSnapshot::for_test( diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 326c5a24b..2318b01c3 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -247,9 +247,16 @@ pub(crate) mod rpc_consumer { pub(crate) mod runtime_sources_consumer { pub(crate) type ECStore = super::ECStore; + pub(crate) type EndpointServerPools = super::EndpointServerPools; pub(crate) use crate::storage::runtime_sources; } +pub(crate) mod heal_control_startup_consumer { + #[cfg(test)] + pub(crate) use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; + pub(crate) use crate::storage::rpc::node_service::initialize_heal_topology_fingerprint; +} + pub(crate) mod s3_api_consumer { pub(crate) mod bucket { pub(crate) use super::super::super::s3_api::bucket::{ @@ -319,7 +326,9 @@ pub(crate) mod timeout_wrapper_consumer { } pub(crate) mod tonic_service_consumer { - pub(crate) use super::super::tonic_service::{make_heal_control_server, make_server}; + #[cfg(test)] + pub(crate) use super::super::tonic_service::{heal_topology_fingerprint, make_heal_control_server_for_source}; + pub(crate) use super::super::tonic_service::{make_heal_control_server_with_cache, make_server}; } #[cfg(test)] @@ -464,11 +473,13 @@ pub(crate) mod ecstore_rio { pub(crate) mod ecstore_rpc { pub(crate) use rustfs_ecstore::api::rpc::{ LocalPeerS3Client, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG, - SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, verify_rpc_signature, - verify_tonic_canonical_body_digest, verify_tonic_rpc_signature, + SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, sign_tonic_rpc_response_proof, + verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_rpc_signature, }; #[cfg(test)] - pub(crate) use rustfs_ecstore::api::rpc::{gen_tonic_signature_headers, set_tonic_canonical_body_digest}; + pub(crate) use rustfs_ecstore::api::rpc::{ + gen_tonic_signature_headers, set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof, + }; } pub(crate) mod ecstore_object { @@ -532,6 +543,9 @@ pub(crate) fn try_current_local_node_name() -> Option { #[cfg(test)] pub(crate) use ecstore_rpc::gen_tonic_signature_headers; +pub(crate) use ecstore_rpc::sign_tonic_rpc_response_proof; +#[cfg(test)] +pub(crate) use ecstore_rpc::verify_tonic_rpc_response_proof; #[cfg(test)] pub(crate) const STORAGE_CLASS_SUB_SYS: &str = ecstore_config::com::STORAGE_CLASS_SUB_SYS; diff --git a/rustfs/src/storage/tonic_service.rs b/rustfs/src/storage/tonic_service.rs index 66f130178..3d2921b3b 100644 --- a/rustfs/src/storage/tonic_service.rs +++ b/rustfs/src/storage/tonic_service.rs @@ -12,6 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub(crate) use crate::storage::rpc::node_service::make_heal_control_server_with_cache; +#[cfg(test)] +pub(crate) use crate::storage::rpc::node_service::{heal::heal_topology_fingerprint, make_heal_control_server_for_source}; pub use crate::storage::rpc::{make_heal_control_server, make_server}; #[allow(dead_code)] pub type NodeService = crate::storage::rpc::NodeService; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 85dca1b29..6a5361362 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -102,7 +102,9 @@ pub(crate) mod server { } #[cfg(test)] - pub(crate) use crate::storage::storage_api::{PeerRestClient, gen_tonic_signature_headers}; + pub(crate) use crate::storage::storage_api::{ + Endpoint, EndpointServerPools, Endpoints, PeerRestClient, PoolEndpoints, gen_tonic_signature_headers, + }; pub(crate) mod ecfs { pub(crate) type FS = crate::storage::storage_api::FS; @@ -128,7 +130,13 @@ pub(crate) mod server { } pub(crate) mod tonic_service { - pub(crate) use crate::storage::storage_api::tonic_service_consumer::{make_heal_control_server, make_server}; + #[cfg(test)] + pub(crate) use crate::storage::storage_api::tonic_service_consumer::{ + heal_topology_fingerprint, make_heal_control_server_for_source, + }; + pub(crate) use crate::storage::storage_api::tonic_service_consumer::{ + make_heal_control_server_with_cache, make_server, + }; } } @@ -165,6 +173,12 @@ pub(crate) mod server { } pub(crate) mod startup { + pub(crate) mod heal_control { + #[cfg(test)] + pub(crate) use crate::storage::storage_api::heal_control_startup_consumer::heal_topology_fingerprint; + pub(crate) use crate::storage::storage_api::heal_control_startup_consumer::initialize_heal_topology_fingerprint; + } + pub(crate) mod background { pub(crate) use crate::storage::storage_api::{ECStore, set_workload_admission_snapshot_provider}; }