From 719c0d6ef098dc8477ed814df78ed0a0a0cd44d6 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 30 Jul 2026 07:12:38 +0800 Subject: [PATCH] feat(rpc): add replay-scoped internode authentication (#5455) --- crates/config/src/constants/internode.rs | 40 +- .../src/internode_rpc_signature_e2e_test.rs | 231 ++++++++- .../e2e_test/src/reliant/grpc_lock_client.rs | 4 +- crates/e2e_test/src/storage_api.rs | 12 +- crates/ecstore/src/api/mod.rs | 16 +- crates/ecstore/src/cluster/rpc/client.rs | 223 ++++++++- crates/ecstore/src/cluster/rpc/http_auth.rs | 444 +++++++++++++++++- crates/ecstore/src/cluster/rpc/mod.rs | 15 +- .../src/cluster/rpc/peer_rest_client.rs | 9 +- .../ecstore/src/cluster/rpc/peer_s3_client.rs | 6 +- crates/ecstore/src/cluster/rpc/remote_disk.rs | 10 +- .../ecstore/src/cluster/rpc/remote_locker.rs | 7 +- crates/io-metrics/src/internode_metrics.rs | 25 + rustfs/src/server/http.rs | 42 +- rustfs/src/storage/storage_api.rs | 22 +- rustfs/src/storage_api.rs | 3 +- 16 files changed, 1022 insertions(+), 87 deletions(-) diff --git a/crates/config/src/constants/internode.rs b/crates/config/src/constants/internode.rs index 64a76bf45..45f979ddc 100644 --- a/crates/config/src/constants/internode.rs +++ b/crates/config/src/constants/internode.rs @@ -158,17 +158,33 @@ pub const DEFAULT_INTERNODE_RPC_BODY_DIGEST_STRICT: bool = false; // rolling upgrades until the fleet-wide body-digest fallback counter reads zero. const _: () = assert!(!DEFAULT_INTERNODE_RPC_BODY_DIGEST_STRICT); -/// Capacity (distinct nonces) of the process-local internode RPC replay cache that enforces -/// one-time consumption of body-bound v2 signatures. +/// Require the replay-scoped internode RPC signature after the fleet has converged on it. /// -/// The cache retains each nonce for the ~10-minute signature freshness envelope, so the steady -/// state holds roughly `mutating RPS x 601s` entries; the default sustains ~1,700 body-bound -/// mutating RPCs per second (about 120 MiB worst case, allocated only under sustained load). -/// Overflow fails closed — legitimate signed traffic is the only thing that can fill the cache -/// (replays are rejected before insertion, and an attacker cannot mint valid nonces without the -/// shared secret) — and increments +/// The default keeps v1/v2 peers available during a rolling upgrade. Operators may set this only +/// after `rustfs_system_network_internode_replay_scope_fallback_total` remains zero for a full +/// release window. The node still accepts a v2-authenticated `Ping` carrying an epoch challenge: +/// that narrowly scoped bootstrap lets an upgraded client learn the receiving process epoch and +/// immediately retry with the replay-scoped signature after a peer restart. +pub const ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT: &str = "RUSTFS_INTERNODE_RPC_REPLAY_SCOPE_STRICT"; +pub const DEFAULT_INTERNODE_RPC_REPLAY_SCOPE_STRICT: bool = false; + +// Compile-time invariant: mixed-version clusters must remain available until operators make the +// observed fallback counter an explicit strictness decision. +const _: () = assert!(!DEFAULT_INTERNODE_RPC_REPLAY_SCOPE_STRICT); + +/// Capacity (distinct nonces) of the process-local internode RPC replay cache that enforces +/// one-time consumption of authenticated RPC signatures. +/// +/// The cache retains each nonce for the ~10-minute signature freshness envelope. Once peers use +/// replay-scoped v3 authentication, every authenticated RPC consumes one entry, so the steady +/// state holds roughly `authenticated RPC RPS x 601s` entries. The default sustains about 1,700 +/// authenticated RPCs per second (about 120 MiB worst case, allocated only under sustained load); +/// operators must size it for the node's aggregate peak RPC rate before enabling strict replay +/// scope. Overflow fails closed — legitimate signed traffic is the only thing that can fill the +/// cache (replays are rejected before insertion, and an attacker cannot mint valid nonces without +/// the shared secret) — and increments /// `rustfs_system_network_internode_replay_cache_overflow_total`, so a sustained non-zero overflow -/// counter means this capacity is undersized for the node's peak mutation rate. +/// counter means this capacity is undersized for the node's peak authenticated RPC rate. pub const ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY: &str = "RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY"; pub const DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY: usize = 1_048_576; @@ -354,6 +370,12 @@ mod tests { assert_eq!(ENV_INTERNODE_RPC_BODY_DIGEST_STRICT, "RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT"); } + #[test] + fn internode_replay_scope_strict_env_name_is_stable() { + // The fail-open default invariant is asserted at compile time next to the definition. + assert_eq!(ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT, "RUSTFS_INTERNODE_RPC_REPLAY_SCOPE_STRICT"); + } + #[test] fn internode_replay_cache_capacity_defaults_and_env_name() { assert_eq!(ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, "RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY"); diff --git a/crates/e2e_test/src/internode_rpc_signature_e2e_test.rs b/crates/e2e_test/src/internode_rpc_signature_e2e_test.rs index 2addee97e..c7632d013 100644 --- a/crates/e2e_test/src/internode_rpc_signature_e2e_test.rs +++ b/crates/e2e_test/src/internode_rpc_signature_e2e_test.rs @@ -13,8 +13,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Cross-process replay / tamper acceptance for the internode NodeService v2 RPC -//! signature (). +//! Cross-process replay / tamper acceptance for internode NodeService RPC +//! signatures (, +//! ). //! //! # Why this exists on top of the in-process tests //! @@ -78,6 +79,8 @@ //! | mixed version: legacy-only still served, not blocked | [`legacy_only_signature_is_accepted_in_default_posture`] | //! | strict flip closes the signature downgrade | [`signature_strict_rejects_legacy_only_downgrade`] | //! | strict flip closes the body-digest downgrade, incl. v1 | [`body_digest_strict_rejects_digestless_mutation`] | +//! | replay scope binds every RPC and rejects restart replay | [`replay_scope_rejects_replay_path_transplant_and_stale_epoch_e2e`] | +//! | strict replay scope allows only Ping bootstrap before v3 | [`replay_scope_strict_requires_v3_after_ping_bootstrap_e2e`] | //! //! Two acceptance items are deliberately left to the in-process tests. A stale //! timestamp cannot be forged from outside — it is inside the HMAC — so @@ -88,18 +91,20 @@ use crate::common::{RustFSTestEnvironment, init_logging}; use crate::storage_api::internode_rpc_signature::{ - TONIC_RPC_PREFIX, gen_signature_headers, gen_tonic_signature_headers, node_service_time_out_client_no_auth, + TONIC_RPC_PREFIX, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, + node_service_time_out_client_no_auth, verify_tonic_boot_epoch_response, }; use http::{HeaderMap, Method}; use rustfs_config::{ - ENV_INTERNODE_RPC_BODY_DIGEST_STRICT, ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, ENV_INTERNODE_RPC_SIGNATURE_STRICT, + ENV_INTERNODE_RPC_BODY_DIGEST_STRICT, ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT, + ENV_INTERNODE_RPC_SIGNATURE_STRICT, }; use rustfs_protos::canonical_make_volume_request_body; -use rustfs_protos::proto_gen::node_service::{MakeVolumeRequest, MakeVolumeResponse}; +use rustfs_protos::proto_gen::node_service::{MakeVolumeRequest, MakeVolumeResponse, PingRequest, PingResponse}; use serial_test::serial; use sha2::{Digest, Sha256}; use std::error::Error; -use tonic::{Code, Request, Status}; +use tonic::{Code, Request, Response, Status}; use uuid::Uuid; type TestResult = Result<(), Box>; @@ -115,12 +120,14 @@ const TEST_RPC_SECRET: &str = "rustfs-internode-signature-e2e-secret"; /// clears authentication stops harmlessly at `find_disk`. const ABSENT_DISK: &str = "/nonexistent/rustfs-signature-e2e-disk"; -/// Wire names of the two v2 headers these tests edit. They are `pub(crate)` in -/// ecstore, so they are repeated here rather than imported — [`overwrite_header`] -/// asserts the header it replaces was actually present, which turns a rename -/// into a loud failure instead of silently reducing an attack to a no-op. +/// Wire names of the v2 and replay-scope headers these black-box tests edit. They are +/// `pub(crate)` in ecstore, so they are repeated here rather than imported. +/// [`overwrite_header`] asserts the header it replaces was actually present, which turns a +/// rename into a loud failure instead of silently reducing an attack to a no-op. const CONTENT_SHA256_HEADER: &str = "x-rustfs-content-sha256"; const NONCE_HEADER: &str = "x-rustfs-rpc-nonce"; +const TIMESTAMP_HEADER: &str = "x-rustfs-timestamp"; +const BOOT_EPOCH_CHALLENGE_HEADER: &str = "x-rustfs-rpc-boot-epoch-challenge"; /// gRPC service name carried in the signed scope, i.e. `TONIC_RPC_PREFIX` /// without its leading `/`. @@ -155,19 +162,28 @@ fn align_rpc_secret_with_server() { /// /// Uses the no-cleanup spawn so a `pkill` pattern cannot reap servers belonging /// to other tests running in the same binary. -async fn start_server(extra_env: &[(&str, &str)]) -> Result> { - let mut env = RustFSTestEnvironment::new().await?; +fn server_env(extra_env: &[(&'static str, &'static str)]) -> Vec<(&'static str, &'static str)> { let mut child_env = vec![ ("RUSTFS_RPC_SECRET", TEST_RPC_SECRET), (ENV_INTERNODE_RPC_SIGNATURE_STRICT, "false"), (ENV_INTERNODE_RPC_BODY_DIGEST_STRICT, "false"), + (ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT, "false"), (ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, "1048576"), ]; child_env.extend_from_slice(extra_env); - env.start_rustfs_server_without_cleanup_with_env(&child_env).await?; + child_env +} + +async fn start_server_with_env(child_env: &[(&str, &str)]) -> Result> { + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_without_cleanup_with_env(child_env).await?; Ok(env) } +async fn start_server(extra_env: &[(&'static str, &'static str)]) -> Result> { + start_server_with_env(&server_env(extra_env)).await +} + /// Stop the child and drop the cached gRPC channel for its address. /// /// `node_service_time_out_client_no_auth` memoises channels in a process-global @@ -238,12 +254,83 @@ fn overwrite_header(headers: &mut HeaderMap, name: &'static str, value: &str) { /// and nothing else — no interceptor adds or rewrites auth metadata, so the /// bytes on the wire are the ones the test chose. async fn call_make_volume(url: &str, request: MakeVolumeRequest, headers: HeaderMap) -> Result { + call_make_volume_response(url, request, headers) + .await + .map(Response::into_inner) +} + +async fn call_make_volume_response( + url: &str, + request: MakeVolumeRequest, + headers: HeaderMap, +) -> Result, Status> { let mut client = node_service_time_out_client_no_auth(&url.to_string()) .await .map_err(|err| Status::unavailable(format!("cannot reach the node service: {err}")))?; let mut rpc_request = Request::new(request); rpc_request.metadata_mut().as_mut().extend(headers); - client.make_volume(rpc_request).await.map(|response| response.into_inner()) + client.make_volume(rpc_request).await +} + +async fn call_ping_response(url: &str, headers: HeaderMap) -> Result, Status> { + let mut client = node_service_time_out_client_no_auth(&url.to_string()) + .await + .map_err(|err| Status::unavailable(format!("cannot reach the node service: {err}")))?; + let mut rpc_request = Request::new(PingRequest { + version: 1, + body: bytes::Bytes::new(), + }); + rpc_request.metadata_mut().as_mut().extend(headers); + client.ping(rpc_request).await +} + +fn attach_boot_epoch_challenge(headers: &mut HeaderMap) -> Uuid { + let challenge = Uuid::new_v4(); + headers.insert( + BOOT_EPOCH_CHALLENGE_HEADER, + challenge.to_string().parse().expect("UUID must be a valid header value"), + ); + challenge +} + +fn mint_replay_scope_headers(audience: &str, path: &str, content_sha256: &str, boot_epoch: Uuid) -> HeaderMap { + let mut headers = mint_v2_headers(audience, "MakeVolume", Some(content_sha256)); + let timestamp = headers + .get(TIMESTAMP_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("v2 headers must carry a timestamp") + .to_string(); + headers.extend( + gen_tonic_replay_scope_headers(audience, path, ×tamp, content_sha256, boot_epoch) + .expect("replay-scope headers must mint with the aligned RPC secret"), + ); + headers +} + +async fn learn_boot_epoch_from_make_volume(url: &str, audience: &str) -> Uuid { + let request = make_volume_request("signature-e2e-epoch-bootstrap"); + let mut headers = mint_v2_headers(audience, "MakeVolume", Some(&canonical_digest(&request))); + let challenge = attach_boot_epoch_challenge(&mut headers); + let response = call_make_volume_response(url, request, headers) + .await + .expect("v2 request with epoch challenge must clear default authentication"); + let boot_epoch = verify_tonic_boot_epoch_response(audience, challenge, response.metadata().as_ref()) + .expect("server must HMAC-authenticate the advertised boot epoch"); + assert_authenticated( + Ok(response.into_inner()), + "a v2 epoch-challenge request in the default replay-scope posture", + ); + boot_epoch +} + +async fn learn_boot_epoch_from_ping(url: &str, audience: &str) -> Uuid { + let mut headers = mint_v2_headers(audience, "Ping", None); + let challenge = attach_boot_epoch_challenge(&mut headers); + let response = call_ping_response(url, headers) + .await + .expect("v2 Ping with an epoch challenge must bootstrap strict replay scope"); + verify_tonic_boot_epoch_response(audience, challenge, response.metadata().as_ref()) + .expect("strict replay-scope Ping must return a valid boot epoch proof") } /// Assert a call cleared authentication. @@ -332,6 +419,122 @@ async fn internode_rpc_signature_default_posture_e2e() -> TestResult { Ok(()) } +/// A replay-scoped signature is usable exactly once against the exact gRPC path and the server +/// process epoch that minted it. This crosses the child-process boundary twice: the HMAC-protected +/// epoch is learned from a real response, then the same server is restarted in place to prove its +/// replacement epoch rejects the captured request even though the nonce cache is necessarily new. +#[tokio::test] +#[serial] +async fn replay_scope_rejects_replay_path_transplant_and_stale_epoch_e2e() -> TestResult { + init_logging(); + align_rpc_secret_with_server(); + let child_env = server_env(&[]); + let mut env = start_server_with_env(&child_env).await?; + let url = env.url.clone(); + let audience = audience_of(&env); + let boot_epoch = learn_boot_epoch_from_make_volume(&url, &audience).await; + + let request = make_volume_request("replay-scope-e2e-once"); + let captured = mint_replay_scope_headers( + &audience, + &format!("{TONIC_RPC_PREFIX}/MakeVolume"), + &canonical_digest(&request), + boot_epoch, + ); + assert_authenticated( + call_make_volume(&url, request.clone(), captured.clone()).await, + "the first replay-scoped mutation delivery", + ); + assert_rejected( + call_make_volume(&url, request.clone(), captured).await, + Code::Unauthenticated, + None, + "the same replay-scoped mutation delivered twice", + ); + + let transplanted = + mint_replay_scope_headers(&audience, &format!("{TONIC_RPC_PREFIX}/Ping"), &canonical_digest(&request), boot_epoch); + assert_rejected( + call_make_volume(&url, request.clone(), transplanted).await, + Code::Unauthenticated, + None, + "a replay-scoped Ping signature transplanted onto MakeVolume", + ); + + let stale_epoch = mint_replay_scope_headers( + &audience, + &format!("{TONIC_RPC_PREFIX}/MakeVolume"), + &canonical_digest(&request), + boot_epoch, + ); + env.restart_server_preserving_data(Vec::new(), &child_env).await?; + rustfs_protos::evict_failed_connection(&url).await; + assert_rejected( + call_make_volume(&url, request.clone(), stale_epoch).await, + Code::Unauthenticated, + None, + "a replay-scoped signature captured before the receiving process restart", + ); + + let restarted_epoch = learn_boot_epoch_from_make_volume(&url, &audience).await; + assert_ne!(boot_epoch, restarted_epoch, "a restarted child process must advertise a new boot epoch"); + let fresh_epoch = mint_replay_scope_headers( + &audience, + &format!("{TONIC_RPC_PREFIX}/MakeVolume"), + &canonical_digest(&request), + restarted_epoch, + ); + assert_authenticated( + call_make_volume(&url, request, fresh_epoch).await, + "a replay-scoped mutation signed with the replacement process epoch", + ); + + stop_server(env, &url).await; + Ok(()) +} + +/// Strict replay scope leaves one authenticated v2 bootstrap: `Ping` carrying a fresh challenge. +/// A mutating v2 request cannot use that lane; once the epoch proof is returned, the first v3 +/// mutation succeeds. This protects a server restart without reopening a general downgrade path. +#[tokio::test] +#[serial] +async fn replay_scope_strict_requires_v3_after_ping_bootstrap_e2e() -> TestResult { + init_logging(); + align_rpc_secret_with_server(); + let env = start_server(&[(ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT, "true")]).await?; + let url = env.url.clone(); + let audience = audience_of(&env); + + let v2_request = make_volume_request("replay-scope-e2e-strict-v2"); + assert_rejected( + call_make_volume( + &url, + v2_request.clone(), + mint_v2_headers(&audience, "MakeVolume", Some(&canonical_digest(&v2_request))), + ) + .await, + Code::Unauthenticated, + None, + "a v2 mutation after replay-scope strictness is enabled", + ); + + let boot_epoch = learn_boot_epoch_from_ping(&url, &audience).await; + let request = make_volume_request("replay-scope-e2e-strict-v3"); + let replay_scoped = mint_replay_scope_headers( + &audience, + &format!("{TONIC_RPC_PREFIX}/MakeVolume"), + &canonical_digest(&request), + boot_epoch, + ); + assert_authenticated( + call_make_volume(&url, request, replay_scoped).await, + "a replay-scoped mutation after Ping bootstrap under strict replay scope", + ); + + stop_server(env, &url).await; + Ok(()) +} + /// Baseline: correctly signed mutations are accepted, both with and without a /// body digest. /// diff --git a/crates/e2e_test/src/reliant/grpc_lock_client.rs b/crates/e2e_test/src/reliant/grpc_lock_client.rs index e1acd50d6..cfacf10f7 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_client.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_client.rs @@ -24,7 +24,7 @@ use rustfs_protos::proto_gen::node_service::{BatchGenerallyLockRequest, Generall use tonic::Request; use tracing::{info, warn}; -use crate::storage_api::grpc_lock::{TonicInterceptor, node_service_time_out_client_no_auth}; +use crate::storage_api::grpc_lock::{AuthenticatedChannel, TonicInterceptor, node_service_time_out_client_no_auth}; /// gRPC lock client without authentication for testing /// Similar to RemoteClient but uses no_auth client @@ -42,7 +42,7 @@ impl GrpcLockClient { &self, ) -> Result< rustfs_protos::proto_gen::node_service::node_service_client::NodeServiceClient< - tonic::service::interceptor::InterceptedService, + tonic::service::interceptor::InterceptedService, >, > { node_service_time_out_client_no_auth(&self.addr) diff --git a/crates/e2e_test/src/storage_api.rs b/crates/e2e_test/src/storage_api.rs index a09c734eb..f8337aa99 100644 --- a/crates/e2e_test/src/storage_api.rs +++ b/crates/e2e_test/src/storage_api.rs @@ -16,9 +16,12 @@ pub(crate) use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys; #[cfg(test)] pub(crate) use rustfs_ecstore::api::disk::{VolumeInfo, WalkDirOptions}; +pub(crate) use rustfs_ecstore::api::rpc::{AuthenticatedChannel, TonicInterceptor, node_service_time_out_client_no_auth}; #[cfg(test)] -pub(crate) use rustfs_ecstore::api::rpc::{TONIC_RPC_PREFIX, gen_signature_headers, gen_tonic_signature_headers}; -pub(crate) use rustfs_ecstore::api::rpc::{TonicInterceptor, node_service_time_out_client_no_auth}; +pub(crate) use rustfs_ecstore::api::rpc::{ + TONIC_RPC_PREFIX, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, + verify_tonic_boot_epoch_response, +}; #[cfg(test)] pub(crate) use rustfs_ecstore::api::rpc::{gen_tonic_signature_interceptor, node_service_time_out_client}; @@ -30,7 +33,7 @@ pub(crate) mod node_interact { } pub(crate) mod grpc_lock { - pub(crate) use super::{TonicInterceptor, node_service_time_out_client_no_auth}; + pub(crate) use super::{AuthenticatedChannel, TonicInterceptor, node_service_time_out_client_no_auth}; } /// Signing/transport surface used by the cross-process internode RPC signature @@ -40,7 +43,8 @@ pub(crate) mod grpc_lock { #[cfg(test)] pub(crate) mod internode_rpc_signature { pub(crate) use super::{ - TONIC_RPC_PREFIX, gen_signature_headers, gen_tonic_signature_headers, node_service_time_out_client_no_auth, + TONIC_RPC_PREFIX, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, + node_service_time_out_client_no_auth, verify_tonic_boot_epoch_response, }; } diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 109e6f6b0..d5bed138b 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -416,13 +416,15 @@ pub mod rio { pub mod rpc { pub use crate::cluster::rpc::{ - 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_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, verify_rpc_signature, - verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, - verify_tonic_rpc_signature, + 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, }; } diff --git a/crates/ecstore/src/cluster/rpc/client.rs b/crates/ecstore/src/cluster/rpc/client.rs index 5bd2c25be..59082b659 100644 --- a/crates/ecstore/src/cluster/rpc/client.rs +++ b/crates/ecstore/src/cluster/rpc/client.rs @@ -12,11 +12,20 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::cluster::rpc::http_auth::RPC_CONTENT_SHA256_HEADER; -use crate::cluster::rpc::{gen_tonic_signature_headers, normalize_tonic_rpc_audience}; +#[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, +}; +#[cfg(test)] +use crate::cluster::rpc::{tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers}; use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError}; use crate::runtime::sources as runtime_sources; -use http::Uri; +use http::{Request as HttpRequest, Response as HttpResponse, Uri}; use rustfs_protos::{ ChannelClass, create_new_channel, get_channel_for_class, proto_gen::node_service::{ @@ -24,9 +33,19 @@ use rustfs_protos::{ tier_mutation_control_service_client::TierMutationControlServiceClient, }, }; -use std::{error::Error, io::ErrorKind}; +use std::{ + collections::HashMap, + error::Error, + future::Future, + io::ErrorKind, + pin::Pin, + sync::{LazyLock, Mutex}, + task::{Context, Poll}, +}; use tonic::{service::interceptor::InterceptedService, transport::Channel}; +use tower::Service; use tracing::debug; +use uuid::Uuid; use super::context_propagation::{inject_request_id_into_metadata, inject_trace_context_into_metadata}; @@ -35,7 +54,7 @@ use super::context_propagation::{inject_request_id_into_metadata, inject_trace_c pub async fn node_service_time_out_client( addr: &String, interceptor: TonicInterceptor, -) -> Result>, Box> { +) -> Result>, Box> { // Default to the latency-sensitive control channel; bulk `bytes` RPCs opt in via the // `_for_class` variant below (grpc-optimization P1). node_service_time_out_client_for_class(addr, interceptor, ChannelClass::Control).await @@ -44,13 +63,14 @@ pub async fn node_service_time_out_client( pub async fn heal_control_time_out_client( addr: &str, interceptor: TonicInterceptor, -) -> Result>, Box> { +) -> Result>, Box> { let interceptor = interceptor.with_rpc_audience(addr)?; let channel = match runtime_sources::cached_node_channel(addr).await { Some(channel) => channel, None => create_new_channel(addr).await?, }; let max_message_size = rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE; + let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience()); Ok(HealControlServiceClient::with_interceptor(channel, interceptor) .max_decoding_message_size(max_message_size) .max_encoding_message_size(max_message_size)) @@ -59,13 +79,14 @@ pub async fn heal_control_time_out_client( pub async fn tier_mutation_control_time_out_client( addr: &str, interceptor: TonicInterceptor, -) -> Result>, Box> { +) -> Result>, Box> { let interceptor = interceptor.with_rpc_audience(addr)?; let channel = match runtime_sources::cached_node_channel(addr).await { Some(channel) => channel, None => create_new_channel(addr).await?, }; let max_message_size = rustfs_protos::TIER_MUTATION_RPC_MAX_MESSAGE_SIZE; + let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience()); Ok(TierMutationControlServiceClient::with_interceptor(channel, interceptor) .max_decoding_message_size(max_message_size) .max_encoding_message_size(max_message_size)) @@ -81,7 +102,7 @@ pub async fn node_service_time_out_client_for_class( addr: &String, interceptor: TonicInterceptor, class: ChannelClass, -) -> Result>, Box> { +) -> Result>, Box> { let interceptor = interceptor.with_rpc_audience(addr)?; let channel = match class { ChannelClass::Control => match runtime_sources::cached_node_channel(addr).await { @@ -96,6 +117,7 @@ pub async fn node_service_time_out_client_for_class( }; let max_message_size = rustfs_protos::internode_rpc_max_message_size(); + let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience()); Ok(NodeServiceClient::with_interceptor(channel, interceptor) .max_decoding_message_size(max_message_size) .max_encoding_message_size(max_message_size)) @@ -103,7 +125,7 @@ pub async fn node_service_time_out_client_for_class( pub async fn node_service_time_out_client_no_auth( addr: &String, -) -> Result>, Box> { +) -> Result>, Box> { node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await } @@ -199,6 +221,104 @@ pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool { } } +/// The transport service that learns an authenticated peer boot epoch and adds the replay-scoped +/// signature only after one has been observed. The v1/v2 interceptor stays inside this wrapper so +/// old servers continue receiving precisely the metadata they understand. +#[derive(Clone, Debug)] +pub struct ReplayScopeChannel { + inner: S, + audience: Option, +} + +/// 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())); + +impl ReplayScopeChannel { + fn new(inner: S, audience: Option) -> Self { + Self { inner, audience } + } +} + +fn cached_peer_boot_epoch(audience: &str) -> Option { + PEER_BOOT_EPOCHS.lock().ok().and_then(|epochs| epochs.get(audience).copied()) +} + +fn remember_peer_boot_epoch(audience: String, epoch: Uuid) { + if let Ok(mut epochs) = PEER_BOOT_EPOCHS.lock() { + epochs.insert(audience, epoch); + } +} + +impl Service> for ReplayScopeChannel +where + S: Service, Response = HttpResponse>, + S::Error: Send + 'static, + S::Future: Send + 'static, + ReqBody: Send + 'static, + ResBody: Send + 'static, +{ + type Response = HttpResponse; + type Error = S::Error; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, mut request: HttpRequest) -> Self::Future { + let authenticated = self.audience.as_ref().is_some_and(|_| { + request + .headers() + .get(RPC_AUTH_VERSION_HEADER) + .and_then(|value| value.to_str().ok()) + == Some(RPC_AUTH_VERSION_V2) + }); + let challenge = authenticated.then(Uuid::new_v4); + 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. + request.headers_mut().insert( + RPC_BOOT_EPOCH_CHALLENGE_HEADER, + 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), + request.headers().get(TIMESTAMP_HEADER).and_then(|value| value.to_str().ok()), + request + .headers() + .get(RPC_CONTENT_SHA256_HEADER) + .and_then(|value| value.to_str().ok()), + ) { + match gen_tonic_replay_scope_headers(audience, request.uri().path(), timestamp, content_sha256, boot_epoch) { + Ok(headers) => request.headers_mut().extend(headers), + Err(error) => debug!(error = %error, "could not attach replay-scoped RPC signature"), + } + } + } + + let audience = self.audience.clone(); + let future = self.inner.call(request); + 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(_) => {} + } + } + Ok(response) + }) + } +} + pub struct TonicSignatureInterceptor { audience: Option, } @@ -257,6 +377,13 @@ impl TonicInterceptor { } Ok(self) } + + fn replay_scope_audience(&self) -> Option { + match self { + Self::Signature(interceptor) => interceptor.audience.clone(), + Self::NoOp(_) => None, + } + } } impl tonic::service::Interceptor for TonicInterceptor { @@ -279,6 +406,38 @@ mod tests { use tracing_opentelemetry::OpenTelemetrySpanExt; use tracing_subscriber::{Registry, layer::SubscriberExt}; + #[derive(Clone)] + struct EpochProofService { + audience: String, + seen_headers: std::sync::Arc>>, + } + + impl Service> for EpochProofService { + 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 { + self.seen_headers + .lock() + .expect("test header capture lock must not be poisoned") + .push(request.headers().clone()); + let challenge = tonic_boot_epoch_challenge(request.headers()) + .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"), + ); + std::future::ready(Ok(response)) + } + } + fn ensure_test_rpc_secret() { runtime_sources::ensure_test_rpc_secret(); } @@ -420,6 +579,52 @@ mod tests { assert_eq!(interceptor.audience.as_deref(), Some("node-a:9000")); } + #[test] + 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); + let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new())); + let service = EpochProofService { + audience: audience.to_string(), + 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 + }; + + 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"); + + let headers = seen_headers.lock().expect("test header capture lock must not be poisoned"); + assert_eq!(headers.len(), 2); + assert!(headers[0].contains_key(RPC_BOOT_EPOCH_CHALLENGE_HEADER)); + assert!( + !headers[0].contains_key(RPC_REPLAY_SCOPE_VERSION_HEADER), + "the first request must remain v2-compatible until the peer proves its epoch" + ); + assert!( + 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); + } + #[test] fn test_signature_interceptor_requires_generated_method_metadata() { ensure_test_rpc_secret(); diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 50342d720..b2e230b2c 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -20,8 +20,8 @@ //! rustfs/rustfs#4402) is anchored by the `ghsa_r5qv_*` tests in the module //! below, plus the broader negative-signature suite. The advisory class is: a //! node must never accept an RPC whose auth is missing, malformed, or signed -//! with the default/empty shared secret. Body-bound v2 requests additionally -//! receive process-local replay protection. See +//! with the default/empty shared secret. Body-bound v2 requests and all replay-scoped v3 +//! requests additionally receive process-local replay protection. See //! `docs/testing/security-regressions.md` for the full advisory -> test map. //! //! Advisory: @@ -50,13 +50,22 @@ use uuid::Uuid; type HmacSha256 = Hmac; const SIGNATURE_HEADER: &str = "x-rustfs-signature"; -const TIMESTAMP_HEADER: &str = "x-rustfs-timestamp"; -const RPC_AUTH_VERSION_HEADER: &str = "x-rustfs-rpc-auth-version"; +pub(crate) const TIMESTAMP_HEADER: &str = "x-rustfs-timestamp"; +pub(crate) const RPC_AUTH_VERSION_HEADER: &str = "x-rustfs-rpc-auth-version"; 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"; +pub(crate) const RPC_AUTH_VERSION_V2: &str = "2"; +pub const RPC_REPLAY_SCOPE_VERSION_HEADER: &str = "x-rustfs-rpc-replay-scope-version"; +pub const RPC_REPLAY_SCOPE_SIGNATURE_HEADER: &str = "x-rustfs-rpc-signature-v3"; +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"; +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 UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned"; const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes @@ -75,9 +84,15 @@ static INTERNODE_RPC_BODY_DIGEST_STRICT: LazyLock = LazyLock::new(|| { rustfs_config::DEFAULT_INTERNODE_RPC_BODY_DIGEST_STRICT, ) }); -// Sized for peak legitimate body-bound mutation RPS x the retention window; overflow fails closed -// and increments the replay-cache overflow counter. Clamped to at least 1 so a misconfigured zero -// cannot disable replay protection by rejecting every body-bound request. +static INTERNODE_RPC_REPLAY_SCOPE_STRICT: LazyLock = LazyLock::new(|| { + get_env_bool( + rustfs_config::ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT, + rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_SCOPE_STRICT, + ) +}); +// Sized for peak legitimate authenticated RPC RPS x the retention window once replay scope is +// active; overflow fails closed and increments the replay-cache overflow counter. Clamped to at +// least 1 so a misconfigured zero cannot disable replay protection by rejecting every request. static REPLAY_CACHE_CAPACITY: LazyLock = LazyLock::new(|| { rustfs_utils::get_env_usize( rustfs_config::ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, @@ -86,6 +101,7 @@ static REPLAY_CACHE_CAPACITY: LazyLock = LazyLock::new(|| { .max(1) }); static RPC_SECRET_RESOLUTION_LOG_ONCE: Once = Once::new(); +static RPC_BOOT_EPOCH: LazyLock = LazyLock::new(Uuid::new_v4); #[derive(Default)] struct RpcNonceCache { @@ -313,6 +329,189 @@ fn verify_signature_v2(secret: &str, scope: SignatureV2Scope<'_>, signature: &st mac.verify_slice(&signature).is_ok() } +#[derive(Clone, Copy)] +struct ReplayScope<'a> { + audience: &'a str, + path: &'a str, + timestamp: &'a str, + nonce: Uuid, + content_sha256: &'a str, + boot_epoch: Uuid, +} + +fn update_replay_scope(mac: &mut HmacSha256, scope: ReplayScope<'_>) { + mac.update(RPC_REPLAY_SCOPE_DOMAIN); + for part in [ + scope.audience.as_bytes(), + b"|", + scope.path.as_bytes(), + b"|POST|", + scope.timestamp.as_bytes(), + b"|", + scope.nonce.as_bytes(), + b"|", + scope.content_sha256.as_bytes(), + b"|", + scope.boot_epoch.as_bytes(), + ] { + mac.update(part); + } +} + +fn generate_replay_scope_signature(secret: &str, scope: ReplayScope<'_>) -> std::io::Result { + let mut mac = + ::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_replay_scope(&mut mac, scope); + Ok(general_purpose::STANDARD.encode(mac.finalize().into_bytes())) +} + +fn verify_replay_scope_signature(secret: &str, scope: ReplayScope<'_>, signature: &str) -> bool { + let Ok(signature) = general_purpose::STANDARD.decode(signature) else { + return false; + }; + let Ok(mut mac) = ::new_from_slice(secret.as_bytes()) else { + return false; + }; + update_replay_scope(&mut mac, scope); + mac.verify_slice(&signature).is_ok() +} + +fn update_boot_epoch_proof(mac: &mut HmacSha256, audience: &str, challenge: Uuid, boot_epoch: Uuid) { + mac.update(RPC_BOOT_EPOCH_PROOF_DOMAIN); + mac.update(audience.as_bytes()); + mac.update(b"|"); + mac.update(challenge.as_bytes()); + mac.update(boot_epoch.as_bytes()); +} + +fn generate_boot_epoch_proof(secret: &str, audience: &str, challenge: Uuid, boot_epoch: Uuid) -> std::io::Result { + if audience.is_empty() || challenge.is_nil() || boot_epoch.is_nil() { + return Err(std::io::Error::other("Invalid RPC boot epoch proof scope")); + } + let mut mac = + ::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_boot_epoch_proof(&mut mac, audience, challenge, boot_epoch); + Ok(general_purpose::STANDARD.encode(mac.finalize().into_bytes())) +} + +fn verify_boot_epoch_proof(secret: &str, audience: &str, challenge: Uuid, boot_epoch: Uuid, proof: &str) -> std::io::Result<()> { + if audience.is_empty() || challenge.is_nil() || boot_epoch.is_nil() { + return Err(std::io::Error::other("Invalid RPC boot epoch proof scope")); + } + let proof = general_purpose::STANDARD + .decode(proof) + .map_err(|_| std::io::Error::other("Invalid RPC boot epoch proof"))?; + let mut mac = + ::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_boot_epoch_proof(&mut mac, audience, challenge, boot_epoch); + mac.verify_slice(&proof) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid RPC boot epoch 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()) + .then_some(value) + .ok_or_else(|| std::io::Error::other(format!("Invalid {name}"))) +} + +fn parse_tonic_rpc_path(path: &str) -> std::io::Result<(&str, &str)> { + path.strip_prefix('/') + .and_then(|path| path.split_once('/')) + .filter(|(service, rpc_method)| !service.is_empty() && !rpc_method.is_empty() && !rpc_method.contains('/')) + .ok_or_else(|| std::io::Error::other("Invalid RPC request path")) +} + +/// The process-unique epoch included in every replay-scoped server verification. +/// +/// A fresh process gets a fresh value, so a signature captured before a server restart cannot be +/// admitted even though the bounded in-memory nonce cache necessarily starts empty again. +pub fn tonic_rpc_boot_epoch() -> Uuid { + *RPC_BOOT_EPOCH +} + +/// Build the additive replay-scope headers for a request that already carries rolling-upgrade-safe +/// v1/v2 metadata. `timestamp` and `content_sha256` are deliberately reused from the v2 scope so +/// old servers can continue validating the same request unchanged. +pub fn gen_tonic_replay_scope_headers( + audience: &str, + path: &str, + timestamp: &str, + content_sha256: &str, + boot_epoch: Uuid, +) -> std::io::Result { + if audience.is_empty() || !path.starts_with('/') || !valid_content_sha256(content_sha256) || boot_epoch.is_nil() { + return Err(std::io::Error::other("Invalid replay-scoped RPC signing scope")); + } + parse_tonic_rpc_path(path)?; + timestamp + .parse::() + .map_err(|_| std::io::Error::other("Invalid timestamp format"))?; + + let nonce = Uuid::new_v4(); + let signature = generate_replay_scope_signature( + &get_shared_secret()?, + ReplayScope { + audience, + path, + timestamp, + nonce, + content_sha256, + boot_epoch, + }, + )?; + let mut headers = HeaderMap::new(); + headers.insert(RPC_REPLAY_SCOPE_VERSION_HEADER, HeaderValue::from_static(RPC_REPLAY_SCOPE_VERSION_V3)); + headers.insert( + RPC_REPLAY_SCOPE_SIGNATURE_HEADER, + header_value(&signature, RPC_REPLAY_SCOPE_SIGNATURE_HEADER)?, + ); + headers.insert( + RPC_REPLAY_SCOPE_NONCE_HEADER, + header_value(&nonce.to_string(), RPC_REPLAY_SCOPE_NONCE_HEADER)?, + ); + headers.insert(RPC_BOOT_EPOCH_HEADER, header_value(&boot_epoch.to_string(), RPC_BOOT_EPOCH_HEADER)?); + Ok(headers) +} + +/// Parse the optional client challenge used to authenticate a server boot-epoch advertisement. +pub fn tonic_boot_epoch_challenge(headers: &HeaderMap) -> std::io::Result> { + headers + .get(RPC_BOOT_EPOCH_CHALLENGE_HEADER) + .map(|value| { + value + .to_str() + .map_err(|_| std::io::Error::other("Invalid RPC boot epoch challenge")) + .and_then(|value| non_nil_uuid(value, "RPC boot epoch challenge")) + }) + .transpose() +} + +/// Build the authenticated response headers for a client boot-epoch challenge. +pub fn tonic_boot_epoch_response_headers(audience: &str, challenge: Uuid) -> std::io::Result { + let boot_epoch = tonic_rpc_boot_epoch(); + let proof = generate_boot_epoch_proof(&get_shared_secret()?, audience, challenge, boot_epoch)?; + let mut headers = HeaderMap::new(); + headers.insert(RPC_BOOT_EPOCH_HEADER, header_value(&boot_epoch.to_string(), RPC_BOOT_EPOCH_HEADER)?); + headers.insert(RPC_BOOT_EPOCH_PROOF_HEADER, header_value(&proof, RPC_BOOT_EPOCH_PROOF_HEADER)?); + Ok(headers) +} + +/// Verify the server boot-epoch response for a challenge generated by this client. +pub fn verify_tonic_boot_epoch_response(audience: &str, challenge: Uuid, headers: &HeaderMap) -> std::io::Result { + let boot_epoch = headers + .get(RPC_BOOT_EPOCH_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing RPC boot epoch")) + .and_then(|value| non_nil_uuid(value, "RPC boot epoch"))?; + let proof = headers + .get(RPC_BOOT_EPOCH_PROOF_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing RPC boot epoch proof"))?; + verify_boot_epoch_proof(&get_shared_secret()?, audience, challenge, boot_epoch, proof)?; + Ok(boot_epoch) +} + fn valid_content_sha256(value: &str) -> bool { value == UNSIGNED_PAYLOAD || (value.len() == 64 @@ -531,6 +730,17 @@ fn has_v2_auth_headers(headers: &HeaderMap) -> bool { .any(|name| headers.contains_key(*name)) } +fn has_replay_scope_headers(headers: &HeaderMap) -> bool { + [ + RPC_REPLAY_SCOPE_VERSION_HEADER, + RPC_REPLAY_SCOPE_SIGNATURE_HEADER, + RPC_REPLAY_SCOPE_NONCE_HEADER, + RPC_BOOT_EPOCH_HEADER, + ] + .iter() + .any(|name| headers.contains_key(*name)) +} + /// Whether the server requires target-bound v2 authentication on every internode gRPC request, /// rejecting the legacy constant-target fallback instead of accepting it. Default-off rollout /// lever gated on the v1-fallback counter reading zero fleet-wide; see @@ -540,9 +750,127 @@ fn internode_rpc_signature_strict() -> bool { *INTERNODE_RPC_SIGNATURE_STRICT } +fn internode_rpc_replay_scope_strict() -> bool { + *INTERNODE_RPC_REPLAY_SCOPE_STRICT +} + +fn verify_tonic_replay_scope_signature(audience: &str, path: &str, headers: &HeaderMap) -> std::io::Result<()> { + if audience.is_empty() { + return Err(std::io::Error::other("Missing RPC audience")); + } + parse_tonic_rpc_path(path)?; + + let version = headers + .get(RPC_REPLAY_SCOPE_VERSION_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing RPC replay scope version"))?; + if version != RPC_REPLAY_SCOPE_VERSION_V3 { + return Err(std::io::Error::other("Unsupported RPC replay scope version")); + } + let signature = headers + .get(RPC_REPLAY_SCOPE_SIGNATURE_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing RPC replay scope signature"))?; + let timestamp = headers + .get(TIMESTAMP_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing timestamp header"))?; + let signed_at = timestamp + .parse::() + .map_err(|_| std::io::Error::other("Invalid timestamp format"))?; + check_timestamp(signed_at)?; + let nonce = headers + .get(RPC_REPLAY_SCOPE_NONCE_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing RPC replay scope nonce")) + .and_then(|value| non_nil_uuid(value, "RPC replay scope nonce"))?; + let content_sha256 = headers + .get(RPC_CONTENT_SHA256_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing RPC content SHA-256"))?; + if !valid_content_sha256(content_sha256) { + return Err(std::io::Error::other("Invalid RPC content SHA-256")); + } + let boot_epoch = headers + .get(RPC_BOOT_EPOCH_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing RPC boot epoch")) + .and_then(|value| non_nil_uuid(value, "RPC boot epoch"))?; + let secret = get_shared_secret()?; + if !verify_replay_scope_signature( + &secret, + ReplayScope { + audience, + path, + timestamp, + nonce, + content_sha256, + boot_epoch, + }, + signature, + ) { + return Err(std::io::Error::other("Invalid RPC replay scope signature")); + } + if boot_epoch != tonic_rpc_boot_epoch() { + return Err(std::io::Error::other("RPC boot epoch is stale")); + } + check_and_record_nonce(nonce, signed_at) +} + /// Verify gRPC authentication, preferring v2 without downgrade on malformed v2 metadata. pub fn verify_tonic_rpc_signature(audience: &str, path: &str, headers: &HeaderMap) -> std::io::Result<()> { - verify_tonic_rpc_signature_with_strictness(audience, path, headers, internode_rpc_signature_strict()) + verify_tonic_rpc_signature_with_policy( + audience, + path, + headers, + internode_rpc_signature_strict(), + internode_rpc_replay_scope_strict(), + false, + ) +} + +/// Verify gRPC authentication while allowing the narrowly scoped v2 `Ping` bootstrap used to +/// obtain an authenticated server boot epoch when replay-scope strictness is enabled. +pub fn verify_tonic_rpc_signature_with_bootstrap( + audience: &str, + path: &str, + headers: &HeaderMap, + allow_replay_scope_bootstrap: bool, +) -> std::io::Result<()> { + verify_tonic_rpc_signature_with_policy( + audience, + path, + headers, + internode_rpc_signature_strict(), + internode_rpc_replay_scope_strict(), + allow_replay_scope_bootstrap, + ) +} + +fn verify_tonic_rpc_signature_with_policy( + audience: &str, + path: &str, + headers: &HeaderMap, + signature_strict: bool, + replay_scope_strict: bool, + allow_replay_scope_bootstrap: bool, +) -> std::io::Result<()> { + if has_replay_scope_headers(headers) { + return verify_tonic_replay_scope_signature(audience, path, headers); + } + + // Only a method-bound v2 Ping with a syntactically valid challenge may bootstrap a strict + // client after its peer restarts. Legacy metadata never gets this exception. + let bootstrap = allow_replay_scope_bootstrap + && has_v2_auth_headers(headers) + && tonic_boot_epoch_challenge(headers).is_ok_and(|challenge| challenge.is_some()); + if replay_scope_strict && !bootstrap { + return Err(std::io::Error::other("RPC replay-scoped authentication required")); + } + + verify_tonic_rpc_signature_with_strictness(audience, path, headers, signature_strict)?; + global_internode_metrics().record_replay_scope_fallback(); + Ok(()) } /// [`verify_tonic_rpc_signature`] with the strict gate injected as a parameter, so both rollout @@ -1273,6 +1601,104 @@ mod tests { assert_eq!(error.to_string(), "Invalid RPC v2 signature"); } + #[test] + fn replay_scope_binds_path_epoch_and_random_nonce() { + ensure_test_rpc_secret(); + let path = "/node_service.NodeService/Ping"; + let mut headers = gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "Ping", None) + .expect("v2 compatibility headers should build"); + let timestamp = headers + .get(TIMESTAMP_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("v2 timestamp") + .to_string(); + let content_sha256 = headers + .get(RPC_CONTENT_SHA256_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("v2 content digest") + .to_string(); + headers.extend( + gen_tonic_replay_scope_headers("node-a:9000", path, ×tamp, &content_sha256, tonic_rpc_boot_epoch()) + .expect("replay-scope headers should build"), + ); + + assert!( + verify_tonic_rpc_signature_with_policy("node-a:9000", path, &headers, false, false, false).is_ok(), + "the first replay-scoped request must be accepted" + ); + let replay = verify_tonic_rpc_signature_with_policy("node-a:9000", path, &headers, false, false, false) + .expect_err("the random replay-scope nonce must be single-use"); + assert_eq!(replay.to_string(), "RPC request replay detected"); + + let path_error = verify_tonic_replay_scope_signature("node-a:9000", "/node_service.NodeService/SignalService", &headers) + .expect_err("a replay-scoped signature must not move to another method"); + assert_eq!(path_error.to_string(), "Invalid RPC replay scope signature"); + } + + #[test] + fn replay_scope_rejects_partial_metadata_and_stale_epoch_without_fallback() { + ensure_test_rpc_secret(); + let path = "/node_service.NodeService/Ping"; + let mut partial = gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "Ping", None) + .expect("v2 compatibility headers should build"); + partial.insert(RPC_REPLAY_SCOPE_VERSION_HEADER, HeaderValue::from_static(RPC_REPLAY_SCOPE_VERSION_V3)); + let error = verify_tonic_rpc_signature_with_policy("node-a:9000", path, &partial, false, false, false) + .expect_err("partial replay-scope metadata must never downgrade to v2"); + assert_eq!(error.to_string(), "Missing RPC replay scope signature"); + + let timestamp = partial + .get(TIMESTAMP_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("v2 timestamp") + .to_string(); + let content_sha256 = partial + .get(RPC_CONTENT_SHA256_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("v2 content digest") + .to_string(); + let stale_epoch = Uuid::new_v4(); + partial.extend( + gen_tonic_replay_scope_headers("node-a:9000", path, ×tamp, &content_sha256, stale_epoch) + .expect("replay-scope headers should build"), + ); + let stale = verify_tonic_rpc_signature_with_policy("node-a:9000", path, &partial, false, false, false) + .expect_err("a signature from a prior server boot epoch must be rejected"); + assert_eq!(stale.to_string(), "RPC boot epoch is stale"); + } + + #[test] + fn replay_scope_strictness_allows_only_authenticated_ping_bootstrap() { + ensure_test_rpc_secret(); + let mut headers = gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "Ping", None) + .expect("v2 compatibility headers should build"); + let rejected = + verify_tonic_rpc_signature_with_policy("node-a:9000", "/node_service.NodeService/Ping", &headers, false, true, false) + .expect_err("strict replay scope must reject stripped new metadata"); + assert_eq!(rejected.to_string(), "RPC replay-scoped authentication required"); + + headers.insert( + RPC_BOOT_EPOCH_CHALLENGE_HEADER, + HeaderValue::from_str(&Uuid::new_v4().to_string()).expect("UUID header"), + ); + assert!( + verify_tonic_rpc_signature_with_policy("node-a:9000", "/node_service.NodeService/Ping", &headers, false, true, true,) + .is_ok(), + "only the signed Ping bootstrap may obtain a new server epoch in strict mode" + ); + } + + #[test] + fn boot_epoch_response_proof_binds_audience_challenge_and_epoch() { + ensure_test_rpc_secret(); + let challenge = Uuid::new_v4(); + let headers = tonic_boot_epoch_response_headers("node-a:9000", challenge).expect("proof headers should build"); + let epoch = + verify_tonic_boot_epoch_response("node-a:9000", challenge, &headers).expect("matching proof headers should verify"); + assert_eq!(epoch, tonic_rpc_boot_epoch()); + assert!(verify_tonic_boot_epoch_response("node-b:9000", challenge, &headers).is_err()); + assert!(verify_tonic_boot_epoch_response("node-a:9000", Uuid::new_v4(), &headers).is_err()); + } + #[test] fn malformed_v2_auth_does_not_downgrade_to_valid_legacy_signature() { ensure_test_rpc_secret(); diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index 74d5bfbe8..0f5ae5c48 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -28,13 +28,18 @@ pub(crate) use background_monitor::pin_callsite_interest_for_test; pub use background_monitor::shutdown_background_monitors; pub(crate) use background_monitor::spawn_background_monitor; pub use client::{ - TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth, + AuthenticatedChannel, TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client, + node_service_time_out_client_no_auth, }; +// Re-exported through `api::rpc`; not every item is consumed inside this crate. +#[allow(unused_imports)] 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, set_tonic_mutation_body_digest, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, - verify_ns_scanner_capability, verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, - verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, + TONIC_RPC_PREFIX, build_auth_headers, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, + normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, sign_ns_scanner_capability, + sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, verify_ns_scanner_capability, + 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, }; #[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 3e47bd349..741bc628d 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::cluster::rpc::client::{ - TonicInterceptor, embedded_tonic_status, gen_tonic_signature_interceptor, heal_control_time_out_client, + AuthenticatedChannel, TonicInterceptor, embedded_tonic_status, gen_tonic_signature_interceptor, heal_control_time_out_client, is_network_like_status, message_has_network_needle, node_service_time_out_client, tier_mutation_control_time_out_client, }; use crate::cluster::rpc::{set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, verify_tonic_rpc_response_proof}; @@ -66,7 +66,6 @@ use std::{ use tokio::{net::TcpStream, time::Duration}; use tonic::Request; use tonic::service::interceptor::InterceptedService; -use tonic::transport::Channel; use tracing::{debug, info, warn}; use uuid::Uuid; @@ -412,7 +411,7 @@ impl PeerRestClient { (remote, all, remote_topology_hosts) } - pub async fn get_client(&self) -> Result>> { + pub async fn get_client(&self) -> Result>> { if self.offline.load(Ordering::Acquire) { self.mark_offline_and_spawn_recovery(); return Err(Error::other(format!("peer {} is temporarily offline", self.grid_host))); @@ -433,7 +432,7 @@ impl PeerRestClient { &self, ) -> Result< rustfs_protos::proto_gen::node_service::heal_control_service_client::HealControlServiceClient< - InterceptedService, + InterceptedService, >, > { if self.offline.load(Ordering::Acquire) { @@ -454,7 +453,7 @@ impl PeerRestClient { async fn get_tier_mutation_control_client( &self, - ) -> Result>> { + ) -> Result>> { if self.offline.load(Ordering::Acquire) { self.mark_offline_and_spawn_recovery(); return Err(Error::other(format!("peer {} is temporarily offline", self.grid_host))); diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 2c4a12474..28d311146 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -14,7 +14,8 @@ use crate::bucket::metadata_sys; use crate::cluster::rpc::client::{ - TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client, + AuthenticatedChannel, TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, + node_service_time_out_client, }; use crate::cluster::rpc::set_tonic_mutation_body_digest; use crate::disk::error::DiskError; @@ -52,7 +53,6 @@ use tokio::{net::TcpStream, sync::RwLock, time}; use tokio_util::sync::CancellationToken; use tonic::Request; use tonic::service::interceptor::InterceptedService; -use tonic::transport::Channel; use tracing::{debug, info, warn}; type Client = Arc>; @@ -832,7 +832,7 @@ impl RemotePeerS3Client { client } - pub async fn get_client(&self) -> Result>> { + pub async fn get_client(&self) -> Result>> { node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) .await .map_err(|err| Error::other(format!("can not get client, err: {err}"))) diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 5151275d2..3a0926d0b 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -13,8 +13,8 @@ // limitations under the License. use crate::cluster::rpc::client::{ - TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client, - node_service_time_out_client_for_class, node_service_time_out_client_no_auth, + AuthenticatedChannel, TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, + node_service_time_out_client, node_service_time_out_client_for_class, node_service_time_out_client_no_auth, }; use crate::cluster::rpc::http_auth::set_tonic_canonical_body_digest; use crate::cluster::rpc::internode_data_transport::{ @@ -71,7 +71,7 @@ use tokio::{ time::timeout, }; use tokio_util::sync::CancellationToken; -use tonic::{Code, Request, service::interceptor::InterceptedService, transport::Channel}; +use tonic::{Code, Request, service::interceptor::InterceptedService}; use tracing::{debug, trace, warn}; use uuid::Uuid; @@ -1083,7 +1083,7 @@ impl RemoteDisk { internode_offline_bypass_reason(&self.addr).map(Error::other) } - async fn get_client(&self) -> Result>> { + async fn get_client(&self) -> Result>> { if let Some(err) = self.offline_bypass_error() { return Err(err); } @@ -1096,7 +1096,7 @@ impl RemoteDisk { /// Routes onto the isolated bulk channel pool so large transfers cannot head-of-line block /// lock/health RPCs (grpc-optimization P1). Falls back to the control channel when isolation /// is disabled. - async fn get_bulk_client(&self) -> Result>> { + async fn get_bulk_client(&self) -> Result>> { if let Some(err) = self.offline_bypass_error() { return Err(err); } diff --git a/crates/ecstore/src/cluster/rpc/remote_locker.rs b/crates/ecstore/src/cluster/rpc/remote_locker.rs index 9707d6c8e..05a8759bb 100644 --- a/crates/ecstore/src/cluster/rpc/remote_locker.rs +++ b/crates/ecstore/src/cluster/rpc/remote_locker.rs @@ -12,7 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::cluster::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client}; +use crate::cluster::rpc::client::{ + AuthenticatedChannel, TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client, +}; use crate::cluster::rpc::set_tonic_mutation_body_digest; use async_trait::async_trait; use bytes::Bytes; @@ -29,7 +31,6 @@ use std::time::Duration; use tokio::time::timeout; use tonic::Request; use tonic::service::interceptor::InterceptedService; -use tonic::transport::Channel; use tracing::{debug, info, warn}; /// Remote lock client implementation @@ -78,7 +79,7 @@ impl RemoteClient { } } - pub async fn get_client(&self) -> Result>> { + pub async fn get_client(&self) -> Result>> { // P3-2 offline bypass (now covering the lock path too): fast-fail a peer already marked // offline instead of paying the connect timeout, so dsync reaches quorum sooner. Does not // change quorum; the self-healing re-probe keeps the peer recoverable. diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 0790c2c8b..0aa69e6fb 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -67,6 +67,7 @@ const INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL: &str = "rustfs_system_network_inter const INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_decode_error_total"; const INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_signature_v1_fallback_total"; const INTERNODE_BODY_DIGEST_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_body_digest_fallback_total"; +const INTERNODE_REPLAY_SCOPE_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_replay_scope_fallback_total"; const INTERNODE_REPLAY_CACHE_OVERFLOW_TOTAL: &str = "rustfs_system_network_internode_replay_cache_overflow_total"; const ERASURE_WRITE_QUORUM_FAILURES_TOTAL: &str = "rustfs_system_storage_erasure_write_quorum_failures_total"; @@ -159,6 +160,7 @@ pub struct InternodeMetricsSnapshot { pub operation_write_shutdown_errors_total: u64, pub signature_v1_fallback_total: u64, pub body_digest_fallback_total: u64, + pub replay_scope_fallback_total: u64, pub replay_cache_overflow_total: u64, } @@ -180,6 +182,7 @@ pub struct InternodeMetrics { msgpack_json_decode_error_total: AtomicU64, signature_v1_fallback_total: AtomicU64, body_digest_fallback_total: AtomicU64, + replay_scope_fallback_total: AtomicU64, replay_cache_overflow_total: AtomicU64, } @@ -431,6 +434,13 @@ impl InternodeMetrics { counter!(INTERNODE_BODY_DIGEST_FALLBACK_TOTAL).increment(1); } + /// Count an accepted v1/v2 request that does not carry the replay-scoped signature. This is + /// the convergence signal for `RUSTFS_INTERNODE_RPC_REPLAY_SCOPE_STRICT`. + pub fn record_replay_scope_fallback(&self) { + self.replay_scope_fallback_total.fetch_add(1, Ordering::Relaxed); + counter!(INTERNODE_REPLAY_SCOPE_FALLBACK_TOTAL).increment(1); + } + /// Count a body-bound internode RPC rejected because the replay-protection nonce cache was /// full. Overflow fails closed, so a sustained non-zero rate means /// `RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY` is undersized for this node's peak legitimate @@ -488,6 +498,7 @@ impl InternodeMetrics { operation_write_shutdown_errors_total: self.operation_write_shutdown_errors_total.load(Ordering::Relaxed), signature_v1_fallback_total: self.signature_v1_fallback_total.load(Ordering::Relaxed), body_digest_fallback_total: self.body_digest_fallback_total.load(Ordering::Relaxed), + replay_scope_fallback_total: self.replay_scope_fallback_total.load(Ordering::Relaxed), replay_cache_overflow_total: self.replay_cache_overflow_total.load(Ordering::Relaxed), } } @@ -510,6 +521,7 @@ impl InternodeMetrics { self.msgpack_json_decode_error_total.store(0, Ordering::Relaxed); self.signature_v1_fallback_total.store(0, Ordering::Relaxed); self.body_digest_fallback_total.store(0, Ordering::Relaxed); + self.replay_scope_fallback_total.store(0, Ordering::Relaxed); self.replay_cache_overflow_total.store(0, Ordering::Relaxed); } } @@ -887,6 +899,19 @@ mod tests { assert_eq!(metrics.snapshot().signature_v1_fallback_total, 0); } + #[test] + fn replay_scope_fallback_counter_updates_snapshot_and_resets() { + let metrics = InternodeMetrics::default(); + assert_eq!(metrics.snapshot().replay_scope_fallback_total, 0); + + metrics.record_replay_scope_fallback(); + metrics.record_replay_scope_fallback(); + assert_eq!(metrics.snapshot().replay_scope_fallback_total, 2); + + metrics.reset_for_test(); + assert_eq!(metrics.snapshot().replay_scope_fallback_total, 0); + } + #[test] fn cluster_peer_flips_offline_after_threshold_and_back_online() { // Unique addr keeps this independent of the process-global registry / other tests. diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index fc8e7d712..8656ca486 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -37,7 +37,8 @@ use crate::storage_api::server::http as storage; use crate::storage_api::server::http::rpc::InternodeRpcService; 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, + ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, tonic_boot_epoch_challenge, + tonic_boot_epoch_response_headers, verify_tonic_rpc_signature_with_bootstrap, }; use bytes::Bytes; use http::{HeaderMap, Method, Request as HttpRequest, Response, Uri}; @@ -152,13 +153,17 @@ impl RpcRequestPathService { } } -impl Service> for RpcRequestPathService +impl Service> for RpcRequestPathService where - S: Service>, + S: Service, Response = Response>, + S::Error: Send + 'static, + S::Future: Send + 'static, + B: Send + 'static, + ResBody: Send + 'static, { - type Response = S::Response; + type Response = Response; type Error = S::Error; - type Future = S::Future; + type Future = Pin> + Send>>; fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { self.inner.poll_ready(cx) @@ -170,7 +175,22 @@ where method: req.method().clone(), }; req.extensions_mut().insert(target); - self.inner.call(req) + let response_headers = tonic_boot_epoch_challenge(req.headers()) + .ok() + .flatten() + .and_then(|challenge| { + storage::try_current_local_node_name() + .and_then(|node| normalize_tonic_rpc_audience(&node).ok()) + .and_then(|audience| tonic_boot_epoch_response_headers(&audience, challenge).ok()) + }); + let future = self.inner.call(req); + Box::pin(async move { + let mut response = future.await?; + if let Some(headers) = response_headers { + response.headers_mut().extend(headers); + } + Ok(response) + }) } } @@ -1862,7 +1882,15 @@ fn check_auth(req: Request<()>) -> std::result::Result, Status> { .filter(|method| !method.is_empty() && !method.contains('/')) .ok_or_else(|| Status::unauthenticated("Invalid RPC request path"))?; debug_assert!(!rpc_method.is_empty()); - verify_tonic_rpc_signature(&audience, target.uri.path(), req.metadata().as_ref()).map_err(|e| { + let allow_replay_scope_bootstrap = target.uri.path() == "/node_service.NodeService/Ping" + && tonic_boot_epoch_challenge(req.metadata().as_ref()).is_ok_and(|challenge| challenge.is_some()); + verify_tonic_rpc_signature_with_bootstrap( + &audience, + target.uri.path(), + req.metadata().as_ref(), + allow_replay_scope_bootstrap, + ) + .map_err(|e| { error!( event = EVENT_RPC_SIGNATURE_VERIFICATION_FAILED, component = LOG_COMPONENT_SERVER, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 65427289f..5cbf1554f 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -495,8 +495,9 @@ 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, verify_rpc_signature, verify_tonic_canonical_body_digest, - verify_tonic_mutation_body_digest, verify_tonic_rpc_signature, + 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::{ @@ -1605,8 +1606,21 @@ pub(crate) fn sign_ns_scanner_capability(challenge: uuid::Uuid, server_epoch: uu ecstore_rpc::sign_ns_scanner_capability(challenge, server_epoch) } -pub(crate) fn verify_tonic_rpc_signature(audience: &str, path: &str, headers: &http::HeaderMap) -> std::io::Result<()> { - ecstore_rpc::verify_tonic_rpc_signature(audience, path, headers) +pub(crate) fn verify_tonic_rpc_signature_with_bootstrap( + audience: &str, + path: &str, + headers: &http::HeaderMap, + allow_replay_scope_bootstrap: bool, +) -> std::io::Result<()> { + ecstore_rpc::verify_tonic_rpc_signature_with_bootstrap(audience, path, headers, allow_replay_scope_bootstrap) +} + +pub(crate) fn tonic_boot_epoch_challenge(headers: &http::HeaderMap) -> std::io::Result> { + ecstore_rpc::tonic_boot_epoch_challenge(headers) +} + +pub(crate) fn tonic_boot_epoch_response_headers(audience: &str, challenge: uuid::Uuid) -> std::io::Result { + ecstore_rpc::tonic_boot_epoch_response_headers(audience, challenge) } pub(crate) fn verify_tonic_canonical_body_digest(request: &tonic::Request, canonical_body: &[u8]) -> std::io::Result<()> { diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index b10ead5f2..3953874c5 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -97,7 +97,8 @@ pub(crate) mod server { pub(crate) mod http { pub(crate) use crate::storage::storage_api::{ - ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, verify_tonic_rpc_signature, + ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, tonic_boot_epoch_challenge, + tonic_boot_epoch_response_headers, verify_tonic_rpc_signature_with_bootstrap, }; pub(crate) fn try_current_local_node_name() -> Option {