gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-07 05:43:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(rpc): authenticate non-disk mutation bodies (#5425)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-29 17:17:30 +08:00
committed by GitHub
parent 453e3d0faa
commit d48870df97
8 changed files with 900 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/storage/rpc/node_service.rs
+287 -34
View File
@@ -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<String>) -> Respons
})
}
fn verify_node_mutation_body<T: CanonicalMutationBody>(request: &Request<T>, 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<HealBucketRequest>) -> Result<Response<HealBucketResponse>, 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<MakeBucketRequest>) -> Result<Response<MakeBucketResponse>, 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<DeleteBucketRequest>) -> Result<Response<DeleteBucketResponse>, 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<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "lock")?;
self.handle_lock(request).await
}
async fn un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "unlock")?;
self.handle_un_lock(request).await
}
async fn force_un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "force unlock")?;
self.handle_force_un_lock(request).await
}
async fn refresh(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "refresh lock")?;
self.handle_refresh(request).await
}
@@ -1214,6 +1233,7 @@ impl Node for NodeService {
&self,
request: Request<BatchGenerallyLockRequest>,
) -> Result<Response<BatchGenerallyLockResponse>, 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<BatchGenerallyLockRequest>,
) -> Result<Response<BatchGenerallyLockResponse>, 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<LoadBucketMetadataRequest>,
) -> Result<Response<LoadBucketMetadataResponse>, 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<DeleteBucketMetadataRequest>,
) -> Result<Response<DeleteBucketMetadataResponse>, Status> {
verify_node_mutation_body(&request, "delete bucket metadata")?;
self.handle_delete_bucket_metadata(request).await
}
async fn delete_policy(&self, request: Request<DeletePolicyRequest>) -> Result<Response<DeletePolicyResponse>, 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<LoadPolicyRequest>) -> Result<Response<LoadPolicyResponse>, 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<LoadPolicyMappingRequest>,
) -> Result<Response<LoadPolicyMappingResponse>, 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<DeleteUserRequest>) -> Result<Response<DeleteUserResponse>, 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<DeleteServiceAccountRequest>,
) -> Result<Response<DeleteServiceAccountResponse>, 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<LoadUserRequest>) -> Result<Response<LoadUserResponse>, 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<LoadServiceAccountRequest>,
) -> Result<Response<LoadServiceAccountResponse>, 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<LoadGroupRequest>) -> Result<Response<LoadGroupResponse>, 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<ReloadSiteReplicationConfigRequest>,
request: Request<ReloadSiteReplicationConfigRequest>,
) -> Result<Response<ReloadSiteReplicationConfigResponse>, 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<SignalServiceRequest>) -> Result<Response<SignalServiceResponse>, 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<ReloadPoolMetaRequest>,
request: Request<ReloadPoolMetaRequest>,
) -> Result<Response<ReloadPoolMetaResponse>, 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<StopRebalanceRequest>) -> Result<Response<StopRebalanceResponse>, 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<LoadRebalanceMetaRequest>,
) -> Result<Response<LoadRebalanceMetaResponse>, 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<StartDecommissionRequest>,
) -> Result<Response<StartDecommissionResponse>, 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<CancelDecommissionRequest>,
) -> Result<Response<CancelDecommissionResponse>, 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<ClearDecommissionRequest>,
) -> Result<Response<ClearDecommissionResponse>, 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<LoadTransitionTierConfigRequest>,
request: Request<LoadTransitionTierConfigRequest>,
) -> Result<Response<LoadTransitionTierConfigResponse>, 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<String, &'static str> {
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,