From d48870df9771ff27856afbdd9db43c107e971647 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 29 Jul 2026 17:17:30 +0800 Subject: [PATCH] fix(rpc): authenticate non-disk mutation bodies (#5425) --- crates/ecstore/src/cluster/rpc/http_auth.rs | 97 ++++- crates/ecstore/src/cluster/rpc/mod.rs | 6 +- .../src/cluster/rpc/peer_rest_client.rs | 63 +-- .../ecstore/src/cluster/rpc/peer_s3_client.rs | 10 +- .../ecstore/src/cluster/rpc/remote_locker.rs | 25 +- crates/protos/src/lib.rs | 362 ++++++++++++++++++ crates/protos/src/node.proto | 176 ++++----- rustfs/src/storage/rpc/node_service.rs | 321 ++++++++++++++-- 8 files changed, 900 insertions(+), 160 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 11aeb6efa..50342d720 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -443,6 +443,16 @@ pub fn set_tonic_canonical_body_digest(request: &mut tonic::Request, canon Ok(()) } +pub fn set_tonic_mutation_body_digest( + request: &mut tonic::Request, +) -> std::io::Result<()> { + let canonical_body = request + .get_ref() + .canonical_body() + .map_err(|_| std::io::Error::other("RPC mutation body length cannot be represented"))?; + set_tonic_canonical_body_digest(request, &canonical_body) +} + pub fn verify_tonic_canonical_body_digest(request: &tonic::Request, canonical_body: &[u8]) -> std::io::Result<()> { let version = request .metadata() @@ -466,7 +476,7 @@ pub fn verify_tonic_canonical_body_digest(request: &tonic::Request, canoni Ok(()) } -/// Verify a mutating disk RPC's canonical body digest with a rolling-upgrade fallback. +/// Verify a mutating RPC's canonical body digest with a rolling-upgrade fallback. /// /// When the request carries a real (non-`UNSIGNED-PAYLOAD`) content SHA-256 it is verified exactly /// like [`verify_tonic_canonical_body_digest`]. The digest value is a member of the signed v2 @@ -497,7 +507,7 @@ fn verify_tonic_mutation_body_digest_with_strictness( Some(digest) if digest != UNSIGNED_PAYLOAD => verify_tonic_canonical_body_digest(request, canonical_body), _ => { // RUSTFS_COMPAT_TODO(disk-mutation-body-digest): accept digestless peers during rolling upgrades. Remove after the - // minimum supported RustFS peer version body-binds every mutating disk RPC. + // minimum supported RustFS peer version body-binds every mutating RPC. if strict { return Err(std::io::Error::other("RPC mutation requires a body-bound v2 signature")); } @@ -677,11 +687,28 @@ mod tests { use crate::cluster::rpc::context_propagation::REQUEST_ID_HEADER; use crate::runtime::sources as runtime_sources; use http::{HeaderMap, Method}; + use rustfs_protos::{ + CanonicalMutationBody as _, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, + proto_gen::node_service::{Mss, SignalServiceRequest}, + }; + use std::collections::HashMap; use std::io::{self, Write}; use std::sync::{Arc, Mutex}; use time::OffsetDateTime; use tracing_subscriber::fmt::MakeWriter; + fn signal_service_request(signal: &str, sub_system: &str, dry_run: &str) -> SignalServiceRequest { + SignalServiceRequest { + vars: Some(Mss { + value: HashMap::from([ + (PEER_RESTSIGNAL.to_string(), signal.to_string()), + (PEER_RESTSUB_SYS.to_string(), sub_system.to_string()), + (PEER_RESTDRY_RUN.to_string(), dry_run.to_string()), + ]), + }), + } + } + #[derive(Clone, Default)] struct CapturedLogs { buffer: Arc>>, @@ -1596,6 +1623,72 @@ mod tests { assert_eq!(stripped.to_string(), "RPC content SHA-256 mismatch"); } + #[test] + fn signal_service_mutation_contract_rejects_tampering_and_replay() { + ensure_test_rpc_secret(); + let body = signal_service_request("2", "scanner", "false") + .canonical_body() + .expect("small signal request should encode"); + let mut request = tonic::Request::new(()); + set_tonic_canonical_body_digest(&mut request, &body).expect("canonical body digest should be attached"); + let content_sha256 = request + .metadata() + .get(RPC_CONTENT_SHA256_HEADER) + .and_then(|value| value.to_str().ok()); + let headers = gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "SignalService", content_sha256) + .expect("body-bound auth headers should build"); + request.metadata_mut().as_mut().extend(headers.clone()); + + assert!( + verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/SignalService", &headers).is_ok(), + "the first body-bound signal request must authenticate" + ); + assert!(verify_tonic_mutation_body_digest(&request, &body).is_ok()); + + let tampered = signal_service_request("1", "scanner", "false") + .canonical_body() + .expect("small signal request should encode"); + let error = verify_tonic_mutation_body_digest(&request, &tampered) + .expect_err("changing the signal must invalidate the signed digest"); + assert_eq!(error.to_string(), "RPC content SHA-256 mismatch"); + + let replay = verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/SignalService", &headers) + .expect_err("reusing the signal nonce must fail"); + assert_eq!(replay.to_string(), "RPC request replay detected"); + } + + #[test] + #[serial_test::serial(rpc_body_digest_fallback_counter)] + fn signal_service_mutation_contract_preserves_rollout_fallback_and_strictness() { + let body = signal_service_request("2", "scanner", "false") + .canonical_body() + .expect("small signal request should encode"); + let before = global_internode_metrics().snapshot().body_digest_fallback_total; + let digestless = tonic::Request::new(()); + + assert!( + verify_tonic_mutation_body_digest_with_strictness(&digestless, &body, false).is_ok(), + "old peers must remain compatible while the rollout gate is open" + ); + assert_eq!( + global_internode_metrics().snapshot().body_digest_fallback_total, + before + 1, + "accepted digestless signal requests must be visible in the fallback metric" + ); + + let error = verify_tonic_mutation_body_digest_with_strictness(&digestless, &body, true) + .expect_err("strict mode must reject a digestless signal request"); + assert_eq!(error.to_string(), "RPC mutation requires a body-bound v2 signature"); + + let mut bound = tonic::Request::new(()); + set_tonic_canonical_body_digest(&mut bound, &body).expect("canonical body digest should be attached"); + bound + .metadata_mut() + .as_mut() + .insert(RPC_AUTH_VERSION_HEADER, HeaderValue::from_static(RPC_AUTH_VERSION_V2)); + assert!(verify_tonic_mutation_body_digest_with_strictness(&bound, &body, true).is_ok()); + } + #[test] fn nonce_cache_rejects_replay_after_wall_clock_regression() { let now = Instant::now(); diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index 10e158275..637fb5849 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -30,9 +30,9 @@ pub use client::{ }; pub use http_auth::{ TONIC_RPC_PREFIX, build_auth_headers, gen_signature_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience, - set_tonic_canonical_body_digest, 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, + 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, }; #[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 ab7a62fc8..203e6064d 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -16,7 +16,7 @@ use crate::cluster::rpc::client::{ 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, verify_tonic_rpc_response_proof}; +use crate::cluster::rpc::{set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, verify_tonic_rpc_response_proof}; use crate::error::{Error, Result}; use crate::storage_api_contracts::internode::{ SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION, @@ -50,6 +50,7 @@ use rustfs_protos::proto_gen::node_service::{ TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient, tier_mutation_control_service_client::TierMutationControlServiceClient, }; +pub use rustfs_protos::{PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS}; use rustfs_protos::{TierMutationRpcPhase, evict_failed_connection}; use rustfs_utils::XHost; use serde::{Deserialize, Serialize as _}; @@ -69,9 +70,6 @@ use tonic::transport::Channel; use tracing::{debug, info, warn}; use uuid::Uuid; -pub const PEER_RESTSIGNAL: &str = "signal"; -pub const PEER_RESTSUB_SYS: &str = "sub-sys"; -pub const PEER_RESTDRY_RUN: &str = "dry-run"; pub const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = 1; pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2; const BACKGROUND_HEAL_STATUS_MAX_MESSAGE_SIZE: usize = 64 * 1024; @@ -1160,10 +1158,11 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(LoadBucketMetadataRequest { + let mut request = Request::new(LoadBucketMetadataRequest { bucket: bucket.to_string(), scanner_maintenance_change, }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.load_bucket_metadata(request).await?.into_inner(); if !response.success { @@ -1183,9 +1182,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(DeleteBucketMetadataRequest { + let mut request = Request::new(DeleteBucketMetadataRequest { bucket: bucket.to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.delete_bucket_metadata(request).await?.into_inner(); if !response.success { @@ -1205,9 +1205,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(DeletePolicyRequest { + let mut request = Request::new(DeletePolicyRequest { policy_name: policy.to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.delete_policy(request).await?.into_inner(); if !response.success { @@ -1227,9 +1228,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(LoadPolicyRequest { + let mut request = Request::new(LoadPolicyRequest { policy_name: policy.to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.load_policy(request).await?.into_inner(); if !response.success { @@ -1249,11 +1251,12 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(LoadPolicyMappingRequest { + let mut request = Request::new(LoadPolicyMappingRequest { user_or_group: user_or_group.to_string(), user_type, is_group, }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.load_policy_mapping(request).await?.into_inner(); if !response.success { @@ -1273,9 +1276,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(DeleteUserRequest { + let mut request = Request::new(DeleteUserRequest { access_key: access_key.to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.delete_user(request).await?.into_inner(); if !response.success { @@ -1295,9 +1299,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(DeleteServiceAccountRequest { + let mut request = Request::new(DeleteServiceAccountRequest { access_key: access_key.to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.delete_service_account(request).await?.into_inner(); if !response.success { @@ -1317,10 +1322,11 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(LoadUserRequest { + let mut request = Request::new(LoadUserRequest { access_key: access_key.to_string(), temp, }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.load_user(request).await?.into_inner(); if !response.success { @@ -1340,9 +1346,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(LoadServiceAccountRequest { + let mut request = Request::new(LoadServiceAccountRequest { access_key: access_key.to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.load_service_account(request).await?.into_inner(); if !response.success { @@ -1362,9 +1369,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(LoadGroupRequest { + let mut request = Request::new(LoadGroupRequest { group: group.to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.load_group(request).await?.into_inner(); if !response.success { @@ -1384,7 +1392,8 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(ReloadSiteReplicationConfigRequest {}); + let mut request = Request::new(ReloadSiteReplicationConfigRequest {}); + set_tonic_mutation_body_digest(&mut request)?; let response = client.reload_site_replication_config(request).await?.into_inner(); if !response.success { @@ -1408,9 +1417,10 @@ impl PeerRestClient { vars.insert(PEER_RESTSIGNAL.to_string(), sig.to_string()); vars.insert(PEER_RESTSUB_SYS.to_string(), sub_sys.to_string()); vars.insert(PEER_RESTDRY_RUN.to_string(), dry_run.to_string()); - let request = Request::new(SignalServiceRequest { + let mut request = Request::new(SignalServiceRequest { vars: Some(Mss { value: vars }), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.signal_service(request).await?.into_inner(); if !response.success { @@ -1479,7 +1489,8 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(ReloadPoolMetaRequest {}); + let mut request = Request::new(ReloadPoolMetaRequest {}); + set_tonic_mutation_body_digest(&mut request)?; let response = client.reload_pool_meta(request).await?.into_inner(); if !response.success { @@ -1500,9 +1511,10 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(StopRebalanceRequest { + let mut request = Request::new(StopRebalanceRequest { expected_rebalance_id: expected_rebalance_id.unwrap_or_default().to_string(), }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.stop_rebalance(request).await?.into_inner(); if !response.success { @@ -1523,7 +1535,8 @@ impl PeerRestClient { self.finalize_result( async { let mut client = self.get_client().await?; - let request = Request::new(LoadRebalanceMetaRequest { start_rebalance }); + let mut request = Request::new(LoadRebalanceMetaRequest { start_rebalance }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.load_rebalance_meta(request).await?.into_inner(); @@ -1562,7 +1575,8 @@ impl PeerRestClient { }) .collect::>>()?; let mut client = self.get_client().await?; - let request = Request::new(StartDecommissionRequest { pool_indices }); + let mut request = Request::new(StartDecommissionRequest { pool_indices }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.start_decommission(request).await?.into_inner(); if !response.success { @@ -1585,7 +1599,8 @@ impl PeerRestClient { let pool_index = u32::try_from(pool_index) .map_err(|_| Error::other(format!("decommission pool index {pool_index} exceeds RPC range")))?; let mut client = self.get_client().await?; - let request = Request::new(CancelDecommissionRequest { pool_index }); + let mut request = Request::new(CancelDecommissionRequest { pool_index }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.cancel_decommission(request).await?.into_inner(); if !response.success { @@ -1608,7 +1623,8 @@ impl PeerRestClient { let pool_index = u32::try_from(pool_index) .map_err(|_| Error::other(format!("decommission pool index {pool_index} exceeds RPC range")))?; let mut client = self.get_client().await?; - let request = Request::new(ClearDecommissionRequest { pool_index }); + let mut request = Request::new(ClearDecommissionRequest { pool_index }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.clear_decommission(request).await?.into_inner(); if !response.success { @@ -1661,6 +1677,9 @@ impl PeerRestClient { Err(err) => return tier_config_reload_connection_outcome(err), }; let mut request = Request::new(LoadTransitionTierConfigRequest {}); + if let Err(err) = set_tonic_mutation_body_digest(&mut request) { + return TierConfigReloadOutcome::Terminal(Error::other(err)); + } request.set_timeout(rustfs_protos::heal_control_execution_timeout()); let response = match client.load_transition_tier_config(request).await { diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 67d04ab23..b4cfcd8b5 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -16,6 +16,7 @@ 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, }; +use crate::cluster::rpc::set_tonic_mutation_body_digest; use crate::disk::error::DiskError; use crate::disk::error::{Error, Result}; use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs}; @@ -930,10 +931,11 @@ impl PeerS3Client for RemotePeerS3Client { || async { let options: String = serde_json::to_string(opts)?; let mut client = self.get_client().await?; - let request = Request::new(HealBucketRequest { + let mut request = Request::new(HealBucketRequest { bucket: bucket.to_string(), options, }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.heal_bucket(request).await?.into_inner(); if !response.success { return if let Some(err) = response.error { @@ -986,10 +988,11 @@ impl PeerS3Client for RemotePeerS3Client { || async { let options = serde_json::to_string(opts)?; let mut client = self.get_client().await?; - let request = Request::new(MakeBucketRequest { + let mut request = Request::new(MakeBucketRequest { name: bucket.to_string(), options, }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.make_bucket(request).await?.into_inner(); if !response.success { @@ -1040,10 +1043,11 @@ impl PeerS3Client for RemotePeerS3Client { let options = serde_json::to_string(opts)?; let mut client = self.get_client().await?; - let request = Request::new(DeleteBucketRequest { + let mut request = Request::new(DeleteBucketRequest { bucket: bucket.to_string(), options, }); + set_tonic_mutation_body_digest(&mut request)?; let response = client.delete_bucket(request).await?.into_inner(); if !response.success { return if let Some(err) = response.error { diff --git a/crates/ecstore/src/cluster/rpc/remote_locker.rs b/crates/ecstore/src/cluster/rpc/remote_locker.rs index a03776d00..9707d6c8e 100644 --- a/crates/ecstore/src/cluster/rpc/remote_locker.rs +++ b/crates/ecstore/src/cluster/rpc/remote_locker.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::cluster::rpc::client::{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; use rustfs_lock::{ @@ -313,10 +314,11 @@ impl LockClient for RemoteClient { info!("remote acquire_exclusive for {}", request.resource); let mut client = self.get_client().await?; let resource_summary = request.resource.to_string(); - let req = Request::new(GenerallyLockRequest { + let mut req = Request::new(GenerallyLockRequest { args: serde_json::to_string(&request) .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?, }); + set_tonic_mutation_body_digest(&mut req)?; let resp = match self.execute_rpc("lock", &resource_summary, client.lock(req)).await { Ok(resp) => resp.into_inner(), @@ -347,7 +349,7 @@ impl LockClient for RemoteClient { let mut client = self.get_client().await?; let resource_summary = Self::summarize_resources(requests); - let req = Request::new(BatchGenerallyLockRequest { + let mut req = Request::new(BatchGenerallyLockRequest { args: requests .iter() .map(|request| { @@ -355,6 +357,7 @@ impl LockClient for RemoteClient { }) .collect::>>()?, }); + set_tonic_mutation_body_digest(&mut req)?; let resp = match self .execute_rpc("lock_batch", &resource_summary, client.lock_batch(req)) @@ -395,7 +398,8 @@ impl LockClient for RemoteClient { .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?; let mut client = self.get_client().await?; let resource_summary = unlock_request.resource.to_string(); - let req = Request::new(GenerallyLockRequest { args: request_string }); + let mut req = Request::new(GenerallyLockRequest { args: request_string }); + set_tonic_mutation_body_digest(&mut req)?; let resp = self .execute_rpc("release", &resource_summary, client.un_lock(req)) .await? @@ -414,7 +418,7 @@ impl LockClient for RemoteClient { let unlock_requests = lock_ids.iter().map(Self::create_unlock_request).collect::>(); let mut client = self.get_client().await?; let resource_summary = Self::summarize_resources(&unlock_requests); - let req = Request::new(BatchGenerallyLockRequest { + let mut req = Request::new(BatchGenerallyLockRequest { args: unlock_requests .iter() .map(|request| { @@ -422,6 +426,7 @@ impl LockClient for RemoteClient { }) .collect::>>()?, }); + set_tonic_mutation_body_digest(&mut req)?; let resp = self .execute_rpc("release_batch", &resource_summary, client.un_lock_batch(req)) @@ -440,10 +445,11 @@ impl LockClient for RemoteClient { let refresh_request = Self::create_unlock_request(lock_id); let mut client = self.get_client().await?; let resource_summary = refresh_request.resource.to_string(); - let req = Request::new(GenerallyLockRequest { + let mut req = Request::new(GenerallyLockRequest { args: serde_json::to_string(&refresh_request) .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?, }); + set_tonic_mutation_body_digest(&mut req)?; let resp = self .execute_rpc("refresh", &resource_summary, client.refresh(req)) .await? @@ -459,10 +465,11 @@ impl LockClient for RemoteClient { let force_request = Self::create_unlock_request(lock_id); let mut client = self.get_client().await?; let resource_summary = force_request.resource.to_string(); - let req = Request::new(GenerallyLockRequest { + let mut req = Request::new(GenerallyLockRequest { args: serde_json::to_string(&force_request) .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?, }); + set_tonic_mutation_body_digest(&mut req)?; let resp = self .execute_rpc("force_release", &resource_summary, client.force_un_lock(req)) .await? @@ -483,10 +490,11 @@ impl LockClient for RemoteClient { let mut client = self.get_client().await?; // Try to acquire a very short-lived lock to test availability - let req = Request::new(GenerallyLockRequest { + let mut req = Request::new(GenerallyLockRequest { args: serde_json::to_string(&status_request) .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?, }); + set_tonic_mutation_body_digest(&mut req)?; // Try exclusive lock first with very short timeout let resp = match self.execute_rpc("check_status", &resource_summary, client.lock(req)).await { @@ -497,10 +505,11 @@ impl LockClient for RemoteClient { if resp.success { // If we successfully acquired the lock, the resource was free. // Immediately release it on a best-effort basis. - let release_req = Request::new(GenerallyLockRequest { + let mut release_req = Request::new(GenerallyLockRequest { args: serde_json::to_string(&status_request) .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?, }); + set_tonic_mutation_body_digest(&mut release_req)?; let _ = self .execute_rpc("check_status_release", &resource_summary, client.un_lock(release_req)) .await; diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 5bbdcd2f9..1d99563b3 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -447,10 +447,19 @@ impl CanonicalBodyBuilder { self.push_bytes(field.as_bytes()) } + fn push_optional_str(&mut self, field: Option<&str>) -> Result<(), std::num::TryFromIntError> { + self.body.push(u8::from(field.is_some())); + self.push_str(field.unwrap_or_default()) + } + fn push_bool(&mut self, field: bool) { self.body.push(u8::from(field)); } + fn push_u32(&mut self, field: u32) { + self.body.extend_from_slice(&field.to_be_bytes()); + } + fn push_u64(&mut self, field: u64) { self.body.extend_from_slice(&field.to_be_bytes()); } @@ -465,6 +474,211 @@ impl CanonicalBodyBuilder { } } +pub const PEER_RESTSIGNAL: &str = "signal"; +pub const PEER_RESTSUB_SYS: &str = "sub-sys"; +pub const PEER_RESTDRY_RUN: &str = "dry-run"; + +/// A stable semantic body for a side-effecting unary RPC. +/// +/// Implementations deliberately enumerate handler-consumed fields instead of re-encoding the +/// protobuf message, whose unknown fields and map order are not a mixed-version contract. +pub trait CanonicalMutationBody { + fn canonical_body(&self) -> Result, std::num::TryFromIntError>; +} + +macro_rules! impl_canonical_mutation_body { + ($request:ty, $domain:expr, |$value:ident, $body:ident| $fields:block) => { + impl CanonicalMutationBody for $request { + fn canonical_body(&self) -> Result, std::num::TryFromIntError> { + let $value = self; + let mut $body = CanonicalBodyBuilder::new($domain); + $fields + Ok($body.finish()) + } + } + }; + ($request:ty, $domain:expr) => { + impl CanonicalMutationBody for $request { + fn canonical_body(&self) -> Result, std::num::TryFromIntError> { + Ok(CanonicalBodyBuilder::new($domain).finish()) + } + } + }; +} + +impl_canonical_mutation_body!( + proto_gen::node_service::SignalServiceRequest, + b"rustfs-signal-service-request-v1\0", + |request, body| { + let vars = request.vars.as_ref().map(|vars| &vars.value); + body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTSIGNAL).map(String::as_str)))?; + body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTSUB_SYS).map(String::as_str)))?; + body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTDRY_RUN).map(String::as_str)))?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::HealBucketRequest, + b"rustfs-heal-bucket-request-v1\0", + |request, body| { + body.push_str(&request.bucket)?; + body.push_str(&request.options)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::MakeBucketRequest, + b"rustfs-make-bucket-request-v1\0", + |request, body| { + body.push_str(&request.name)?; + body.push_str(&request.options)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::DeleteBucketRequest, + b"rustfs-delete-bucket-request-v1\0", + |request, body| { + body.push_str(&request.bucket)?; + body.push_str(&request.options)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::GenerallyLockRequest, + b"rustfs-lock-request-v1\0", + |request, body| { + body.push_str(&request.args)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::BatchGenerallyLockRequest, + b"rustfs-lock-batch-request-v1\0", + |request, body| { + body.push_count(request.args.len())?; + for arg in &request.args { + body.push_str(arg)?; + } + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadBucketMetadataRequest, + b"rustfs-load-bucket-metadata-request-v1\0", + |request, body| { + body.push_str(&request.bucket)?; + body.push_bool(request.scanner_maintenance_change); + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::DeleteBucketMetadataRequest, + b"rustfs-delete-bucket-metadata-request-v1\0", + |request, body| { + body.push_str(&request.bucket)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::DeletePolicyRequest, + b"rustfs-delete-policy-request-v1\0", + |request, body| { + body.push_str(&request.policy_name)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadPolicyRequest, + b"rustfs-load-policy-request-v1\0", + |request, body| { + body.push_str(&request.policy_name)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadPolicyMappingRequest, + b"rustfs-load-policy-mapping-request-v1\0", + |request, body| { + body.push_str(&request.user_or_group)?; + body.push_u64(request.user_type); + body.push_bool(request.is_group); + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::DeleteUserRequest, + b"rustfs-delete-user-request-v1\0", + |request, body| { + body.push_str(&request.access_key)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::DeleteServiceAccountRequest, + b"rustfs-delete-service-account-request-v1\0", + |request, body| { + body.push_str(&request.access_key)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadUserRequest, + b"rustfs-load-user-request-v1\0", + |request, body| { + body.push_str(&request.access_key)?; + body.push_bool(request.temp); + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadServiceAccountRequest, + b"rustfs-load-service-account-request-v1\0", + |request, body| { + body.push_str(&request.access_key)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadGroupRequest, + b"rustfs-load-group-request-v1\0", + |request, body| { + body.push_str(&request.group)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::ReloadSiteReplicationConfigRequest, + b"rustfs-reload-site-replication-config-request-v1\0" +); +impl_canonical_mutation_body!(proto_gen::node_service::ReloadPoolMetaRequest, b"rustfs-reload-pool-meta-request-v1\0"); +impl_canonical_mutation_body!( + proto_gen::node_service::StopRebalanceRequest, + b"rustfs-stop-rebalance-request-v1\0", + |request, body| { + body.push_str(&request.expected_rebalance_id)?; + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadRebalanceMetaRequest, + b"rustfs-load-rebalance-meta-request-v1\0", + |request, body| { + body.push_bool(request.start_rebalance); + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::StartDecommissionRequest, + b"rustfs-start-decommission-request-v1\0", + |request, body| { + body.push_count(request.pool_indices.len())?; + for pool_index in &request.pool_indices { + body.push_u32(*pool_index); + } + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::CancelDecommissionRequest, + b"rustfs-cancel-decommission-request-v1\0", + |request, body| { + body.push_u32(request.pool_index); + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::ClearDecommissionRequest, + b"rustfs-clear-decommission-request-v1\0", + |request, body| { + body.push_u32(request.pool_index); + } +); +impl_canonical_mutation_body!( + proto_gen::node_service::LoadTransitionTierConfigRequest, + b"rustfs-load-transition-tier-config-request-v1\0" +); + // Canonical request bodies for the mutating NodeService disk RPCs (backlog#1327 body-digest // binding). Each covers every semantic wire field — including both the msgpack `_bin` payload and // its JSON compatibility copy — so tampering with either encoding, or stripping `_bin` to force @@ -1136,6 +1350,154 @@ mod disk_mutation_canonical_tests { } } +#[cfg(test)] +mod non_disk_mutation_canonical_tests { + use super::*; + use proto_gen::node_service::*; + use std::collections::HashMap; + + macro_rules! assert_fields_bound { + ($request:ty, {$($field:ident: $value:expr),+ $(,)?}) => {{ + let baseline = <$request>::default(); + let expected = baseline.canonical_body().expect("baseline canonical body should encode"); + $( + let mut variant = baseline.clone(); + variant.$field = $value; + assert_ne!( + expected, + variant.canonical_body().expect("variant canonical body should encode"), + concat!(stringify!($request), " omitted field ", stringify!($field)), + ); + )+ + }}; + } + + fn signal_request(signal: Option<&str>, sub_system: Option<&str>, dry_run: Option<&str>) -> SignalServiceRequest { + let mut value = HashMap::new(); + if let Some(signal) = signal { + value.insert(PEER_RESTSIGNAL.to_string(), signal.to_string()); + } + if let Some(sub_system) = sub_system { + value.insert(PEER_RESTSUB_SYS.to_string(), sub_system.to_string()); + } + if let Some(dry_run) = dry_run { + value.insert(PEER_RESTDRY_RUN.to_string(), dry_run.to_string()); + } + SignalServiceRequest { + vars: Some(Mss { value }), + } + } + + #[test] + fn signal_service_canonical_body_is_versioned_and_binds_every_semantic_field() { + let request = signal_request(Some("2"), Some("scanner"), Some("false")); + let baseline = request.canonical_body().expect("small signal request should encode"); + assert!(baseline.starts_with(b"rustfs-signal-service-request-v1\0")); + + for variant in [ + signal_request(Some("1"), Some("scanner"), Some("false")), + signal_request(Some("2"), Some("heal"), Some("false")), + signal_request(Some("2"), Some("scanner"), Some("true")), + signal_request(None, Some("scanner"), Some("false")), + signal_request(Some("2"), None, Some("false")), + signal_request(Some("2"), Some("scanner"), None), + ] { + assert_ne!(baseline, variant.canonical_body().expect("small signal request should encode")); + } + + let mut ignored_field = request; + ignored_field + .vars + .as_mut() + .expect("signal vars should exist") + .value + .insert("future-field".to_string(), "ignored".to_string()); + assert_eq!( + baseline, + ignored_field + .canonical_body() + .expect("unknown signal field should not affect the semantic body"), + ); + } + + #[test] + fn signal_service_canonical_body_distinguishes_missing_and_empty_fields() { + assert_ne!( + signal_request(None, Some("scanner"), Some("false")) + .canonical_body() + .expect("missing signal should encode"), + signal_request(Some(""), Some("scanner"), Some("false")) + .canonical_body() + .expect("empty signal should encode"), + ); + } + + #[test] + fn bucket_and_lock_canonical_bodies_bind_every_semantic_field() { + assert_fields_bound!(HealBucketRequest, { bucket: "bucket".into(), options: "opts".into() }); + assert_fields_bound!(MakeBucketRequest, { name: "bucket".into(), options: "opts".into() }); + assert_fields_bound!(DeleteBucketRequest, { bucket: "bucket".into(), options: "opts".into() }); + assert_fields_bound!(GenerallyLockRequest, { args: "lock".into() }); + assert_fields_bound!(BatchGenerallyLockRequest, { args: vec!["first".into(), "second".into()] }); + + let first = BatchGenerallyLockRequest { + args: vec!["first".into(), "second".into()], + }; + let reversed = BatchGenerallyLockRequest { + args: vec!["second".into(), "first".into()], + }; + assert_ne!(first.canonical_body().unwrap(), reversed.canonical_body().unwrap()); + } + + #[test] + fn metadata_and_iam_canonical_bodies_bind_every_semantic_field() { + assert_fields_bound!(LoadBucketMetadataRequest, { + bucket: "bucket".into(), + scanner_maintenance_change: true, + }); + assert_fields_bound!(DeleteBucketMetadataRequest, { bucket: "bucket".into() }); + assert_fields_bound!(DeletePolicyRequest, { policy_name: "policy".into() }); + assert_fields_bound!(LoadPolicyRequest, { policy_name: "policy".into() }); + assert_fields_bound!(LoadPolicyMappingRequest, { + user_or_group: "user".into(), + user_type: 1, + is_group: true, + }); + assert_fields_bound!(DeleteUserRequest, { access_key: "user".into() }); + assert_fields_bound!(DeleteServiceAccountRequest, { access_key: "service".into() }); + assert_fields_bound!(LoadUserRequest, { access_key: "user".into(), temp: true }); + assert_fields_bound!(LoadServiceAccountRequest, { access_key: "service".into() }); + assert_fields_bound!(LoadGroupRequest, { group: "group".into() }); + } + + #[test] + fn control_plane_canonical_bodies_bind_every_semantic_field() { + assert_fields_bound!(StopRebalanceRequest, { expected_rebalance_id: "rebalance".into() }); + assert_fields_bound!(LoadRebalanceMetaRequest, { start_rebalance: true }); + assert_fields_bound!(StartDecommissionRequest, { pool_indices: vec![1, 2] }); + assert_fields_bound!(CancelDecommissionRequest, { pool_index: 1 }); + assert_fields_bound!(ClearDecommissionRequest, { pool_index: 1 }); + + let first = StartDecommissionRequest { + pool_indices: vec![1, 2], + }; + let reversed = StartDecommissionRequest { + pool_indices: vec![2, 1], + }; + assert_ne!(first.canonical_body().unwrap(), reversed.canonical_body().unwrap()); + + let empty_domains = [ + ReloadSiteReplicationConfigRequest::default().canonical_body().unwrap(), + ReloadPoolMetaRequest::default().canonical_body().unwrap(), + LoadTransitionTierConfigRequest::default().canonical_body().unwrap(), + ]; + assert!(empty_domains.iter().all(|body| !body.is_empty())); + assert_ne!(empty_domains[0], empty_domains[1]); + assert_ne!(empty_domains[0], empty_domains[2]); + assert_ne!(empty_domains[1], empty_domains[2]); + } +} + #[cfg(test)] mod scanner_activity_tests { use super::{ diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 112e88763..fd347ea04 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -988,107 +988,107 @@ message GetLiveEventsResponse { service NodeService { /* -------------------------------meta service-------------------------- */ - rpc Ping(PingRequest) returns (PingResponse) {}; - rpc HealBucket(HealBucketRequest) returns (HealBucketResponse) {}; - rpc ListBucket(ListBucketRequest) returns (ListBucketResponse) {}; - rpc MakeBucket(MakeBucketRequest) returns (MakeBucketResponse) {}; - rpc GetBucketInfo(GetBucketInfoRequest) returns (GetBucketInfoResponse) {}; - rpc DeleteBucket(DeleteBucketRequest) returns (DeleteBucketResponse) {}; + rpc Ping(PingRequest) returns (PingResponse) {}; // auth-policy: read-only + rpc HealBucket(HealBucketRequest) returns (HealBucketResponse) {}; // auth-policy: body-bound + rpc ListBucket(ListBucketRequest) returns (ListBucketResponse) {}; // auth-policy: read-only + rpc MakeBucket(MakeBucketRequest) returns (MakeBucketResponse) {}; // auth-policy: body-bound + rpc GetBucketInfo(GetBucketInfoRequest) returns (GetBucketInfoResponse) {}; // auth-policy: read-only + rpc DeleteBucket(DeleteBucketRequest) returns (DeleteBucketResponse) {}; // auth-policy: body-bound /* -------------------------------disk service-------------------------- */ - rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {}; - rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {}; - rpc Delete(DeleteRequest) returns (DeleteResponse) {}; - rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {}; - rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {}; - rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {}; - rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {}; - rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {}; - rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {}; - rpc PreparePartTransaction(PreparePartTransactionRequest) returns (PreparePartTransactionResponse) {}; - rpc RenamePart(RenamePartRequest) returns (RenamePartResponse) {}; - rpc SettlePartTransaction(SettlePartTransactionRequest) returns (SettlePartTransactionResponse) {}; - rpc RenameFile(RenameFileRequest) returns (RenameFileResponse) {}; - rpc Write(WriteRequest) returns (WriteResponse) {}; - rpc WriteStream(stream WriteRequest) returns (stream WriteResponse) {}; + rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {}; // auth-policy: read-only + rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {}; // auth-policy: body-bound + rpc Delete(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound + rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound + rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound + rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {}; // auth-policy: body-bound + rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {}; // auth-policy: read-only + rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {}; // auth-policy: read-only + rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {}; // auth-policy: read-only + rpc PreparePartTransaction(PreparePartTransactionRequest) returns (PreparePartTransactionResponse) {}; // auth-policy: body-bound + rpc RenamePart(RenamePartRequest) returns (RenamePartResponse) {}; // auth-policy: body-bound + rpc SettlePartTransaction(SettlePartTransactionRequest) returns (SettlePartTransactionResponse) {}; // auth-policy: body-bound + rpc RenameFile(RenameFileRequest) returns (RenameFileResponse) {}; // auth-policy: body-bound + rpc Write(WriteRequest) returns (WriteResponse) {}; // auth-policy: unimplemented + rpc WriteStream(stream WriteRequest) returns (stream WriteResponse) {}; // auth-policy: streaming // rpc Append(AppendRequest) returns (AppendResponse) {}; - rpc ReadAt(stream ReadAtRequest) returns (stream ReadAtResponse) {}; - rpc ListDir(ListDirRequest) returns (ListDirResponse) {}; - rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {}; - rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {}; - rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {}; - rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {}; - rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {}; - rpc StatVolume(StatVolumeRequest) returns (StatVolumeResponse) {}; - rpc DeletePaths(DeletePathsRequest) returns (DeletePathsResponse) {}; - rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {}; - rpc ReadMetadata(ReadMetadataRequest) returns (ReadMetadataResponse) {}; - rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {}; - rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {}; - rpc BatchReadVersion(BatchReadVersionRequest) returns (BatchReadVersionResponse) {}; - rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {}; - rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {}; - rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {}; - rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {}; - rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {}; - rpc DiskInfo(DiskInfoRequest) returns (DiskInfoResponse) {}; + rpc ReadAt(stream ReadAtRequest) returns (stream ReadAtResponse) {}; // auth-policy: streaming + rpc ListDir(ListDirRequest) returns (ListDirResponse) {}; // auth-policy: read-only + rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {}; // auth-policy: streaming + rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {}; // auth-policy: body-bound + rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {}; // auth-policy: body-bound + rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {}; // auth-policy: body-bound + rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {}; // auth-policy: read-only + rpc StatVolume(StatVolumeRequest) returns (StatVolumeResponse) {}; // auth-policy: read-only + rpc DeletePaths(DeletePathsRequest) returns (DeletePathsResponse) {}; // auth-policy: body-bound + rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {}; // auth-policy: body-bound + rpc ReadMetadata(ReadMetadataRequest) returns (ReadMetadataResponse) {}; // auth-policy: read-only + rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {}; // auth-policy: body-bound + rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {}; // auth-policy: read-only + rpc BatchReadVersion(BatchReadVersionRequest) returns (BatchReadVersionResponse) {}; // auth-policy: read-only + rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {}; // auth-policy: read-only + rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {}; // auth-policy: body-bound + rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {}; // auth-policy: body-bound + rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {}; // auth-policy: read-only + rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {}; // auth-policy: body-bound + rpc DiskInfo(DiskInfoRequest) returns (DiskInfoResponse) {}; // auth-policy: read-only /* -------------------------------lock service-------------------------- */ - rpc Lock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; - rpc UnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; - rpc ForceUnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; - rpc Refresh(GenerallyLockRequest) returns (GenerallyLockResponse) {}; - rpc LockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; - rpc UnLockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; + rpc Lock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound + rpc UnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound + rpc ForceUnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound + rpc Refresh(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound + rpc LockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; // auth-policy: body-bound + rpc UnLockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; // auth-policy: body-bound /* -------------------------------peer rest service-------------------------- */ - rpc LocalStorageInfo(LocalStorageInfoRequest) returns (LocalStorageInfoResponse) {}; - rpc ServerInfo(ServerInfoRequest) returns (ServerInfoResponse) {}; - rpc GetCpus(GetCpusRequest) returns (GetCpusResponse) {}; - rpc GetNetInfo(GetNetInfoRequest) returns (GetNetInfoResponse) {}; - rpc GetPartitions(GetPartitionsRequest) returns (GetPartitionsResponse) {}; - rpc GetOsInfo(GetOsInfoRequest) returns (GetOsInfoResponse) {}; - rpc GetSELinuxInfo(GetSELinuxInfoRequest) returns (GetSELinuxInfoResponse) {}; - rpc GetSysConfig(GetSysConfigRequest) returns (GetSysConfigResponse) {}; - rpc GetSysErrors(GetSysErrorsRequest) returns (GetSysErrorsResponse) {}; - rpc GetMemInfo(GetMemInfoRequest) returns (GetMemInfoResponse) {}; - rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {}; - rpc GetProcInfo(GetProcInfoRequest) returns (GetProcInfoResponse) {}; - rpc StartProfiling(StartProfilingRequest) returns (StartProfilingResponse) {}; - rpc DownloadProfileData(DownloadProfileDataRequest) returns (DownloadProfileDataResponse) {}; - rpc GetBucketStats(GetBucketStatsDataRequest) returns (GetBucketStatsDataResponse) {}; - rpc GetSRMetrics(GetSRMetricsDataRequest) returns (GetSRMetricsDataResponse) {}; - rpc GetAllBucketStats(GetAllBucketStatsRequest) returns (GetAllBucketStatsResponse) {}; - rpc LoadBucketMetadata(LoadBucketMetadataRequest) returns (LoadBucketMetadataResponse) {}; - rpc DeleteBucketMetadata(DeleteBucketMetadataRequest) returns (DeleteBucketMetadataResponse) {}; - rpc DeletePolicy(DeletePolicyRequest) returns (DeletePolicyResponse) {}; - rpc LoadPolicy(LoadPolicyRequest) returns (LoadPolicyResponse) {}; - rpc LoadPolicyMapping(LoadPolicyMappingRequest) returns (LoadPolicyMappingResponse) {}; - rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse) {}; - rpc DeleteServiceAccount(DeleteServiceAccountRequest) returns (DeleteServiceAccountResponse) {}; - rpc LoadUser(LoadUserRequest) returns (LoadUserResponse) {}; - rpc LoadServiceAccount(LoadServiceAccountRequest) returns (LoadServiceAccountResponse) {}; - rpc LoadGroup(LoadGroupRequest) returns (LoadGroupResponse) {}; - rpc ReloadSiteReplicationConfig(ReloadSiteReplicationConfigRequest) returns (ReloadSiteReplicationConfigResponse) {}; + rpc LocalStorageInfo(LocalStorageInfoRequest) returns (LocalStorageInfoResponse) {}; // auth-policy: read-only + rpc ServerInfo(ServerInfoRequest) returns (ServerInfoResponse) {}; // auth-policy: read-only + rpc GetCpus(GetCpusRequest) returns (GetCpusResponse) {}; // auth-policy: read-only + rpc GetNetInfo(GetNetInfoRequest) returns (GetNetInfoResponse) {}; // auth-policy: read-only + rpc GetPartitions(GetPartitionsRequest) returns (GetPartitionsResponse) {}; // auth-policy: read-only + rpc GetOsInfo(GetOsInfoRequest) returns (GetOsInfoResponse) {}; // auth-policy: read-only + rpc GetSELinuxInfo(GetSELinuxInfoRequest) returns (GetSELinuxInfoResponse) {}; // auth-policy: read-only + rpc GetSysConfig(GetSysConfigRequest) returns (GetSysConfigResponse) {}; // auth-policy: read-only + rpc GetSysErrors(GetSysErrorsRequest) returns (GetSysErrorsResponse) {}; // auth-policy: read-only + rpc GetMemInfo(GetMemInfoRequest) returns (GetMemInfoResponse) {}; // auth-policy: read-only + rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {}; // auth-policy: read-only + rpc GetProcInfo(GetProcInfoRequest) returns (GetProcInfoResponse) {}; // auth-policy: read-only + rpc StartProfiling(StartProfilingRequest) returns (StartProfilingResponse) {}; // auth-policy: unimplemented + rpc DownloadProfileData(DownloadProfileDataRequest) returns (DownloadProfileDataResponse) {}; // auth-policy: unimplemented + rpc GetBucketStats(GetBucketStatsDataRequest) returns (GetBucketStatsDataResponse) {}; // auth-policy: read-only + rpc GetSRMetrics(GetSRMetricsDataRequest) returns (GetSRMetricsDataResponse) {}; // auth-policy: unimplemented + rpc GetAllBucketStats(GetAllBucketStatsRequest) returns (GetAllBucketStatsResponse) {}; // auth-policy: unimplemented + rpc LoadBucketMetadata(LoadBucketMetadataRequest) returns (LoadBucketMetadataResponse) {}; // auth-policy: body-bound + rpc DeleteBucketMetadata(DeleteBucketMetadataRequest) returns (DeleteBucketMetadataResponse) {}; // auth-policy: body-bound + rpc DeletePolicy(DeletePolicyRequest) returns (DeletePolicyResponse) {}; // auth-policy: body-bound + rpc LoadPolicy(LoadPolicyRequest) returns (LoadPolicyResponse) {}; // auth-policy: body-bound + rpc LoadPolicyMapping(LoadPolicyMappingRequest) returns (LoadPolicyMappingResponse) {}; // auth-policy: body-bound + rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse) {}; // auth-policy: body-bound + rpc DeleteServiceAccount(DeleteServiceAccountRequest) returns (DeleteServiceAccountResponse) {}; // auth-policy: body-bound + rpc LoadUser(LoadUserRequest) returns (LoadUserResponse) {}; // auth-policy: body-bound + rpc LoadServiceAccount(LoadServiceAccountRequest) returns (LoadServiceAccountResponse) {}; // auth-policy: body-bound + rpc LoadGroup(LoadGroupRequest) returns (LoadGroupResponse) {}; // auth-policy: body-bound + rpc ReloadSiteReplicationConfig(ReloadSiteReplicationConfigRequest) returns (ReloadSiteReplicationConfigResponse) {}; // auth-policy: body-bound // rpc VerifyBinary() returns () {}; // rpc CommitBinary() returns () {}; - rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {}; - rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {}; - rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {}; - rpc GetMetacacheListing(GetMetacacheListingRequest) returns (GetMetacacheListingResponse) {}; - rpc UpdateMetacacheListing(UpdateMetacacheListingRequest) returns (UpdateMetacacheListingResponse) {}; - rpc ReloadPoolMeta(ReloadPoolMetaRequest) returns (ReloadPoolMetaResponse) {}; - rpc StopRebalance(StopRebalanceRequest) returns (StopRebalanceResponse) {}; - rpc LoadRebalanceMeta(LoadRebalanceMetaRequest) returns (LoadRebalanceMetaResponse) {}; - rpc StartDecommission(StartDecommissionRequest) returns (StartDecommissionResponse) {}; - rpc CancelDecommission(CancelDecommissionRequest) returns (CancelDecommissionResponse) {}; - rpc ClearDecommission(ClearDecommissionRequest) returns (ClearDecommissionResponse) {}; - rpc LoadTransitionTierConfig(LoadTransitionTierConfigRequest) returns (LoadTransitionTierConfigResponse) {}; - rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {}; + rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {}; // auth-policy: body-bound + rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {}; // auth-policy: body-bound + rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {}; // auth-policy: read-only + rpc GetMetacacheListing(GetMetacacheListingRequest) returns (GetMetacacheListingResponse) {}; // auth-policy: unimplemented + rpc UpdateMetacacheListing(UpdateMetacacheListingRequest) returns (UpdateMetacacheListingResponse) {}; // auth-policy: unimplemented + rpc ReloadPoolMeta(ReloadPoolMetaRequest) returns (ReloadPoolMetaResponse) {}; // auth-policy: body-bound + rpc StopRebalance(StopRebalanceRequest) returns (StopRebalanceResponse) {}; // auth-policy: body-bound + rpc LoadRebalanceMeta(LoadRebalanceMetaRequest) returns (LoadRebalanceMetaResponse) {}; // auth-policy: body-bound + rpc StartDecommission(StartDecommissionRequest) returns (StartDecommissionResponse) {}; // auth-policy: body-bound + rpc CancelDecommission(CancelDecommissionRequest) returns (CancelDecommissionResponse) {}; // auth-policy: body-bound + rpc ClearDecommission(ClearDecommissionRequest) returns (ClearDecommissionResponse) {}; // auth-policy: body-bound + rpc LoadTransitionTierConfig(LoadTransitionTierConfigRequest) returns (LoadTransitionTierConfigResponse) {}; // auth-policy: body-bound + rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {}; // auth-policy: read-only } service HealControlService { diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 4b8b94a8d..98663a753 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -28,7 +28,9 @@ use crate::storage::storage_api::rpc_consumer::node_service::{ reload_transition_tier_config, }; use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, runtime_sources}; -use crate::storage::storage_api::{sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest}; +use crate::storage::storage_api::{ + sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, +}; use bytes::Bytes; use futures::Stream; use futures_util::future::join_all; @@ -40,6 +42,7 @@ use rustfs_filemeta::MetacacheReader; use rustfs_iam::store::UserType; use rustfs_lock::LockClient; use rustfs_protos::{ + CanonicalMutationBody, models::{PingBody, PingBodyBuilder}, proto_gen::node_service::{node_service_server::NodeService as Node, *}, }; @@ -87,6 +90,15 @@ fn signal_service_response(success: bool, error_info: Option) -> Respons }) } +fn verify_node_mutation_body(request: &Request, operation: &'static str) -> Result<(), Status> { + let canonical_body = request + .get_ref() + .canonical_body() + .map_err(|_| Status::invalid_argument(format!("{operation} request length cannot be represented")))?; + verify_tonic_mutation_body_digest(request, &canonical_body) + .map_err(|err| Status::permission_denied(format!("{operation} authentication failed: {err}"))) +} + fn supports_dynamic_config_rpc(sub_system: &str) -> bool { NOTIFY_SUB_SYSTEMS.contains(&sub_system) || matches!( @@ -890,6 +902,7 @@ impl Node for NodeService { } async fn heal_bucket(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "heal bucket")?; self.handle_heal_bucket(request).await } @@ -898,6 +911,7 @@ impl Node for NodeService { } async fn make_bucket(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "make bucket")?; self.handle_make_bucket(request).await } @@ -906,6 +920,7 @@ impl Node for NodeService { } async fn delete_bucket(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "delete bucket")?; self.handle_delete_bucket(request).await } @@ -1195,18 +1210,22 @@ impl Node for NodeService { } async fn lock(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "lock")?; self.handle_lock(request).await } async fn un_lock(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "unlock")?; self.handle_un_lock(request).await } async fn force_un_lock(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "force unlock")?; self.handle_force_un_lock(request).await } async fn refresh(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "refresh lock")?; self.handle_refresh(request).await } @@ -1214,6 +1233,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "lock batch")?; self.handle_lock_batch(request).await } @@ -1221,6 +1241,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "unlock batch")?; self.handle_un_lock_batch(request).await } @@ -1340,6 +1361,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "load bucket metadata")?; self.handle_load_bucket_metadata(request).await } @@ -1347,10 +1369,12 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "delete bucket metadata")?; self.handle_delete_bucket_metadata(request).await } async fn delete_policy(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "delete policy")?; let request = request.into_inner(); let policy = request.policy_name; if policy.is_empty() { @@ -1381,6 +1405,7 @@ impl Node for NodeService { } async fn load_policy(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "load policy")?; let request = request.into_inner(); let policy = request.policy_name; if policy.is_empty() { @@ -1413,6 +1438,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "load policy mapping")?; let request = request.into_inner(); let user_or_group = request.user_or_group; if user_or_group.is_empty() { @@ -1448,6 +1474,7 @@ impl Node for NodeService { } async fn delete_user(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "delete user")?; let request = request.into_inner(); let access_key = request.access_key; if access_key.is_empty() { @@ -1480,6 +1507,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "delete service account")?; let request = request.into_inner(); let access_key = request.access_key; if access_key.is_empty() { @@ -1515,6 +1543,7 @@ impl Node for NodeService { } async fn load_user(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "load user")?; let request = request.into_inner(); let access_key = request.access_key; let temp = request.temp; @@ -1552,6 +1581,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "load service account")?; let request = request.into_inner(); let access_key = request.access_key; if access_key.is_empty() { @@ -1583,6 +1613,7 @@ impl Node for NodeService { } async fn load_group(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "load group")?; let request = request.into_inner(); let group = request.group; if group.is_empty() { @@ -1614,8 +1645,9 @@ impl Node for NodeService { async fn reload_site_replication_config( &self, - _request: Request, + request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "reload site replication config")?; let Some(_store) = self.resolve_object_store() else { return Ok(Response::new(ReloadSiteReplicationConfigResponse { success: false, @@ -1635,6 +1667,7 @@ impl Node for NodeService { } async fn signal_service(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "signal service")?; let request = request.into_inner(); let vars = match request.vars { Some(vars) => vars.value, @@ -1835,8 +1868,9 @@ impl Node for NodeService { async fn reload_pool_meta( &self, - _request: Request, + request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "reload pool metadata")?; let Some(store) = self.resolve_object_store() else { return Ok(Response::new(ReloadPoolMetaResponse { success: false, @@ -1862,6 +1896,7 @@ impl Node for NodeService { } async fn stop_rebalance(&self, request: Request) -> Result, Status> { + verify_node_mutation_body(&request, "stop rebalance")?; let Some(store) = self.resolve_object_store() else { return Ok(Response::new(StopRebalanceResponse { success: false, @@ -1882,6 +1917,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "load rebalance metadata")?; let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner(); let Some(store) = self.resolve_object_store() else { log_load_rebalance_meta_rejected!("server_not_initialized", start_rebalance); @@ -1927,6 +1963,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "start decommission")?; let Some(store) = runtime_sources::current_object_store_handle() else { return Ok(Response::new(StartDecommissionResponse { success: false, @@ -1958,6 +1995,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "cancel decommission")?; let Some(store) = runtime_sources::current_object_store_handle() else { return Ok(Response::new(CancelDecommissionResponse { success: false, @@ -1990,6 +2028,7 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "clear decommission")?; let Some(store) = runtime_sources::current_object_store_handle() else { return Ok(Response::new(ClearDecommissionResponse { success: false, @@ -2020,8 +2059,9 @@ impl Node for NodeService { async fn load_transition_tier_config( &self, - _request: Request, + request: Request, ) -> Result, Status> { + verify_node_mutation_body(&request, "load transition tier config")?; let Some(store) = self.resolve_object_store() else { return Ok(Response::new(LoadTransitionTierConfigResponse { success: false, @@ -2072,21 +2112,24 @@ mod tests { sys::NewServiceAccountOpts, }; use rustfs_kms::KmsServiceManager; + use rustfs_protos::CanonicalMutationBody as _; use rustfs_protos::models::PingBodyBuilder; use rustfs_protos::proto_gen::node_service::{ - BackgroundHealStatusRequest, CheckPartsRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, - DeletePolicyRequest, DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest, - DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest, - GetAllBucketStatsRequest, GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, - GetMetacacheListingRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, - GetProcInfoRequest, GetSeLinuxInfoRequest, GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, - HealBucketRequest, HealControlRequest, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, - LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, + BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest, + ClearDecommissionRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, DeletePolicyRequest, + DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest, DeleteVersionsRequest, + DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest, GetAllBucketStatsRequest, + GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, GetMetacacheListingRequest, + GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, + GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, HealBucketRequest, HealControlRequest, + ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, LoadGroupRequest, + LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest, MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest, RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ServerInfoRequest, SettlePartTransactionRequest, - SignalServiceRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState, + SignalServiceRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, + StartDecommissionRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState, TierMutationPrepareRequest, UpdateMetacacheListingRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, WriteRequest, heal_control_service_client::HealControlServiceClient, @@ -2095,13 +2138,79 @@ mod tests { node_service_server::NodeServiceServer, tier_mutation_control_service_server::TierMutationControlService as _, }; - use std::{collections::HashMap, sync::Arc}; + use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + }; use time::OffsetDateTime; use tokio::net::TcpListener; use tokio::time::Duration; use tokio_stream::wrappers::TcpListenerStream; use tonic::{Request, Response, Status}; + const DISK_MUTATION_RPC_METHODS: [&str; 18] = [ + "renamedata", + "deleteversion", + "deleteversions", + "writemetadata", + "updatemetadata", + "writeall", + "delete", + "deletepaths", + "renamefile", + "renamepart", + "prepareparttransaction", + "settleparttransaction", + "deletevolume", + "makevolume", + "makevolumes", + "acquiresnapshotlease", + "renewsnapshotlease", + "releasesnapshotlease", + ]; + + fn normalized_rpc_method(method: &str) -> String { + method.replace('_', "").to_ascii_lowercase() + } + + fn node_service_auth_policies() -> HashMap { + const POLICY_MARKER: &str = "// auth-policy: "; + + let schema = include_str!("../../../../crates/protos/src/node.proto"); + let service = schema + .split_once("service NodeService {") + .expect("NodeService must exist in node.proto") + .1 + .split_once("\n}") + .expect("NodeService must have a closing brace") + .0; + let mut policies = HashMap::new(); + for declaration in service.lines().filter_map(|line| line.trim().strip_prefix("rpc ")) { + let (rpc, policy) = declaration + .split_once(POLICY_MARKER) + .expect("every NodeService RPC must declare an auth-policy beside its proto definition"); + let method = rpc.split_once('(').expect("RPC declaration must have a request type").0; + assert!( + policies.insert(normalized_rpc_method(method), policy.trim()).is_none(), + "duplicate NodeService RPC {method}", + ); + } + assert!(!policies.is_empty(), "NodeService must declare RPC methods"); + policies + } + + #[test] + fn every_node_service_rpc_declares_an_auth_policy() { + const VALID_POLICIES: [&str; 4] = ["body-bound", "read-only", "streaming", "unimplemented"]; + + for (method, policy) in node_service_auth_policies() { + assert!( + VALID_POLICIES.contains(&policy), + "NodeService RPC {method} has unsupported auth-policy {policy:?}", + ); + } + } + struct HealControlMockStorage; #[async_trait::async_trait] @@ -2485,9 +2594,14 @@ mod tests { async fn every_mutating_handler_enforces_its_body_digest() { let service = make_server(); let disk = "http://node-a:9000/data/rustfs0".to_string(); + let mut covered_methods = HashSet::new(); macro_rules! assert_gated { ($method:ident, $msg:expr, $canonical:path) => {{ + assert!( + covered_methods.insert(normalized_rpc_method(stringify!($method))), + concat!("duplicate disk mutation test for ", stringify!($method)), + ); let msg = $msg; // Correct digest: the gate passes and the handler proceeds to the (unknown) disk @@ -2603,6 +2717,37 @@ mod tests { }, rustfs_protos::canonical_delete_request_body ); + assert_gated!( + acquire_snapshot_lease, + SnapshotLeaseRequest { + disk: disk.clone(), + volume: "v".into(), + path: "p".into(), + ttl_ms: 60_000, + }, + rustfs_protos::canonical_snapshot_lease_request_body + ); + assert_gated!( + renew_snapshot_lease, + SnapshotLeaseRenewRequest { + disk: disk.clone(), + volume: "v".into(), + path: "p".into(), + token: vec![1; 16].into(), + ttl_ms: 60_000, + }, + rustfs_protos::canonical_snapshot_lease_renew_request_body + ); + assert_gated!( + release_snapshot_lease, + SnapshotLeaseReleaseRequest { + disk: disk.clone(), + volume: "v".into(), + path: "p".into(), + token: vec![1; 16].into(), + }, + rustfs_protos::canonical_snapshot_lease_release_request_body + ); assert_gated!( delete_paths, DeletePathsRequest { @@ -2682,6 +2827,12 @@ mod tests { }, rustfs_protos::canonical_make_volumes_request_body ); + + let expected_methods = DISK_MUTATION_RPC_METHODS.into_iter().map(String::from).collect(); + assert_eq!( + covered_methods, expected_methods, + "the disk mutation exclusion set must exactly match handlers exercised by the independent digest test", + ); } #[tokio::test] @@ -4774,6 +4925,128 @@ mod tests { assert_eq!(signal_response.error_info.as_deref(), Some("unsupported service signal: 99")); } + #[tokio::test] + async fn signal_service_body_digest_gate_runs_before_request_handling() { + let service = create_test_node_service(); + let mut vars = HashMap::new(); + vars.insert(PEER_RESTSIGNAL.to_string(), "99".to_string()); + vars.insert(PEER_RESTSUB_SYS.to_string(), "scanner".to_string()); + vars.insert(PEER_RESTDRY_RUN.to_string(), "false".to_string()); + let message = SignalServiceRequest { + vars: Some(Mss { value: vars }), + }; + + let mut other = message.clone(); + other + .vars + .as_mut() + .expect("signal vars should exist") + .value + .insert(PEER_RESTSIGNAL.to_string(), "1".to_string()); + let mut tampered = Request::new(message.clone()); + let other_body = other.canonical_body().expect("small signal request should encode"); + set_tonic_canonical_body_digest(&mut tampered, &other_body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut tampered); + let error = service + .signal_service(tampered) + .await + .expect_err("a tampered signal request must fail before handler logic"); + assert_eq!(error.code(), tonic::Code::PermissionDenied); + + let mut signed = Request::new(message); + let body = signed.get_ref().canonical_body().expect("small signal request should encode"); + set_tonic_canonical_body_digest(&mut signed, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut signed); + let response = service + .signal_service(signed) + .await + .expect("a correctly body-bound signal request must reach handler logic") + .into_inner(); + assert!(!response.success); + assert_eq!(response.error_info.as_deref(), Some("unsupported service signal: 99")); + } + + #[tokio::test] + async fn every_non_disk_mutation_rejects_a_mismatched_body_digest() { + let service = create_test_node_service(); + let mut covered_methods = HashSet::new(); + + macro_rules! assert_tampered { + ($method:ident, $message:expr) => {{ + assert!( + covered_methods.insert(normalized_rpc_method(stringify!($method))), + concat!("duplicate non-disk mutation test for ", stringify!($method)), + ); + let mut request = Request::new($message); + set_tonic_canonical_body_digest(&mut request, b"unrelated-canonical-body") + .expect("digest metadata should encode"); + mark_v2_authenticated(&mut request); + let error = service + .$method(request) + .await + .expect_err(concat!(stringify!($method), " must reject a mismatched body digest")); + assert_eq!( + error.code(), + tonic::Code::PermissionDenied, + concat!(stringify!($method), " must authenticate before any mutation"), + ); + }}; + } + + assert_tampered!(heal_bucket, HealBucketRequest::default()); + assert_tampered!(make_bucket, MakeBucketRequest::default()); + assert_tampered!(delete_bucket, DeleteBucketRequest::default()); + assert_tampered!(lock, GenerallyLockRequest::default()); + assert_tampered!(un_lock, GenerallyLockRequest::default()); + assert_tampered!(force_un_lock, GenerallyLockRequest::default()); + assert_tampered!(refresh, GenerallyLockRequest::default()); + assert_tampered!(lock_batch, BatchGenerallyLockRequest::default()); + assert_tampered!(un_lock_batch, BatchGenerallyLockRequest::default()); + assert_tampered!(load_bucket_metadata, LoadBucketMetadataRequest::default()); + assert_tampered!(delete_bucket_metadata, DeleteBucketMetadataRequest::default()); + assert_tampered!(delete_policy, DeletePolicyRequest::default()); + assert_tampered!(load_policy, LoadPolicyRequest::default()); + assert_tampered!(load_policy_mapping, LoadPolicyMappingRequest::default()); + assert_tampered!(delete_user, DeleteUserRequest::default()); + assert_tampered!(delete_service_account, DeleteServiceAccountRequest::default()); + assert_tampered!(load_user, LoadUserRequest::default()); + assert_tampered!(load_service_account, LoadServiceAccountRequest::default()); + assert_tampered!(load_group, LoadGroupRequest::default()); + assert_tampered!(reload_site_replication_config, ReloadSiteReplicationConfigRequest::default()); + assert_tampered!(signal_service, SignalServiceRequest::default()); + assert_tampered!( + scanner_activity, + ScannerActivityRequest { + challenge: vec![7; 16].into(), + protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION, + acknowledge_instance_id: String::new(), + acknowledge_dirty_usage_generation: 0, + } + ); + assert_tampered!(reload_pool_meta, ReloadPoolMetaRequest::default()); + assert_tampered!(stop_rebalance, StopRebalanceRequest::default()); + assert_tampered!(load_rebalance_meta, LoadRebalanceMetaRequest::default()); + assert_tampered!(start_decommission, StartDecommissionRequest::default()); + assert_tampered!(cancel_decommission, CancelDecommissionRequest::default()); + assert_tampered!(clear_decommission, ClearDecommissionRequest::default()); + assert_tampered!(load_transition_tier_config, LoadTransitionTierConfigRequest::default()); + + let body_bound_methods: HashSet<_> = node_service_auth_policies() + .into_iter() + .filter_map(|(method, policy)| (policy == "body-bound").then_some(method)) + .collect(); + let disk_methods: HashSet<_> = DISK_MUTATION_RPC_METHODS.into_iter().map(String::from).collect(); + assert!( + disk_methods.is_subset(&body_bound_methods), + "every independently tested disk mutation must remain declared body-bound", + ); + let expected_methods: HashSet<_> = body_bound_methods.difference(&disk_methods).cloned().collect(); + assert_eq!( + covered_methods, expected_methods, + "proto body-bound non-disk RPCs must exactly match handlers exercised by mismatch tests", + ); + } + #[tokio::test] async fn test_scanner_activity_requires_body_bound_auth_before_storage_lookup() { let service = create_test_node_service(); @@ -4838,26 +5111,6 @@ mod tests { .expect_err("a signed malformed challenge must fail before storage lookup"); assert_eq!(malformed_current.code(), tonic::Code::InvalidArgument); - let mut tampered = Request::new(ScannerActivityRequest { - challenge: vec![7; 16].into(), - protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION, - acknowledge_instance_id: String::new(), - acknowledge_dirty_usage_generation: 0, - }); - let other = ScannerActivityRequest { - challenge: vec![8; 16].into(), - ..tampered.get_ref().clone() - }; - let other_canonical = - rustfs_protos::canonical_scanner_activity_request_body(&other).expect("scanner activity request should encode"); - set_tonic_canonical_body_digest(&mut tampered, &other_canonical).expect("digest metadata should encode"); - mark_v2_authenticated(&mut tampered); - let tampered = service - .scanner_activity(tampered) - .await - .expect_err("tampered activity challenge must fail before storage lookup"); - assert_eq!(tampered.code(), tonic::Code::PermissionDenied); - let mut downgraded = Request::new(ScannerActivityRequest { challenge: vec![7; 16].into(), protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,