From 4e34f97dd7354cbc32de459be67fd3db97ca8057 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 1 Aug 2026 13:41:02 +0800 Subject: [PATCH] feat(kms): converge KMS configuration across nodes after a runtime change (#5551) --- crates/ecstore/src/api/mod.rs | 18 +- crates/ecstore/src/cluster/rpc/mod.rs | 2 +- .../src/cluster/rpc/peer_rest_client.rs | 50 ++- .../ecstore/src/services/notification_sys.rs | 50 +++ .../src/generated/proto_gen/node_service.rs | 6 + crates/protos/src/node.proto | 5 + rustfs/src/admin/handlers/kms_dynamic.rs | 296 +++++++++++++++++- rustfs/src/admin/handlers/kms_management.rs | 158 +++++++++- rustfs/src/storage/rpc/node_service.rs | 69 +++- rustfs/src/storage/storage_api.rs | 21 +- scripts/layer-dependency-baseline.txt | 2 + 11 files changed, 642 insertions(+), 35 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index cfba93f88..d2ddae305 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -419,15 +419,15 @@ pub mod rio { pub mod rpc { pub use crate::cluster::rpc::{ - AuthenticatedChannel, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, - PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerBucketListing, - ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, gen_signature_headers, gen_tonic_replay_scope_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, - sign_ns_scanner_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, - verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest, - verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, - verify_tonic_rpc_signature_with_bootstrap, + AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, + PeerRestClient, PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, + ScannerBucketListing, ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, gen_signature_headers, + gen_tonic_replay_scope_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, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, + tonic_boot_epoch_response_headers, verify_rpc_signature, verify_tonic_boot_epoch_response, + verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, + verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap, }; } diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index 32e2404e9..618083eb1 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -44,7 +44,7 @@ pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport; pub use internode_data_transport::build_internode_data_transport_from_env; pub(crate) use peer_rest_client::TierConfigReloadOutcome; pub use peer_rest_client::{ - PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG, + KMS_SIGNAL_SUBSYSTEM, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity, }; pub(crate) use peer_s3_client::heal_bucket_local_on_disks; diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index dc4f0bef6..1eaa62027 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -45,9 +45,9 @@ use rustfs_protos::proto_gen::node_service::{ HealControlRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ScannerActivityRequest, - ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, StartDecommissionRequest, StartProfilingRequest, - StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse, - TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient, + ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest, + StartProfilingRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest, + TierMutationControlResponse, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient, tier_mutation_control_service_client::TierMutationControlServiceClient, }; pub use rustfs_protos::{PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS}; @@ -71,6 +71,12 @@ use uuid::Uuid; pub const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = 1; pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2; +/// Dynamic config subsystem for the cluster-persisted KMS configuration. +/// +/// KMS configuration lives in its own cluster object rather than in the server +/// config document, so it is not a `ServerConfig` subsystem; it only shares the +/// reload signal transport. +pub const KMS_SIGNAL_SUBSYSTEM: &str = "kms"; const BACKGROUND_HEAL_STATUS_MAX_MESSAGE_SIZE: usize = 64 * 1024; const HEAL_CONTROL_FINGERPRINT_MAX_SIZE: usize = 256; const HEAL_CONTROL_PAYLOAD_MAX_SIZE: usize = 64 * 1024; @@ -99,8 +105,13 @@ fn decode_bucket_stats_response(response: GetBucketStatsDataResponse) -> Result< } fn validate_signal_service_protocol(sig: u64, sub_sys: &str, protocol_version: u32) -> Result<()> { + // The version stays pinned to DYNAMIC_CONFIG_PROTOCOL_VERSION rather than + // being bumped per subsystem: the comparison is shared, so raising it would + // retire peers that already converge scanner and heal config correctly. + // Subsystems added after a peer was built are rejected by that peer's own + // subsystem allow-list, which surfaces as an explicit failed signal. if sig == SERVICE_SIGNAL_RELOAD_DYNAMIC - && matches!(sub_sys, SCANNER_SUB_SYS | HEAL_SUB_SYS) + && matches!(sub_sys, SCANNER_SUB_SYS | HEAL_SUB_SYS | KMS_SIGNAL_SUBSYSTEM) && protocol_version < rustfs_protos::DYNAMIC_CONFIG_PROTOCOL_VERSION { return Err(Error::other(format!("peer does not support dynamic {sub_sys} config convergence"))); @@ -1500,6 +1511,22 @@ impl PeerRestClient { } pub async fn signal_service(&self, sig: u64, sub_sys: &str, dry_run: bool, _exec_at: SystemTime) -> Result<()> { + self.signal_service_checked(sig, sub_sys, dry_run).await.map(|_| ()) + } + + /// Report the KMS configuration fingerprint the peer is currently running. + /// + /// Sent as a dry-run reload signal so the peer answers without swapping its + /// own configuration. `None` means the peer has no KMS configuration. The + /// fingerprint is advisory and feeds cluster status reporting only, so the + /// response is not proof-signed. + pub async fn kms_config_fingerprint(&self) -> Result> { + self.signal_service_checked(SERVICE_SIGNAL_RELOAD_DYNAMIC, KMS_SIGNAL_SUBSYSTEM, true) + .await + .map(|response| response.config_fingerprint) + } + + async fn signal_service_checked(&self, sig: u64, sub_sys: &str, dry_run: bool) -> Result { self.finalize_result( async { let mut client = self.get_client().await?; @@ -1520,7 +1547,7 @@ impl PeerRestClient { return Err(Error::other("")); } validate_signal_service_protocol(sig, sub_sys, response.protocol_version)?; - Ok(()) + Ok(response) } .await, ) @@ -2347,6 +2374,19 @@ mod tests { .expect("full refresh compatibility is guarded by its scanner preflight"); } + #[test] + fn dynamic_kms_config_requires_versioned_peer_acknowledgement() { + let err = validate_signal_service_protocol(SERVICE_SIGNAL_RELOAD_DYNAMIC, KMS_SIGNAL_SUBSYSTEM, 0) + .expect_err("an unversioned peer must not claim KMS config convergence"); + assert!(err.to_string().contains("does not support dynamic")); + validate_signal_service_protocol( + SERVICE_SIGNAL_RELOAD_DYNAMIC, + KMS_SIGNAL_SUBSYSTEM, + rustfs_protos::DYNAMIC_CONFIG_PROTOCOL_VERSION, + ) + .expect("a current peer should support dynamic KMS config"); + } + #[test] fn peer_rest_client_marks_network_like_errors() { assert!(PeerRestClient::is_network_like_error(&Error::other("transport error"))); diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index c690707ad..b08211853 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -337,6 +337,15 @@ pub struct NotificationPeerErr { pub err: Option, } +/// One peer's answer to a KMS configuration fingerprint probe. +pub struct PeerKmsConfigFingerprint { + pub host: String, + /// `None` when the peer has no KMS configuration of its own, or could not + /// be asked at all, in which case `err` carries the reason. + pub fingerprint: Option, + pub err: Option, +} + fn notification_peer_result(host: String, result: Result) -> NotificationPeerErr { NotificationPeerErr { host, err: result.err() } } @@ -518,6 +527,47 @@ impl NotificationSys { self.signal_dynamic_config(sub_sys, false).await } + /// Ask every peer to re-read the cluster-persisted KMS configuration. + /// + /// Best-effort by contract: the caller has already switched locally, so a + /// peer that fails is reported rather than rolled back. Peers built before + /// the KMS subsystem existed reject the signal with an explicit error. + pub async fn reload_kms_config(&self) -> Vec { + self.reload_dynamic_config(crate::cluster::rpc::KMS_SIGNAL_SUBSYSTEM).await + } + + /// Collect the KMS configuration fingerprint each peer is running. + /// + /// A peer whose build predates the KMS subsystem rejects the probe, so it + /// is reported as an error rather than silently agreeing with this node. + pub async fn kms_config_fingerprints(&self) -> Vec { + let mut futures = Vec::with_capacity(self.peer_clients.len()); + for client in self.peer_clients.iter() { + futures.push(async move { + let Some(client) = client else { + return PeerKmsConfigFingerprint { + host: String::new(), + fingerprint: None, + err: Some(Error::other("peer is not reachable")), + }; + }; + match client.kms_config_fingerprint().await { + Ok(fingerprint) => PeerKmsConfigFingerprint { + host: client.host.to_string(), + fingerprint, + err: None, + }, + Err(e) => PeerKmsConfigFingerprint { + host: client.host.to_string(), + fingerprint: None, + err: Some(e), + }, + } + }); + } + join_all(futures).await + } + pub async fn refresh_config_snapshot(&self) -> Vec { let mut futures = Vec::with_capacity(self.peer_clients.len()); for client in self.peer_clients.iter() { diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 35c320d7a..72d7326e0 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -1171,6 +1171,12 @@ pub struct SignalServiceResponse { pub error_info: ::core::option::Option<::prost::alloc::string::String>, #[prost(uint32, tag = "3")] pub protocol_version: u32, + /// Redacted fingerprint of the KMS configuration this node is running. + /// Only set for the "kms" dynamic config subsystem, and left unset when the + /// responder has no KMS configuration. Peers built before that subsystem + /// reject the signal outright rather than answering without this field. + #[prost(string, optional, tag = "4")] + pub config_fingerprint: ::core::option::Option<::prost::alloc::string::String>, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ScannerActivityRequest { diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index fd347ea04..3f5ae7f22 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -817,6 +817,11 @@ message SignalServiceResponse { bool success = 1; optional string error_info = 2; uint32 protocol_version = 3; + // Redacted fingerprint of the KMS configuration this node is running. + // Only set for the "kms" dynamic config subsystem, and left unset when the + // responder has no KMS configuration. Peers built before that subsystem + // reject the signal outright rather than answering without this field. + optional string config_fingerprint = 4; } message ScannerActivityRequest { diff --git a/rustfs/src/admin/handlers/kms_dynamic.rs b/rustfs/src/admin/handlers/kms_dynamic.rs index 6a6da70d4..526079d28 100644 --- a/rustfs/src/admin/handlers/kms_dynamic.rs +++ b/rustfs/src/admin/handlers/kms_dynamic.rs @@ -17,8 +17,8 @@ use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::{ - current_app_context, current_kms_runtime_service_manager, current_object_store_handle_for_context, - current_or_init_kms_runtime_service_manager, + current_app_context, current_kms_runtime_service_manager, current_notification_system_for_context, + current_object_store_handle_for_context, current_or_init_kms_runtime_service_manager, }; use crate::admin::storage_api::config::{read_admin_config, save_admin_config}; use crate::auth::{check_key_valid, get_session_token}; @@ -32,6 +32,7 @@ use rustfs_kms::{ }; use rustfs_policy::policy::action::{Action, KmsAction}; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; +use sha2::{Digest, Sha256}; use tracing::{error, info, instrument, warn}; /// Path to store KMS configuration in the cluster metadata @@ -41,6 +42,13 @@ const STATIC_KMS_LOCAL_CONFIG_REQUIRED: &str = const LOG_COMPONENT_ADMIN: &str = "admin"; const LOG_SUBSYSTEM_KMS: &str = "kms"; const EVENT_ADMIN_KMS_DYNAMIC_STATE: &str = "admin_kms_dynamic_state"; +/// Substituted for every secret value before a configuration is fingerprinted, +/// so the digest depends on the shape of the configuration and never on key +/// material an operator could recover by replaying candidate secrets. +const REDACTED_CONFIG_VALUE: &str = "[redacted]"; +/// Configuration field names carrying secrets. Matched at any depth so a +/// backend that nests its credentials cannot leak them into the fingerprint. +const REDACTED_CONFIG_FIELDS: [&str; 6] = ["token", "secret_key", "secret_id", "master_key", "password", "client_secret"]; fn kms_service_manager_from_context() -> std::sync::Arc { current_kms_runtime_service_manager().unwrap_or_else(|| { @@ -284,6 +292,198 @@ pub async fn load_kms_config() -> Option { } } +fn redact_config_secrets(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(object) => { + for (key, field) in object.iter_mut() { + if REDACTED_CONFIG_FIELDS.contains(&key.as_str()) { + *field = serde_json::Value::String(REDACTED_CONFIG_VALUE.to_string()); + } else { + redact_config_secrets(field); + } + } + } + serde_json::Value::Array(items) => items.iter_mut().for_each(redact_config_secrets), + _ => {} + } +} + +/// Serialize with object keys ordered so two nodes holding the same +/// configuration always hash the same bytes, whatever their map iteration order. +fn write_canonical_json(value: &serde_json::Value, out: &mut String) { + match value { + serde_json::Value::Object(object) => { + let mut keys = object.keys().collect::>(); + keys.sort_unstable(); + out.push('{'); + for (index, key) in keys.into_iter().enumerate() { + if index > 0 { + out.push(','); + } + out.push_str(&serde_json::Value::String(key.clone()).to_string()); + out.push(':'); + write_canonical_json(&object[key], out); + } + out.push('}'); + } + serde_json::Value::Array(items) => { + out.push('['); + for (index, item) in items.iter().enumerate() { + if index > 0 { + out.push(','); + } + write_canonical_json(item, out); + } + out.push(']'); + } + other => out.push_str(&other.to_string()), + } +} + +fn redacted_canonical_config(config: &KmsConfig) -> Option { + let mut value = serde_json::to_value(config).ok()?; + redact_config_secrets(&mut value); + let mut canonical = String::new(); + write_canonical_json(&value, &mut canonical); + Some(canonical) +} + +/// Fingerprint of a KMS configuration in its persisted form, with every secret +/// replaced before hashing. +/// +/// Comparing fingerprints across nodes makes a configuration split visible +/// without exposing what any node is configured with. Redaction is deliberate: +/// two nodes holding the same backend with different credentials fingerprint +/// alike, which is the price of a digest that is safe to report. +pub fn kms_config_fingerprint(config: &KmsConfig) -> Option { + let canonical = redacted_canonical_config(config)?; + Some(hex_simd::encode_to_string( + Sha256::digest(canonical.as_bytes()), + hex_simd::AsciiCase::Lower, + )) +} + +/// Fingerprint of the configuration this node is currently running. +/// +/// `None` when KMS has never been configured on this node, which is itself a +/// divergence from a cluster that has. +pub async fn current_kms_config_fingerprint() -> Option { + kms_config_fingerprint(&kms_service_manager_from_context().get_config().await?) +} + +fn kms_config_is_unchanged(current: &KmsConfig, candidate: &KmsConfig) -> bool { + match (serde_json::to_vec(current), serde_json::to_vec(candidate)) { + (Ok(current), Ok(candidate)) => current == candidate, + _ => false, + } +} + +/// Re-read the cluster-persisted KMS configuration and switch this node to it. +/// +/// Invoked on peers by the reload signal that a configure or reconfigure +/// request broadcasts, so that a runtime change reaches every node instead of +/// only the one that served the admin request. +pub async fn reload_persisted_kms_config() -> Result<(), String> { + let Some(config) = load_kms_config().await else { + return Err("no persisted KMS configuration is available".to_string()); + }; + + let service_manager = kms_service_manager_from_context(); + if service_manager + .get_config() + .await + .is_some_and(|current| kms_config_is_unchanged(¤t, &config)) + { + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "peer_reload", + state = "already_current", + "admin kms dynamic state" + ); + return Ok(()); + } + + service_manager.reconfigure(config).await.map_err(|err| { + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "peer_reload", + state = "reload_failed", + error = %err, + "admin kms dynamic state" + ); + format!("failed to apply persisted KMS configuration: {err}") + })?; + + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "peer_reload", + state = "reconfigured", + "admin kms dynamic state" + ); + Ok(()) +} + +/// Ask every peer to adopt the configuration that was just persisted. +/// +/// Returns the hosts that did not converge. Reporting is best effort by +/// design: this node has already switched and KMS configuration has no quorum +/// or authoritative holder to roll back to, so refusing the local change would +/// trade a bounded divergence window for an outage. +async fn broadcast_kms_config_reload() -> Vec { + let context = current_app_context(); + let Some(notification_sys) = current_notification_system_for_context(context.as_deref()) else { + return Vec::new(); + }; + + let mut unconverged = Vec::new(); + for outcome in notification_sys.reload_kms_config().await { + let Some(err) = outcome.err else { + continue; + }; + let host = if outcome.host.is_empty() { + "".to_string() + } else { + outcome.host + }; + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_peer_config_reload_failed", + peer = %host, + result = "peer_reload_failed", + error = %err, + "admin kms dynamic state" + ); + unconverged.push(host); + } + unconverged +} + +/// Fold a best-effort peer reload result into the local outcome. +/// +/// The local outcome stays successful whatever the peers reported: this node +/// has already switched, and a failed peer is named in the message so the +/// operator can act on the divergence instead of being told nothing happened. +fn local_success_with_peer_report(message: &str, unconverged: &[String]) -> (bool, String) { + if unconverged.is_empty() { + return (true, message.to_string()); + } + ( + true, + format!( + "{message}. {} peer(s) did not reload the new configuration and keep serving their previous KMS configuration until they do: {}", + unconverged.len(), + unconverged.join(", ") + ), + ) +} + pub fn register_kms_dynamic_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( Method::POST, @@ -413,7 +613,9 @@ impl Operation for ConfigureKmsHandler { status = ?status, "admin kms dynamic state" ); - (true, "KMS configured successfully".to_string(), status) + let unconverged = broadcast_kms_config_reload().await; + let (success, message) = local_success_with_peer_report("KMS configured successfully", &unconverged); + (success, message, status) } Err(e) => { let error_msg = format!("Failed to configure KMS: {e}"); @@ -884,7 +1086,10 @@ impl Operation for ReconfigureKmsHandler { status = ?status, "admin kms dynamic state" ); - (true, "KMS reconfigured and restarted successfully".to_string(), status) + let unconverged = broadcast_kms_config_reload().await; + let (success, message) = + local_success_with_peer_report("KMS reconfigured and restarted successfully", &unconverged); + (success, message, status) } Err(e) => { let error_msg = format!("Failed to reconfigure KMS: {e}"); @@ -934,8 +1139,9 @@ impl Operation for ReconfigureKmsHandler { #[cfg(test)] mod tests { use super::{ - decode_persisted_kms_config, ensure_kms_config_persistable, kms_configure_actions, kms_service_control_actions, - normalize_configure_request_secrets, + decode_persisted_kms_config, ensure_kms_config_persistable, kms_config_fingerprint, kms_config_is_unchanged, + kms_configure_actions, kms_service_control_actions, local_success_with_peer_report, normalize_configure_request_secrets, + redacted_canonical_config, }; use rustfs_policy::policy::action::{Action, AdminAction, KmsAction}; use std::path::PathBuf; @@ -1171,4 +1377,82 @@ mod tests { .expect_err("changing from the local backend must be rejected"); assert_eq!(backend_error, "Changing from the Local KMS backend is not supported"); } + + const VAULT_TOKEN: &str = "hvs-super-secret-token"; + + fn vault_config(address: &str, token: &str) -> rustfs_kms::KmsConfig { + let vault = rustfs_kms::config::VaultConfig { + address: address.to_string(), + auth_method: rustfs_kms::config::VaultAuthMethod::Token { + token: token.to_string(), + }, + ..Default::default() + }; + rustfs_kms::KmsConfig { + backend: rustfs_kms::KmsBackend::VaultKv2, + backend_config: rustfs_kms::BackendConfig::VaultKv2(Box::new(vault)), + ..Default::default() + } + } + + #[test] + fn config_fingerprint_input_carries_no_credentials() { + let canonical = redacted_canonical_config(&vault_config("https://vault.internal:8200", VAULT_TOKEN)) + .expect("vault configuration should serialize"); + assert!(!canonical.contains(VAULT_TOKEN), "fingerprint input must not carry the vault token"); + assert!(canonical.contains(super::REDACTED_CONFIG_VALUE)); + assert!( + canonical.contains("vault.internal"), + "non-secret configuration must still drive the fingerprint" + ); + } + + #[test] + fn config_fingerprint_agrees_across_nodes_holding_the_same_configuration() { + let first = kms_config_fingerprint(&vault_config("https://vault.internal:8200", VAULT_TOKEN)) + .expect("fingerprint should be computable"); + let second = kms_config_fingerprint(&vault_config("https://vault.internal:8200", VAULT_TOKEN)) + .expect("fingerprint should be computable"); + assert_eq!(first, second); + + let other_backend = kms_config_fingerprint(&vault_config("https://vault.other:8200", VAULT_TOKEN)) + .expect("fingerprint should be computable"); + assert_ne!(first, other_backend, "a different backend address must be visible as a split"); + + // Credentials are redacted before hashing, so a rotated token is not a + // split. Documented here so the trade-off is not silently regressed. + let rotated_token = kms_config_fingerprint(&vault_config("https://vault.internal:8200", "hvs-rotated-token")) + .expect("fingerprint should be computable"); + assert_eq!(first, rotated_token); + } + + #[test] + fn peer_reload_replays_only_when_the_persisted_configuration_moved() { + let current = vault_config("https://vault.internal:8200", VAULT_TOKEN); + assert!(kms_config_is_unchanged( + ¤t, + &vault_config("https://vault.internal:8200", VAULT_TOKEN) + )); + assert!(!kms_config_is_unchanged(¤t, &vault_config("https://vault.other:8200", VAULT_TOKEN))); + assert!( + !kms_config_is_unchanged(¤t, &vault_config("https://vault.internal:8200", "hvs-rotated-token")), + "a credential-only change must still be adopted, unlike the redacted fingerprint" + ); + } + + #[test] + fn peer_reload_failures_are_reported_without_failing_the_local_change() { + let (success, message) = local_success_with_peer_report("KMS configured successfully", &[]); + assert!(success); + assert_eq!(message, "KMS configured successfully"); + + let (success, message) = local_success_with_peer_report( + "KMS configured successfully", + &["10.0.0.2:9000".to_string(), "10.0.0.3:9000".to_string()], + ); + assert!(success, "a peer that failed to reload must not fail the node that already switched"); + assert!(message.contains("2 peer(s)")); + assert!(message.contains("10.0.0.2:9000")); + assert!(message.contains("10.0.0.3:9000")); + } } diff --git a/rustfs/src/admin/handlers/kms_management.rs b/rustfs/src/admin/handlers/kms_management.rs index 7fa45b372..978848fb4 100644 --- a/rustfs/src/admin/handlers/kms_management.rs +++ b/rustfs/src/admin/handlers/kms_management.rs @@ -14,10 +14,13 @@ //! KMS management route registration. +use super::kms_dynamic::current_kms_config_fingerprint; use super::kms_keys::{CreateKeyHandler, DescribeKeyHandler, GenerateDataKeyHandler, ListKeysHandler}; use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; -use crate::admin::runtime_sources::{current_kms_runtime_service_manager, current_or_init_kms_runtime_service_manager}; +use crate::admin::runtime_sources::{ + current_kms_runtime_service_manager, current_notification_system, current_or_init_kms_runtime_service_manager, +}; use crate::auth::{check_key_valid, get_session_token}; use crate::server::{ADMIN_PREFIX, RemoteAddr}; use hyper::{HeaderMap, Method, StatusCode}; @@ -76,6 +79,79 @@ pub struct KmsStatusResponse { /// older servers, so it must stay optional for consumers. #[serde(default, skip_serializing_if = "Option::is_none")] pub capabilities: Option, + /// Per-node fingerprint of the running KMS configuration. Additive field: + /// omitted by older servers, so it must stay optional for consumers. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cluster_config: Option, +} + +/// Cluster-wide view of which KMS configuration each node is running. +/// +/// KMS configuration is applied per node, so a runtime change that fails to +/// reach a peer leaves that peer serving a different backend. Comparing +/// redacted fingerprints makes that divergence observable and alertable. +#[derive(Debug, Serialize, Deserialize)] +pub struct KmsClusterConfigStatus { + /// True only when every node answered with the same fingerprint. + pub consistent: bool, + pub nodes: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct KmsNodeConfigStatus { + /// Peer address, or `local` for the node serving this request. + pub host: String, + /// `None` when the node has no KMS configuration, or could not be asked at + /// all, in which case `error` carries the reason. + pub config_fingerprint: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +const LOCAL_NODE_HOST: &str = "local"; +const UNKNOWN_PEER_HOST: &str = ""; + +/// Collect the configuration fingerprint of this node and of every peer. +async fn collect_cluster_config_status() -> KmsClusterConfigStatus { + let mut nodes = vec![KmsNodeConfigStatus { + host: LOCAL_NODE_HOST.to_string(), + config_fingerprint: current_kms_config_fingerprint().await, + error: None, + }]; + + if let Some(notification_sys) = current_notification_system() { + for peer in notification_sys.kms_config_fingerprints().await { + nodes.push(KmsNodeConfigStatus { + host: if peer.host.is_empty() { + UNKNOWN_PEER_HOST.to_string() + } else { + peer.host + }, + config_fingerprint: peer.fingerprint, + error: peer.err.map(|err| err.to_string()), + }); + } + } + + let consistent = cluster_config_is_consistent(&nodes); + KmsClusterConfigStatus { consistent, nodes } +} + +/// Whether every node was observed running the same KMS configuration. +/// +/// A node that could not be asked, or that answered without a fingerprint, +/// counts as divergent: the field exists to refuse agreement that was never +/// observed, so an unreachable peer never reads as converged. +fn cluster_config_is_consistent(nodes: &[KmsNodeConfigStatus]) -> bool { + let Some(first) = nodes.first() else { + return false; + }; + if first.config_fingerprint.is_none() { + return false; + } + nodes + .iter() + .all(|node| node.error.is_none() && node.config_fingerprint == first.config_fingerprint) } #[derive(Debug, Serialize, Deserialize)] @@ -219,6 +295,7 @@ impl Operation for KmsStatusHandler { cache_stats, default_key_id: service.get_default_key_id().cloned(), capabilities: Some(service.backend_capabilities()), + cluster_config: Some(collect_cluster_config_status().await), }; let data = serde_json::to_vec(&response).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?; @@ -384,4 +461,83 @@ mod tests { assert_eq!(capabilities.get("encrypt"), Some(&serde_json::Value::Bool(true))); assert_eq!(capabilities.get("rotate"), Some(&serde_json::Value::Bool(false))); } + + fn node(host: &str, fingerprint: Option<&str>, error: Option<&str>) -> super::KmsNodeConfigStatus { + super::KmsNodeConfigStatus { + host: host.to_string(), + config_fingerprint: fingerprint.map(str::to_string), + error: error.map(str::to_string), + } + } + + #[test] + fn cluster_config_is_consistent_only_when_every_node_reports_the_same_fingerprint() { + assert!(super::cluster_config_is_consistent(&[ + node("local", Some("abc"), None), + node("peer-1", Some("abc"), None), + ])); + assert!(!super::cluster_config_is_consistent(&[ + node("local", Some("abc"), None), + node("peer-1", Some("def"), None), + ])); + } + + #[test] + fn cluster_config_consistency_never_claims_agreement_it_did_not_observe() { + // An unreachable peer, a peer whose build does not report a + // fingerprint, and a node without any configuration all have to read as + // divergent rather than silently agreeing. + assert!(!super::cluster_config_is_consistent(&[ + node("local", Some("abc"), None), + node("peer-1", None, Some("peer is not reachable")), + ])); + assert!(!super::cluster_config_is_consistent(&[ + node("local", Some("abc"), None), + node("peer-1", None, None), + ])); + assert!(!super::cluster_config_is_consistent(&[ + node("local", None, None), + node("peer-1", None, None), + ])); + assert!(!super::cluster_config_is_consistent(&[])); + } + + /// The `cluster_config` field is additive for the same reason + /// `capabilities` is: older servers omit it and consumers must keep + /// deserializing their payloads. + #[test] + fn kms_status_response_cluster_config_field_is_additive() { + let legacy_json = serde_json::json!({ + "backend_type": "local", + "backend_status": "healthy", + "cache_enabled": true, + "cache_stats": null, + "default_key_id": null, + }); + let legacy: super::KmsStatusResponse = + serde_json::from_value(legacy_json).expect("legacy status payload should deserialize"); + assert!(legacy.cluster_config.is_none()); + let serialized = serde_json::to_value(&legacy).expect("status response should serialize"); + assert!(serialized.get("cluster_config").is_none(), "unset cluster config must be omitted"); + + let with_cluster_config = super::KmsStatusResponse { + cluster_config: Some(super::KmsClusterConfigStatus { + consistent: false, + nodes: vec![node("local", Some("abc"), None), node("peer-1", None, Some("unreachable"))], + }), + ..legacy + }; + let serialized = serde_json::to_value(&with_cluster_config).expect("status response should serialize"); + let cluster_config = serialized + .get("cluster_config") + .expect("cluster config must be present when set"); + assert_eq!(cluster_config.get("consistent"), Some(&serde_json::Value::Bool(false))); + assert_eq!( + cluster_config + .get("nodes") + .and_then(serde_json::Value::as_array) + .map(Vec::len), + Some(2) + ); + } } diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 592355d11..26f600191 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::admin::handlers::kms_dynamic::{current_kms_config_fingerprint, reload_persisted_kms_config}; use crate::admin::service::{ config::{reload_dynamic_config_runtime_state, reload_runtime_config_snapshot}, site_replication::reload_site_replication_runtime_state, @@ -22,7 +23,7 @@ use crate::storage::storage_api::rpc_consumer::node_service::STORAGE_CLASS_SUB_S #[cfg(test)] use crate::storage::storage_api::rpc_consumer::node_service::{CollectMetricsOpts, MetricType}; use crate::storage::storage_api::rpc_consumer::node_service::{ - DiskStore, ECStore, Error, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, + DiskStore, ECStore, Error, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _, StorageResult, all_local_disk_path, find_local_disk_by_ref, reload_transition_tier_config, @@ -87,6 +88,29 @@ fn signal_service_response(success: bool, error_info: Option) -> Respons success, error_info, protocol_version: rustfs_protos::DYNAMIC_CONFIG_PROTOCOL_VERSION, + config_fingerprint: None, + }) +} + +/// Answer a KMS dynamic config signal. +/// +/// A dry run doubles as the cluster fingerprint probe: it reports what this +/// node is running without touching its configuration. A real signal reloads +/// the cluster-persisted configuration first, and still answers with the +/// fingerprint of whatever it ends up running so a failed reload is visible as +/// divergence rather than only as an error string. +async fn kms_dynamic_config_signal_response(dry_run: bool) -> Response { + let outcome = if dry_run { + Ok(()) + } else { + reload_persisted_kms_config().await + }; + let fingerprint = current_kms_config_fingerprint().await; + Response::new(SignalServiceResponse { + success: outcome.is_ok(), + error_info: outcome.err(), + protocol_version: rustfs_protos::DYNAMIC_CONFIG_PROTOCOL_VERSION, + config_fingerprint: fingerprint, }) } @@ -1736,6 +1760,12 @@ impl Node for NodeService { Err(_) => Ok(signal_service_response(false, Some("runtime config snapshot reload failed".to_string()))), }, Some(SERVICE_SIGNAL_RELOAD_DYNAMIC) => { + // KMS configuration is persisted outside the server config + // document, so it converges through its own reload rather than + // through the server config publication fence. + if sub_system == KMS_SIGNAL_SUBSYSTEM { + return Ok(kms_dynamic_config_signal_response(dry_run).await); + } let supported = sub_system == MODULE_SWITCHES_SIGNAL_SUBSYSTEM || supports_dynamic_config_rpc(sub_system); if !supported { return Ok(signal_service_response( @@ -2130,8 +2160,8 @@ impl Node for NodeService { #[allow(unused_imports)] mod tests { use super::{ - CollectMetricsOpts, DiskStore, Error, HEAL_CONTROL_PAYLOAD_MAX_SIZE, MetricType, Node as _, NodeService, - PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, + CollectMetricsOpts, DiskStore, Error, HEAL_CONTROL_PAYLOAD_MAX_SIZE, KMS_SIGNAL_SUBSYSTEM, MetricType, Node as _, + NodeService, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, STORAGE_CLASS_SUB_SYS, admit_heal_control_replay, background_rebalance_start_error_message, execute_heal_control_envelope_with_manager, initialize_heal_topology_fingerprint, @@ -5363,6 +5393,39 @@ mod tests { assert!(super::supports_dynamic_config_rpc(sub_system)); } assert!(!super::supports_dynamic_config_rpc("identity_openid")); + // KMS configuration is not a server config subsystem: it converges + // through its own branch, so it must stay out of this allow-list. + assert!(!super::supports_dynamic_config_rpc(KMS_SIGNAL_SUBSYSTEM)); + } + + #[tokio::test] + #[serial_test::serial] + async fn test_signal_service_kms_dry_run_reports_without_reconfiguring() { + let service = create_test_node_service(); + + let mut vars = HashMap::new(); + vars.insert(PEER_RESTSIGNAL.to_string(), SERVICE_SIGNAL_RELOAD_DYNAMIC.to_string()); + vars.insert(PEER_RESTSUB_SYS.to_string(), KMS_SIGNAL_SUBSYSTEM.to_string()); + vars.insert(PEER_RESTDRY_RUN.to_string(), true.to_string()); + + let response = service + .signal_service(Request::new(SignalServiceRequest { + vars: Some(Mss { value: vars }), + })) + .await + .expect("KMS capability probe should return a response") + .into_inner(); + + // A probe must answer even where no configuration was ever applied: + // that answer is what makes an unconfigured node visible as divergent. + assert!(response.success, "new nodes must advertise KMS config convergence support"); + assert!(response.error_info.is_none()); + assert_eq!(response.protocol_version, rustfs_protos::DYNAMIC_CONFIG_PROTOCOL_VERSION); + assert_eq!( + response.config_fingerprint, + super::current_kms_config_fingerprint().await, + "a probe must report the configuration this node is running" + ); } #[tokio::test] diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 5bfbbef7b..a50a141e9 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -231,11 +231,11 @@ pub(crate) mod rpc_consumer { }; pub(crate) use super::super::{ BatchReadVersionReq, BatchReadVersionResp, CollectMetricsOpts, DeleteOptions, DiskError, DiskInfoOptions, DiskStore, - ECStore, Error, FileInfoVersions, LocalPeerS3Client, MetricType, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, - ReadMultipleReq, ReadMultipleResp, ReadOptions, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, - StorageDiskRpcExt, StoragePeerS3ClientExt, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, - find_local_disk_by_ref, get_local_server_property, reload_bucket_metadata, reload_transition_tier_config, - remove_bucket_metadata, validate_batch_read_version_item_count, + ECStore, Error, FileInfoVersions, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MetricType, PEER_RESTDRY_RUN, + PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq, ReadMultipleResp, ReadOptions, SERVICE_SIGNAL_REFRESH_CONFIG, + SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt, StoragePeerS3ClientExt, UpdateMetadataOpts, all_local_disk_path, + collect_local_metrics, find_local_disk_by_ref, get_local_server_property, reload_bucket_metadata, + reload_transition_tier_config, remove_bucket_metadata, validate_batch_read_version_item_count, }; pub(crate) type StorageResult = super::super::Result; @@ -494,11 +494,11 @@ pub(crate) mod ecstore_rio { pub(crate) mod ecstore_rpc { pub(crate) use rustfs_ecstore::api::rpc::{ - LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, - SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, - sign_ns_scanner_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, - verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, - verify_tonic_rpc_signature_with_bootstrap, + KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, + PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, + normalize_tonic_rpc_audience, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, + tonic_boot_epoch_response_headers, verify_rpc_signature, verify_tonic_canonical_body_digest, + verify_tonic_mutation_body_digest, verify_tonic_rpc_signature_with_bootstrap, }; #[cfg(test)] pub(crate) use rustfs_ecstore::api::rpc::{ @@ -562,6 +562,7 @@ pub(crate) const OBJECT_LOCK_CONFIG: &str = ecstore_bucket::metadata::OBJECT_LOC pub(crate) const PEER_RESTDRY_RUN: &str = ecstore_rpc::PEER_RESTDRY_RUN; pub(crate) const PEER_RESTSIGNAL: &str = ecstore_rpc::PEER_RESTSIGNAL; pub(crate) const PEER_RESTSUB_SYS: &str = ecstore_rpc::PEER_RESTSUB_SYS; +pub(crate) const KMS_SIGNAL_SUBSYSTEM: &str = ecstore_rpc::KMS_SIGNAL_SUBSYSTEM; pub(crate) const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = ecstore_rpc::SERVICE_SIGNAL_REFRESH_CONFIG; pub(crate) const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = ecstore_rpc::SERVICE_SIGNAL_RELOAD_DYNAMIC; pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET; diff --git a/scripts/layer-dependency-baseline.txt b/scripts/layer-dependency-baseline.txt index 33354b8eb..29f53f27d 100644 --- a/scripts/layer-dependency-baseline.txt +++ b/scripts/layer-dependency-baseline.txt @@ -36,6 +36,8 @@ dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::is_notify_modul dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_audit_module_enabled dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_notify_module_enabled dep|rustfs/src/storage/rpc/http_service.rs|infra->interface|crate::server::RPC_PREFIX +dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::handlers::kms_dynamic::current_kms_config_fingerprint +dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::handlers::kms_dynamic::reload_persisted_kms_config dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::config::reload_dynamic_config_runtime_state dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::config::reload_runtime_config_snapshot dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::site_replication::reload_site_replication_runtime_state