From 849837e262c6ebe1d16a9715481a42f8d0d833ca Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 11 Aug 2026 09:03:25 +0800 Subject: [PATCH] fix(rpc): negotiate replay-safe mutation authentication (#5928) * fix(rpc): negotiate replay-safe mutation auth * fix(rpc): preserve strict legacy replay scope --- crates/ecstore/src/cluster/rpc/client.rs | 678 ++++++++++++++++-- crates/ecstore/src/cluster/rpc/http_auth.rs | 148 +++- crates/ecstore/src/cluster/rpc/mod.rs | 11 +- crates/ecstore/src/cluster/rpc/remote_disk.rs | 19 +- .../ecstore/src/cluster/rpc/remote_locker.rs | 47 +- docs/architecture/compat-cleanup-register.md | 2 +- 6 files changed, 838 insertions(+), 67 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/client.rs b/crates/ecstore/src/cluster/rpc/client.rs index 59082b659..0e1395fff 100644 --- a/crates/ecstore/src/cluster/rpc/client.rs +++ b/crates/ecstore/src/cluster/rpc/client.rs @@ -15,12 +15,12 @@ #[cfg(test)] use crate::cluster::rpc::http_auth::RPC_REPLAY_SCOPE_VERSION_HEADER; use crate::cluster::rpc::http_auth::{ - RPC_AUTH_VERSION_HEADER, RPC_AUTH_VERSION_V2, RPC_BOOT_EPOCH_CHALLENGE_HEADER, RPC_BOOT_EPOCH_HEADER, - RPC_BOOT_EPOCH_PROOF_HEADER, RPC_CONTENT_SHA256_HEADER, TIMESTAMP_HEADER, -}; -use crate::cluster::rpc::{ - gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience, verify_tonic_boot_epoch_response, + AuthenticatedPeerReplayCapabilities, RPC_AUTH_VERSION_HEADER, RPC_AUTH_VERSION_V2, RPC_BOOT_EPOCH_CHALLENGE_HEADER, + RPC_CONTENT_SHA256_HEADER, RPC_REPLAY_CACHE_CAPABILITY_HEADER, RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER, + RollingMutationBodyDigest, TIMESTAMP_HEADER, internode_rpc_body_digest_strict, + verify_tonic_peer_replay_capabilities_response, }; +use crate::cluster::rpc::{gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience}; #[cfg(test)] use crate::cluster::rpc::{tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers}; use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError}; @@ -233,7 +233,22 @@ pub struct ReplayScopeChannel { /// The channel type used by internode clients after v2 authentication and replay-scope handling. pub type AuthenticatedChannel = ReplayScopeChannel; -static PEER_BOOT_EPOCHS: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum PeerReplayCapability { + Capable { boot_epoch: Uuid }, + Revoked, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +struct PeerReplayState { + boot_epoch: Option, + cache_capability: Option, +} + +#[derive(Clone, Copy, Debug)] +struct PeerReplayStateSnapshot(PeerReplayState); + +static PEER_REPLAY_STATES: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); impl ReplayScopeChannel { fn new(inner: S, audience: Option) -> Self { @@ -241,13 +256,67 @@ impl ReplayScopeChannel { } } -fn cached_peer_boot_epoch(audience: &str) -> Option { - PEER_BOOT_EPOCHS.lock().ok().and_then(|epochs| epochs.get(audience).copied()) +fn peer_replay_state(audience: &str) -> PeerReplayState { + PEER_REPLAY_STATES + .lock() + .ok() + .and_then(|states| states.get(audience).copied()) + .unwrap_or_default() } -fn remember_peer_boot_epoch(audience: String, epoch: Uuid) { - if let Ok(mut epochs) = PEER_BOOT_EPOCHS.lock() { - epochs.insert(audience, epoch); +fn apply_peer_replay_response( + audience: String, + sent_state: PeerReplayState, + response: std::io::Result, +) { + if let Ok(mut states) = PEER_REPLAY_STATES.lock() { + let current_state = states.get(&audience).copied().unwrap_or_default(); + let mut next_state = current_state; + if let Ok(response) = &response + && sent_state.boot_epoch == current_state.boot_epoch + { + next_state.boot_epoch = Some(response.boot_epoch); + } + + if sent_state.boot_epoch == current_state.boot_epoch { + let response_capability = response + .as_ref() + .ok() + .filter(|response| response.dynamic_replay_cache) + .map(|response| response.boot_epoch); + match (sent_state.cache_capability, current_state.cache_capability, response_capability) { + (None, None, Some(boot_epoch)) + | (Some(PeerReplayCapability::Revoked), Some(PeerReplayCapability::Revoked), Some(boot_epoch)) => { + next_state.cache_capability = Some(PeerReplayCapability::Capable { boot_epoch }); + } + ( + Some(PeerReplayCapability::Capable { + boot_epoch: sent_boot_epoch, + }), + Some(PeerReplayCapability::Capable { + boot_epoch: current_boot_epoch, + }), + Some(response_boot_epoch), + ) if sent_boot_epoch == current_boot_epoch => { + next_state.cache_capability = Some(PeerReplayCapability::Capable { + boot_epoch: response_boot_epoch, + }); + } + ( + Some(PeerReplayCapability::Capable { + boot_epoch: sent_boot_epoch, + }), + Some(PeerReplayCapability::Capable { + boot_epoch: current_boot_epoch, + }), + None, + ) if sent_boot_epoch == current_boot_epoch => { + next_state.cache_capability = Some(PeerReplayCapability::Revoked); + } + _ => {} + } + } + states.insert(audience, next_state); } } @@ -276,6 +345,11 @@ where == Some(RPC_AUTH_VERSION_V2) }); let challenge = authenticated.then(Uuid::new_v4); + let sent_state = request + .extensions() + .get::() + .map(|snapshot| snapshot.0) + .unwrap_or_default(); if let (Some(audience), Some(challenge)) = (self.audience.as_deref(), challenge) { // The challenge is independently HMAC-authenticated by the response proof. It is not // part of v2 so old peers ignore it, while a new peer can safely advertise its epoch. @@ -284,7 +358,7 @@ where challenge.to_string().parse().expect("UUID must be a valid header value"), ); if let (Some(boot_epoch), Some(timestamp), Some(content_sha256)) = ( - cached_peer_boot_epoch(audience), + sent_state.boot_epoch, request.headers().get(TIMESTAMP_HEADER).and_then(|value| value.to_str().ok()), request .headers() @@ -303,16 +377,21 @@ where Box::pin(async move { let response = future.await?; if let (Some(audience), Some(challenge)) = (audience, challenge) { - match verify_tonic_boot_epoch_response(&audience, challenge, response.headers()) { - Ok(epoch) => remember_peer_boot_epoch(audience, epoch), - Err(error) - if response.headers().contains_key(RPC_BOOT_EPOCH_HEADER) - || response.headers().contains_key(RPC_BOOT_EPOCH_PROOF_HEADER) => - { - debug!(error = %error, "peer boot epoch response proof was rejected") - } - Err(_) => {} + let response_state = verify_tonic_peer_replay_capabilities_response(&audience, challenge, response.headers()); + if let Err(error) = &response_state + && (response.headers().contains_key(RPC_REPLAY_CACHE_CAPABILITY_HEADER) + || response.headers().contains_key(RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER)) + { + debug!( + event = "internode_rpc_capability_proof_rejected", + component = "ecstore", + subsystem = "rpc_client", + result = "rejected", + error = %error, + "internode RPC capability proof rejected" + ) } + apply_peer_replay_response(audience, sent_state, response_state); } Ok(response) }) @@ -321,6 +400,7 @@ where pub struct TonicSignatureInterceptor { audience: Option, + body_digest_strict: bool, } impl tonic::service::Interceptor for TonicSignatureInterceptor { @@ -337,9 +417,31 @@ impl tonic::service::Interceptor for TonicSignatureInterceptor { .metadata() .get(RPC_CONTENT_SHA256_HEADER) .and_then(|value| value.to_str().ok()); + // RUSTFS_COMPAT_TODO(disk-mutation-body-digest): use cache-free v2 for peers without an authenticated boot epoch. Remove after every supported peer advertises the authenticated dynamic replay-cache capability and body-digest strict mode is the default. + // beta.11 verifies v2 body digests but stores their nonces in a fixed-size cache. + let rolling_mutation = req.extensions().get::().is_some(); + let peer_state = PEER_REPLAY_STATES + .lock() + .map_err(|_| tonic::Status::unauthenticated("RPC peer capability state unavailable"))? + .get(audience) + .copied() + .unwrap_or_default(); + let content_sha256 = if content_sha256.is_some() { + if peer_state.cache_capability == Some(PeerReplayCapability::Revoked) { + return Err(tonic::Status::unauthenticated("RPC peer replay capability changed")); + } + if rolling_mutation && !self.body_digest_strict && peer_state.boot_epoch.is_none() { + None + } else { + content_sha256 + } + } else { + content_sha256 + }; let headers = gen_tonic_signature_headers(audience, method.service(), method.method(), content_sha256) .map_err(|_| tonic::Status::unauthenticated("No valid auth token"))?; req.metadata_mut().as_mut().extend(headers); + req.extensions_mut().insert(PeerReplayStateSnapshot(peer_state)); inject_trace_context_into_metadata(req.metadata_mut()); inject_request_id_into_metadata(req.metadata_mut()); Ok(req) @@ -347,7 +449,10 @@ impl tonic::service::Interceptor for TonicSignatureInterceptor { } pub fn gen_tonic_signature_interceptor() -> TonicSignatureInterceptor { - TonicSignatureInterceptor { audience: None } + TonicSignatureInterceptor { + audience: None, + body_digest_strict: internode_rpc_body_digest_strict(), + } } pub struct NoOpInterceptor; @@ -409,6 +514,7 @@ mod tests { #[derive(Clone)] struct EpochProofService { audience: String, + include_capability: bool, seen_headers: std::sync::Arc>>, } @@ -430,29 +536,97 @@ mod tests { .expect("client challenge must be syntactically valid") .expect("authenticated client request must carry a boot epoch challenge"); let mut response = HttpResponse::new(()); - response.headers_mut().extend( - tonic_boot_epoch_response_headers(&self.audience, challenge) - .expect("test server must be able to sign an epoch proof"), - ); + let mut headers = tonic_boot_epoch_response_headers(&self.audience, challenge) + .expect("test server must be able to sign an epoch proof"); + if !self.include_capability { + headers.remove(RPC_REPLAY_CACHE_CAPABILITY_HEADER); + headers.remove(RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER); + } + response.headers_mut().extend(headers); std::future::ready(Ok(response)) } } + #[derive(Clone)] + struct MissingProofService; + + impl Service> for MissingProofService { + type Response = HttpResponse<()>; + type Error = std::convert::Infallible; + type Future = std::future::Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _request: HttpRequest<()>) -> Self::Future { + std::future::ready(Ok(HttpResponse::new(()))) + } + } + fn ensure_test_rpc_secret() { runtime_sources::ensure_test_rpc_secret(); } fn test_request() -> tonic::Request<()> { + test_request_for("Ping") + } + + fn test_request_for(method: &'static str) -> tonic::Request<()> { let mut request = tonic::Request::new(()); request .extensions_mut() - .insert(tonic::GrpcMethod::new("node_service.NodeService", "Ping")); + .insert(tonic::GrpcMethod::new("node_service.NodeService", method)); request } fn test_interceptor() -> TonicSignatureInterceptor { + test_interceptor_for("node-a:9000", false) + } + + fn test_interceptor_for(audience: &str, body_digest_strict: bool) -> TonicSignatureInterceptor { TonicSignatureInterceptor { - audience: Some("node-a:9000".to_string()), + audience: Some(audience.to_string()), + body_digest_strict, + } + } + + fn clear_peer_capability(audience: &str) { + PEER_REPLAY_STATES + .lock() + .expect("peer capability cache lock must not be poisoned") + .remove(audience); + } + + fn rolling_mutation_request(method: &'static str) -> tonic::Request<()> { + let mut request = tonic::Request::new(rustfs_protos::proto_gen::node_service::GenerallyLockRequest { + args: "canonical mutation request".to_string(), + }); + request + .extensions_mut() + .insert(tonic::GrpcMethod::new("node_service.NodeService", method)); + crate::cluster::rpc::set_tonic_rolling_mutation_body_digest(&mut request).expect("test mutation digest must be attached"); + request.map(|_| ()) + } + + fn replay_scope_request(audience: &str, method: &'static str) -> HttpRequest<()> { + let mut request = HttpRequest::builder() + .uri(format!("/node_service.NodeService/{method}")) + .body(()) + .expect("test RPC request must build"); + request.headers_mut().extend( + gen_tonic_signature_headers(audience, "node_service.NodeService", method, None).expect("v2 test headers must mint"), + ); + request + .extensions_mut() + .insert(PeerReplayStateSnapshot(peer_replay_state(audience))); + request + } + + fn authenticated_peer_response(boot_epoch: Uuid, dynamic_replay_cache: bool) -> AuthenticatedPeerReplayCapabilities { + AuthenticatedPeerReplayCapabilities { + boot_epoch, + dynamic_replay_cache, } } @@ -567,6 +741,431 @@ mod tests { ); } + #[test] + fn unknown_peer_mutations_use_cache_free_unsigned_v2() { + ensure_test_rpc_secret(); + let audience = "legacy-body-digest-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, false); + for method in ["Lock", "WriteAll"] { + let request = interceptor + .call(rolling_mutation_request(method)) + .expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some("UNSIGNED-PAYLOAD") + ); + assert_eq!( + request + .metadata() + .get("x-rustfs-rpc-nonce") + .and_then(|value| value.to_str().ok()), + Some("unsigned") + ); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + &format!("/node_service.NodeService/{method}"), + request.metadata().as_ref(), + ) + .is_ok(), + "the cache-free request must retain valid audience- and method-bound v2 authentication" + ); + } + } + + #[test] + fn unknown_peer_exact_body_contract_remains_body_bound() { + ensure_test_rpc_secret(); + let audience = "exact-body-contract-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, false); + let mut request = test_request_for("ScannerActivity"); + crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, b"exact scanner activity body") + .expect("test exact body digest must be attached"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test request must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/ScannerActivity", + request.metadata().as_ref(), + ) + .is_ok() + ); + } + + #[test] + fn unknown_peer_iam_mutation_helper_remains_body_bound() { + ensure_test_rpc_secret(); + let audience = "exact-iam-mutation-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, false); + let mut request = tonic::Request::new(rustfs_protos::proto_gen::node_service::DeleteUserRequest { + access_key: "target-access-key".to_string(), + }); + request + .extensions_mut() + .insert(tonic::GrpcMethod::new("node_service.NodeService", "DeleteUser")); + crate::cluster::rpc::set_tonic_mutation_body_digest(&mut request).expect("test IAM mutation digest must be attached"); + let request = request.map(|_| ()); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test IAM mutation must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/DeleteUser", + request.metadata().as_ref(), + ) + .is_ok(), + "IAM mutations must remain body-bound before capability discovery" + ); + } + + #[test] + fn authenticated_replay_cache_capability_enables_body_binding() { + ensure_test_rpc_secret(); + let audience = "body-digest-capable-client-test:9000"; + clear_peer_capability(audience); + let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new())); + let service = EpochProofService { + audience: audience.to_string(), + include_capability: true, + seen_headers, + }; + let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("authenticated capability probe must complete"); + + let mut interceptor = test_interceptor_for(audience, false); + let request = rolling_mutation_request("Lock"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test mutation must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + let nonce = request + .metadata() + .get("x-rustfs-rpc-nonce") + .and_then(|value| value.to_str().ok()) + .and_then(|value| Uuid::parse_str(value).ok()) + .expect("capable peer body-bound mutation must carry a UUID nonce"); + assert!(!nonce.is_nil()); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/Lock", + request.metadata().as_ref(), + ) + .is_ok(), + "the body-bound request must retain valid audience- and method-bound v2 authentication" + ); + clear_peer_capability(audience); + } + + #[test] + fn invalid_capability_proof_does_not_enable_body_binding() { + ensure_test_rpc_secret(); + let audience = "invalid-capability-client-test:9000"; + clear_peer_capability(audience); + let service = EpochProofService { + audience: "wrong-capability-audience:9000".to_string(), + include_capability: true, + seen_headers: std::sync::Arc::new(Mutex::new(Vec::new())), + }; + let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("invalid capability response must still complete"); + + let mut interceptor = test_interceptor_for(audience, false); + let request = interceptor + .call(rolling_mutation_request("Lock")) + .expect("legacy-compatible mutation must still be signed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some("UNSIGNED-PAYLOAD") + ); + } + + #[test] + fn legacy_boot_proof_keeps_mutations_body_bound_and_enables_non_ping_v3() { + ensure_test_rpc_secret(); + let audience = "legacy-boot-proof-client-test:9000"; + clear_peer_capability(audience); + let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new())); + let service = EpochProofService { + audience: audience.to_string(), + include_capability: false, + seen_headers: seen_headers.clone(), + }; + let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("legacy boot proof response must complete"); + + let state = peer_replay_state(audience); + assert!(state.boot_epoch.is_some(), "authenticated legacy proof must enable replay-scoped v3"); + assert_eq!(state.cache_capability, None); + + let mut interceptor = test_interceptor_for(audience, false); + let request = rolling_mutation_request("Lock"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test mutation must carry its digest") + .to_string(); + let request = interceptor.call(request).expect("legacy-compatible mutation must be signed"); + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + let (metadata, extensions, body) = request.into_parts(); + let mut request = HttpRequest::new(body); + *request.uri_mut() = "/node_service.NodeService/Lock".parse().expect("test RPC URI must parse"); + *request.headers_mut() = metadata.into_headers(); + *request.extensions_mut() = extensions; + futures::executor::block_on(channel.call(request)).expect("legacy strict-compatible lock request must complete"); + + let headers = seen_headers.lock().expect("test header capture lock must not be poisoned"); + assert!( + headers[1].contains_key(RPC_REPLAY_SCOPE_VERSION_HEADER), + "authenticated legacy boot proof must enable v3 on a non-Ping request" + ); + } + + #[test] + fn reordered_capability_responses_cannot_undo_newer_state() { + let audience = "reordered-capability-client-test:9000"; + let epoch_one = Uuid::new_v4(); + let epoch_two = Uuid::new_v4(); + clear_peer_capability(audience); + + let unknown = PeerReplayState::default(); + apply_peer_replay_response(audience.to_string(), unknown, Ok(authenticated_peer_response(epoch_one, true))); + apply_peer_replay_response(audience.to_string(), unknown, Err(std::io::Error::other("delayed legacy response"))); + let epoch_one_state = PeerReplayState { + boot_epoch: Some(epoch_one), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch: epoch_one }), + }; + assert_eq!(peer_replay_state(audience), epoch_one_state); + + apply_peer_replay_response(audience.to_string(), epoch_one_state, Err(std::io::Error::other("rollback response"))); + apply_peer_replay_response(audience.to_string(), epoch_one_state, Ok(authenticated_peer_response(epoch_one, true))); + assert_eq!( + peer_replay_state(audience), + PeerReplayState { + boot_epoch: Some(epoch_one), + cache_capability: Some(PeerReplayCapability::Revoked), + } + ); + + let revoked = peer_replay_state(audience); + apply_peer_replay_response(audience.to_string(), revoked, Ok(authenticated_peer_response(epoch_two, true))); + apply_peer_replay_response(audience.to_string(), epoch_one_state, Ok(authenticated_peer_response(epoch_one, true))); + assert_eq!( + peer_replay_state(audience), + PeerReplayState { + boot_epoch: Some(epoch_two), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch: epoch_two }), + } + ); + clear_peer_capability(audience); + } + + #[test] + fn stale_capability_response_cannot_cross_a_new_boot_epoch() { + let audience = "cross-epoch-capability-client-test:9000"; + let epoch_one = Uuid::new_v4(); + let epoch_two = Uuid::new_v4(); + let epoch_three = Uuid::new_v4(); + clear_peer_capability(audience); + + let revoked_epoch_one = PeerReplayState { + boot_epoch: Some(epoch_one), + cache_capability: Some(PeerReplayCapability::Revoked), + }; + PEER_REPLAY_STATES + .lock() + .expect("peer replay state lock must not be poisoned") + .insert(audience.to_string(), revoked_epoch_one); + + apply_peer_replay_response( + audience.to_string(), + revoked_epoch_one, + Ok(authenticated_peer_response(epoch_three, false)), + ); + apply_peer_replay_response(audience.to_string(), revoked_epoch_one, Ok(authenticated_peer_response(epoch_two, true))); + + assert_eq!( + peer_replay_state(audience), + PeerReplayState { + boot_epoch: Some(epoch_three), + cache_capability: Some(PeerReplayCapability::Revoked), + }, + "a stale dynamic-cache proof must not cross a newer authenticated boot epoch" + ); + clear_peer_capability(audience); + } + + #[test] + fn interceptor_snapshot_prevents_delayed_legacy_response_from_revoking_capability() { + ensure_test_rpc_secret(); + let audience = "capability-snapshot-client-test:9000"; + clear_peer_capability(audience); + let boot_epoch = Uuid::new_v4(); + let mut interceptor = test_interceptor_for(audience, false); + let request = interceptor + .call(rolling_mutation_request("Lock")) + .expect("legacy-compatible request must pass the interceptor"); + assert_eq!( + request + .extensions() + .get::() + .map(|snapshot| snapshot.0), + Some(PeerReplayState::default()), + "interceptor must preserve its unknown-state admission snapshot" + ); + + let capable_state = PeerReplayState { + boot_epoch: Some(boot_epoch), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch }), + }; + PEER_REPLAY_STATES + .lock() + .expect("peer capability cache lock must not be poisoned") + .insert(audience.to_string(), capable_state); + let (metadata, extensions, body) = request.into_parts(); + let mut request = HttpRequest::new(body); + *request.uri_mut() = "/node_service.NodeService/Lock".parse().expect("test RPC URI must parse"); + *request.headers_mut() = metadata.into_headers(); + *request.extensions_mut() = extensions; + let mut channel = ReplayScopeChannel::new(MissingProofService, Some(audience.to_string())); + + futures::executor::block_on(channel.call(request)).expect("in-flight request response must complete"); + + assert_eq!(peer_replay_state(audience), capable_state); + clear_peer_capability(audience); + } + + #[test] + fn strict_mode_keeps_unknown_peer_mutations_body_bound() { + ensure_test_rpc_secret(); + let audience = "strict-body-digest-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, true); + let request = rolling_mutation_request("WriteAll"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test mutation must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + let nonce = request + .metadata() + .get("x-rustfs-rpc-nonce") + .and_then(|value| value.to_str().ok()) + .and_then(|value| Uuid::parse_str(value).ok()) + .expect("strict body-bound mutation must carry a UUID nonce"); + assert!(!nonce.is_nil()); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/WriteAll", + request.metadata().as_ref(), + ) + .is_ok() + ); + } + + #[test] + fn missing_capability_after_pin_fails_closed() { + ensure_test_rpc_secret(); + let audience = "revoked-capability-client-test:9000"; + let boot_epoch = Uuid::new_v4(); + PEER_REPLAY_STATES + .lock() + .expect("peer capability cache lock must not be poisoned") + .insert( + audience.to_string(), + PeerReplayState { + boot_epoch: Some(boot_epoch), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch }), + }, + ); + let mut channel = ReplayScopeChannel::new(MissingProofService, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("legacy response must complete before capability rejection"); + + let mut interceptor = test_interceptor_for(audience, false); + let error = interceptor + .call(rolling_mutation_request("Lock")) + .expect_err("a peer that loses its pinned capability must fail closed"); + + assert_eq!(error.code(), tonic::Code::Unauthenticated); + assert_eq!(error.message(), "RPC peer replay capability changed"); + clear_peer_capability(audience); + } + #[test] fn test_signature_interceptor_binds_audience_from_peer_uri() { let interceptor = TonicInterceptor::Signature(gen_tonic_signature_interceptor()) @@ -583,27 +1182,15 @@ mod tests { fn replay_scope_channel_uses_epoch_proof_before_sending_v3() { ensure_test_rpc_secret(); let audience = "replay-scope-client-test:9000"; - PEER_BOOT_EPOCHS - .lock() - .expect("peer epoch cache lock must not be poisoned") - .remove(audience); + clear_peer_capability(audience); let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new())); let service = EpochProofService { audience: audience.to_string(), + include_capability: true, seen_headers: seen_headers.clone(), }; let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); - let make_request = || { - let mut request = HttpRequest::builder() - .uri("/node_service.NodeService/Ping") - .body(()) - .expect("test RPC request must build"); - request.headers_mut().extend( - gen_tonic_signature_headers(audience, "node_service.NodeService", "Ping", None) - .expect("v2 test headers must mint"), - ); - request - }; + let make_request = || replay_scope_request(audience, "Ping"); futures::executor::block_on(channel.call(make_request())).expect("first request must complete"); futures::executor::block_on(channel.call(make_request())).expect("second request must complete"); @@ -619,10 +1206,7 @@ mod tests { headers[1].contains_key(RPC_REPLAY_SCOPE_VERSION_HEADER), "the second request must carry the replay-scoped v3 signature" ); - PEER_BOOT_EPOCHS - .lock() - .expect("peer epoch cache lock must not be poisoned") - .remove(audience); + clear_peer_capability(audience); } #[test] diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 48c3ce696..a73ac22fe 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -73,10 +73,14 @@ pub const RPC_REPLAY_SCOPE_NONCE_HEADER: &str = "x-rustfs-rpc-replay-nonce"; pub const RPC_BOOT_EPOCH_HEADER: &str = "x-rustfs-rpc-boot-epoch"; pub const RPC_BOOT_EPOCH_CHALLENGE_HEADER: &str = "x-rustfs-rpc-boot-epoch-challenge"; pub const RPC_BOOT_EPOCH_PROOF_HEADER: &str = "x-rustfs-rpc-boot-epoch-proof"; +pub(crate) const RPC_REPLAY_CACHE_CAPABILITY_HEADER: &str = "x-rustfs-rpc-replay-cache-capability"; +pub(crate) const RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER: &str = "x-rustfs-rpc-replay-cache-capability-proof"; const RPC_REPLAY_SCOPE_VERSION_V3: &str = "3"; const RPC_RESPONSE_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-response-proof-v1\0"; const RPC_REPLAY_SCOPE_DOMAIN: &[u8] = b"rustfs-rpc-replay-scope-v3\0"; const RPC_BOOT_EPOCH_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-boot-epoch-proof-v1\0"; +const RPC_REPLAY_CACHE_CAPABILITY_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-replay-cache-capability-proof-v1\0"; +const RPC_REPLAY_CACHE_CAPABILITY_V1: &str = "dynamic-replay-cache-v1"; const HTTP_PUT_FILE_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-auth-v1\0"; const HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-capability-v1\0"; const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; @@ -102,6 +106,10 @@ static INTERNODE_RPC_BODY_DIGEST_STRICT: LazyLock = LazyLock::new(|| { rustfs_config::DEFAULT_INTERNODE_RPC_BODY_DIGEST_STRICT, ) }); + +pub(crate) fn internode_rpc_body_digest_strict() -> bool { + *INTERNODE_RPC_BODY_DIGEST_STRICT +} static INTERNODE_RPC_REPLAY_SCOPE_STRICT: LazyLock = LazyLock::new(|| { get_env_bool( rustfs_config::ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT, @@ -789,6 +797,50 @@ fn verify_boot_epoch_proof(secret: &str, audience: &str, challenge: Uuid, boot_e .map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid RPC boot epoch proof")) } +fn update_replay_cache_capability_proof(mac: &mut HmacSha256, audience: &str, challenge: Uuid, boot_epoch: Uuid) { + mac.update(RPC_REPLAY_CACHE_CAPABILITY_PROOF_DOMAIN); + for part in [ + audience.as_bytes(), + b"|", + challenge.as_bytes(), + b"|", + boot_epoch.as_bytes(), + b"|", + RPC_REPLAY_CACHE_CAPABILITY_V1.as_bytes(), + ] { + mac.update(part); + } +} + +fn generate_replay_cache_capability_proof( + secret: &str, + audience: &str, + challenge: Uuid, + boot_epoch: Uuid, +) -> std::io::Result { + let mut mac = + ::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_replay_cache_capability_proof(&mut mac, audience, challenge, boot_epoch); + Ok(general_purpose::STANDARD.encode(mac.finalize().into_bytes())) +} + +fn verify_replay_cache_capability_proof( + secret: &str, + audience: &str, + challenge: Uuid, + boot_epoch: Uuid, + proof: &str, +) -> std::io::Result<()> { + let proof = general_purpose::STANDARD + .decode(proof) + .map_err(|_| std::io::Error::other("Invalid RPC replay cache capability proof"))?; + let mut mac = + ::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_replay_cache_capability_proof(&mut mac, audience, challenge, boot_epoch); + mac.verify_slice(&proof) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid RPC replay cache capability proof")) +} + fn non_nil_uuid(value: &str, name: &str) -> std::io::Result { let value = Uuid::parse_str(value).map_err(|_| std::io::Error::other(format!("Invalid {name}")))?; (!value.is_nil()) @@ -871,15 +923,34 @@ pub fn tonic_boot_epoch_challenge(headers: &HeaderMap) -> std::io::Result