gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-29 08:27:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(admin): route pool, rebalance, and system authorization through the shared gate (#6687)

Browse Source
This commit is contained in:
Zhengchao An
2026-08-27 03:16:31 +08:00
committed by GitHub
parent 3699b6b88d
commit 0e92eac2c2
3 changed files with 419 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/pools.rs
+170 -80
View File
@@ -30,11 +30,10 @@ use crate::{
AdminPoolStatus, QueryPoolStatusRequest, current_endpoints_handle, current_notification_system, default_admin_usecase, AdminPoolStatus, QueryPoolStatusRequest, current_endpoints_handle, current_notification_system, default_admin_usecase,
}, },
admin::{ admin::{
auth::validate_admin_request, auth::authorize_admin_request,
router::{AdminOperation, Operation, S3Router}, router::{AdminOperation, Operation, S3Router},
storage_api::runtime::{EndpointServerPools, PeerRestClient}, storage_api::runtime::{EndpointServerPools, PeerRestClient},
}, },
auth::{check_key_valid, get_session_token},
error::ApiError, error::ApiError,
server::{ADMIN_PREFIX, RemoteAddr}, server::{ADMIN_PREFIX, RemoteAddr},
}; };
@@ -480,23 +479,16 @@ impl Operation for ListPools {
// GET <endpoint>/<admin-API>/pools/list // GET <endpoint>/<admin-API>/pools/list
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else { if req.credentials.is_none() {
return Err(pool_admin_missing_credentials_error("list pools")); return Err(pool_admin_missing_credentials_error("list pools"));
}; }
let (cred, owner) = authorize_admin_request(
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; &req,
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![ vec![
Action::AdminAction(AdminAction::ServerInfoAdminAction), Action::AdminAction(AdminAction::ServerInfoAdminAction),
Action::AdminAction(AdminAction::DecommissionAdminAction), Action::AdminAction(AdminAction::DecommissionAdminAction),
], ],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
) )
.await?; .await?;
@@ -577,23 +569,16 @@ impl Operation for StatusPool {
// GET <endpoint>/<admin-API>/pools/status?pool=http://server{1...4}/disk{1...4} // GET <endpoint>/<admin-API>/pools/status?pool=http://server{1...4}/disk{1...4}
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else { if req.credentials.is_none() {
return Err(pool_admin_missing_credentials_error("load pool status")); return Err(pool_admin_missing_credentials_error("load pool status"));
}; }
let (cred, owner) = authorize_admin_request(
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; &req,
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![ vec![
Action::AdminAction(AdminAction::ServerInfoAdminAction), Action::AdminAction(AdminAction::ServerInfoAdminAction),
Action::AdminAction(AdminAction::DecommissionAdminAction), Action::AdminAction(AdminAction::DecommissionAdminAction),
], ],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
) )
.await?; .await?;
@@ -632,23 +617,16 @@ impl Operation for StatusDecommission {
// GET <endpoint>/<admin-API>/decommission/status[?pool=http://server{1...4}/disk{1...4}] // GET <endpoint>/<admin-API>/decommission/status[?pool=http://server{1...4}/disk{1...4}]
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else { if req.credentials.is_none() {
return Err(pool_admin_missing_credentials_error("load decommission status")); return Err(pool_admin_missing_credentials_error("load decommission status"));
}; }
let (cred, owner) = authorize_admin_request(
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; &req,
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![ vec![
Action::AdminAction(AdminAction::ServerInfoAdminAction), Action::AdminAction(AdminAction::ServerInfoAdminAction),
Action::AdminAction(AdminAction::DecommissionAdminAction), Action::AdminAction(AdminAction::DecommissionAdminAction),
], ],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
) )
.await?; .await?;
@@ -702,27 +680,16 @@ impl Operation for StartDecommission {
"admin pool request state" "admin pool request state"
); );
let Some(input_cred) = req.credentials else { let Some(input_cred) = req.credentials.as_ref() else {
return Err(pool_admin_missing_credentials_error_with_request( return Err(pool_admin_missing_credentials_error_with_request(
"start decommission", "start decommission",
&request_id, &request_id,
&remote_addr, &remote_addr,
)); ));
}; };
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let actor = MaskedAccessKey(&input_cred.access_key).to_string(); let actor = MaskedAccessKey(&input_cred.access_key).to_string();
validate_admin_request( authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::DecommissionAdminAction)]).await?;
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::DecommissionAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
let audit = PoolAuditContext::new(&request_id, &actor, &remote_addr); let audit = PoolAuditContext::new(&request_id, &actor, &remote_addr);
let Some(endpoints) = endpoints_from_context() else { let Some(endpoints) = endpoints_from_context() else {
@@ -866,27 +833,16 @@ impl Operation for CancelDecommission {
"admin pool request state" "admin pool request state"
); );
let Some(input_cred) = req.credentials else { let Some(input_cred) = req.credentials.as_ref() else {
return Err(pool_admin_missing_credentials_error_with_request( return Err(pool_admin_missing_credentials_error_with_request(
"cancel decommission", "cancel decommission",
&request_id, &request_id,
&remote_addr, &remote_addr,
)); ));
}; };
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let actor = MaskedAccessKey(&input_cred.access_key).to_string(); let actor = MaskedAccessKey(&input_cred.access_key).to_string();
validate_admin_request( authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::DecommissionAdminAction)]).await?;
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::DecommissionAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
let audit = PoolAuditContext::new(&request_id, &actor, &remote_addr); let audit = PoolAuditContext::new(&request_id, &actor, &remote_addr);
let Some(endpoints) = endpoints_from_context() else { let Some(endpoints) = endpoints_from_context() else {
@@ -979,27 +935,16 @@ impl Operation for ClearDecommission {
"admin pool request state" "admin pool request state"
); );
let Some(input_cred) = req.credentials else { let Some(input_cred) = req.credentials.as_ref() else {
return Err(pool_admin_missing_credentials_error_with_request( return Err(pool_admin_missing_credentials_error_with_request(
"clear decommission", "clear decommission",
&request_id, &request_id,
&remote_addr, &remote_addr,
)); ));
}; };
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let actor = MaskedAccessKey(&input_cred.access_key).to_string(); let actor = MaskedAccessKey(&input_cred.access_key).to_string();
validate_admin_request( authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::DecommissionAdminAction)]).await?;
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::DecommissionAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
let audit = PoolAuditContext::new(&request_id, &actor, &remote_addr); let audit = PoolAuditContext::new(&request_id, &actor, &remote_addr);
let Some(endpoints) = endpoints_from_context() else { let Some(endpoints) = endpoints_from_context() else {
@@ -1073,15 +1018,40 @@ impl Operation for ClearDecommission {
#[cfg(test)] #[cfg(test)]
mod pools_handler_tests { mod pools_handler_tests {
use super::{ use super::{
AdminPoolStatus, PoolAuditContext, contextualize_admin_pool_api_error, AdminPoolStatus, Body, CancelDecommission, ClearDecommission, HeaderMap, ListPools, Method, Operation, Params,
decommission_admin_not_initialized_error_with_audit, decommission_peer_target, has_duplicate_indices, PoolAuditContext, S3ErrorCode, S3Request, StartDecommission, StatusDecommission, StatusPool, Uri,
parse_mutation_pool_query, parse_pool_idx_by_id, parse_status_pool_query, pool_admin_missing_credentials_error, contextualize_admin_pool_api_error, decommission_admin_not_initialized_error_with_audit, decommission_peer_target,
pool_admin_missing_credentials_error_with_request, pool_admin_pool_index_error_with_audit, has_duplicate_indices, parse_mutation_pool_query, parse_pool_idx_by_id, parse_status_pool_query,
pool_admin_pool_not_found_error_with_audit, pool_admin_pool_parse_error_with_audit, pool_admin_query_parse_error, pool_admin_missing_credentials_error, pool_admin_missing_credentials_error_with_request,
pool_admin_query_parse_error_with_audit, validate_pool_mutation_leader, validate_start_decommission_guards, pool_admin_pool_index_error_with_audit, pool_admin_pool_not_found_error_with_audit,
pool_admin_pool_parse_error_with_audit, pool_admin_query_parse_error, pool_admin_query_parse_error_with_audit,
validate_pool_mutation_leader, validate_start_decommission_guards,
}; };
use crate::admin::storage_api::runtime::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; use crate::admin::storage_api::runtime::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints};
fn credential_less_request(method: Method, uri: &'static str) -> S3Request<Body> {
S3Request {
input: Body::empty(),
method,
uri: Uri::from_static(uri),
headers: HeaderMap::new(),
extensions: http::Extensions::new(),
credentials: None,
region: None,
service: None,
trailing_headers: None,
}
}
async fn assert_missing_credentials(operation: &dyn Operation, method: Method, uri: &'static str, message: &str) {
let err = operation
.call(credential_less_request(method, uri), Params::new())
.await
.expect_err("a pool admin request without credentials must fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(err.message(), Some(message));
}
fn test_pool_endpoints(is_local: bool) -> EndpointServerPools { fn test_pool_endpoints(is_local: bool) -> EndpointServerPools {
let mut endpoint = Endpoint::try_from("http://127.0.0.1:9000/disk").expect("test endpoint should parse"); let mut endpoint = Endpoint::try_from("http://127.0.0.1:9000/disk").expect("test endpoint should parse");
endpoint.is_local = is_local; endpoint.is_local = is_local;
@@ -1380,4 +1350,124 @@ mod pools_handler_tests {
assert_eq!(value["admin_discovery"]["clusterSnapshot"], "/rustfs/admin/v4/cluster/snapshot"); assert_eq!(value["admin_discovery"]["clusterSnapshot"], "/rustfs/admin/v4/cluster/snapshot");
assert_eq!(value["admin_discovery"]["extensionsCatalog"], "/rustfs/admin/v4/extensions/catalog"); assert_eq!(value["admin_discovery"]["extensionsCatalog"], "/rustfs/admin/v4/extensions/catalog");
} }
/// Routing the pool handlers through the shared admin gate must not change
/// the wire response a caller sees when it sends no credentials at all: each
/// handler keeps its own operation-scoped message (rustfs/backlog#1829).
#[tokio::test]
async fn pool_handlers_keep_their_missing_credentials_response() {
assert_missing_credentials(
&ListPools {},
Method::GET,
"/rustfs/admin/v3/pools/list",
"Failed to list pools: missing credentials",
)
.await;
assert_missing_credentials(
&StatusPool {},
Method::GET,
"/rustfs/admin/v3/pools/status",
"Failed to load pool status: missing credentials",
)
.await;
assert_missing_credentials(
&StatusDecommission {},
Method::GET,
"/rustfs/admin/v3/decommission/status",
"Failed to load decommission status: missing credentials",
)
.await;
assert_missing_credentials(
&StartDecommission {},
Method::POST,
"/rustfs/admin/v3/pools/decommission",
"Failed to start decommission: missing credentials",
)
.await;
assert_missing_credentials(
&CancelDecommission {},
Method::POST,
"/rustfs/admin/v3/pools/cancel",
"Failed to cancel decommission: missing credentials",
)
.await;
assert_missing_credentials(
&ClearDecommission {},
Method::POST,
"/rustfs/admin/v3/pools/clear",
"Failed to clear decommission: missing credentials",
)
.await;
}
fn source_block<'a>(production: &'a str, marker: &str) -> &'a str {
let block = production
.split_once(marker)
.unwrap_or_else(|| panic!("{marker} should exist"))
.1;
let end = ["\npub struct ", "\nasync fn ", "\npub(crate) async fn ", "\n#[cfg(test)]"]
.into_iter()
.filter_map(|boundary| block.find(boundary))
.min()
.unwrap_or(block.len());
&block[..end]
}
fn assert_shared_gate_wiring(block: &str, item: &str, actions: &[&str], binds_credentials: bool) {
assert_eq!(
block.matches("authorize_admin_request(").count(),
1,
"{item} must use exactly one shared gate"
);
assert_eq!(
block.matches("Action::AdminAction(").count(),
actions.len(),
"{item} must preserve its exact action-vector length"
);
for action in actions {
assert!(block.contains(&format!("AdminAction::{action}")), "{item} must authorize with {action}");
}
assert_eq!(
block.contains("let cred = authorize_admin_request("),
binds_credentials,
"{item} credential binding must match its payload-processing contract"
);
}
/// Pins the gate wiring itself: every pool handler authorizes through
/// `authorize_admin_request` with the same action vector it used before the
/// deduplication, and the mutating handlers keep deriving their audit actor
/// from the caller-supplied access key (rustfs/backlog#1829).
#[test]
fn pool_handlers_use_the_shared_admin_gate_with_their_actions() {
let production = include_str!("pools.rs")
.split("\n#[cfg(test)]\nmod ")
.next()
.expect("production source must precede the test module");
let read_actions = ["ServerInfoAdminAction", "DecommissionAdminAction"];
let mutate_actions = ["DecommissionAdminAction"];
for (handler, actions) in [
("ListPools", read_actions.as_slice()),
("StatusPool", read_actions.as_slice()),
("StatusDecommission", read_actions.as_slice()),
("StartDecommission", mutate_actions.as_slice()),
("CancelDecommission", mutate_actions.as_slice()),
("ClearDecommission", mutate_actions.as_slice()),
] {
let block = source_block(production, &format!("impl Operation for {handler}"));
assert_shared_gate_wiring(block, handler, actions, false);
}
for handler in ["StartDecommission", "CancelDecommission", "ClearDecommission"] {
let block = source_block(production, &format!("impl Operation for {handler}"));
assert!(
block.contains("let actor = MaskedAccessKey(&input_cred.access_key).to_string();"),
"{handler} must keep masking the caller access key for its audit trail"
);
}
assert!(!production.contains("check_key_valid(get_session_token"));
assert!(!production.contains("validate_admin_request("));
}
} }
rustfs/src/admin/handlers/rebalance.rs
+102 -42
View File
@@ -24,10 +24,9 @@ use crate::admin::storage_api::runtime::{ECStore, NotificationSys};
use crate::{ use crate::{
admin::runtime_sources::current_notification_system, admin::runtime_sources::current_notification_system,
admin::{ admin::{
auth::validate_admin_request, auth::authorize_admin_request,
router::{AdminOperation, Operation, S3Router}, router::{AdminOperation, Operation, S3Router},
}, },
auth::{check_key_valid, get_session_token},
server::{ADMIN_PREFIX, RemoteAddr}, server::{ADMIN_PREFIX, RemoteAddr},
}; };
use http::{HeaderMap, HeaderValue, StatusCode, Uri}; use http::{HeaderMap, HeaderValue, StatusCode, Uri};
@@ -497,23 +496,12 @@ impl Operation for RebalanceStart {
"admin rebalance state" "admin rebalance state"
); );
let Some(input_cred) = req.credentials else { let Some(input_cred) = req.credentials.as_ref() else {
return Err(s3_error!(InvalidRequest, "authentication required")); return Err(s3_error!(InvalidRequest, "authentication required"));
}; };
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let actor = MaskedAccessKey(&input_cred.access_key).to_string(); let actor = MaskedAccessKey(&input_cred.access_key).to_string();
validate_admin_request( authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::RebalanceAdminAction)]).await?;
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::RebalanceAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
if rebalance_query_present(&req.uri) { if rebalance_query_present(&req.uri) {
log_rebalance_request_rejected("start", "invalid_query_parameters", &request_id, &actor, &remote_addr); log_rebalance_request_rejected("start", "invalid_query_parameters", &request_id, &actor, &remote_addr);
@@ -792,23 +780,12 @@ impl Operation for RebalanceStatus {
"admin rebalance state" "admin rebalance state"
); );
let Some(input_cred) = req.credentials else { let Some(input_cred) = req.credentials.as_ref() else {
return Err(s3_error!(InvalidRequest, "authentication required")); return Err(s3_error!(InvalidRequest, "authentication required"));
}; };
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let actor = MaskedAccessKey(&input_cred.access_key).to_string(); let actor = MaskedAccessKey(&input_cred.access_key).to_string();
validate_admin_request( authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::RebalanceAdminAction)]).await?;
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::RebalanceAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
let Some(store) = object_store_from_extensions(&req.extensions) else { let Some(store) = object_store_from_extensions(&req.extensions) else {
return Err(s3_error!(InternalError, "object layer is not initialized")); return Err(s3_error!(InternalError, "object layer is not initialized"));
@@ -924,23 +901,12 @@ impl Operation for RebalanceStop {
"admin rebalance state" "admin rebalance state"
); );
let Some(input_cred) = req.credentials else { let Some(input_cred) = req.credentials.as_ref() else {
return Err(s3_error!(InvalidRequest, "authentication required")); return Err(s3_error!(InvalidRequest, "authentication required"));
}; };
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let actor = MaskedAccessKey(&input_cred.access_key).to_string(); let actor = MaskedAccessKey(&input_cred.access_key).to_string();
validate_admin_request( authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::RebalanceAdminAction)]).await?;
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::RebalanceAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
if rebalance_query_present(&req.uri) { if rebalance_query_present(&req.uri) {
log_rebalance_request_rejected("stop", "invalid_query_parameters", &request_id, &actor, &remote_addr); log_rebalance_request_rejected("stop", "invalid_query_parameters", &request_id, &actor, &remote_addr);
@@ -1092,7 +1058,8 @@ mod rebalance_handler_tests {
use super::build_rebalance_pool_progress; use super::build_rebalance_pool_progress;
use super::calculate_rebalance_progress; use super::calculate_rebalance_progress;
use super::{ use super::{
RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStartStep, RebalanceStopPropagationStatus, Body, HeaderMap, Method, Operation, Params, RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStart,
RebalanceStartStep, RebalanceStatus, RebalanceStop, RebalanceStopPropagationStatus, S3ErrorCode, S3Request, Uri,
build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status, build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status,
rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets, rebalance_rollback_failure_message, rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets, rebalance_rollback_failure_message,
rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_start_steps, rebalance_stop_target_id, rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_start_steps, rebalance_stop_target_id,
@@ -1104,6 +1071,29 @@ mod rebalance_handler_tests {
}; };
use time::OffsetDateTime; use time::OffsetDateTime;
fn credential_less_request(method: Method, uri: &'static str) -> S3Request<Body> {
S3Request {
input: Body::empty(),
method,
uri: Uri::from_static(uri),
headers: HeaderMap::new(),
extensions: http::Extensions::new(),
credentials: None,
region: None,
service: None,
trailing_headers: None,
}
}
async fn assert_missing_credentials(operation: &dyn Operation, method: Method, uri: &'static str) {
let err = operation
.call(credential_less_request(method, uri), Params::new())
.await
.expect_err("a rebalance admin request without credentials must fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(err.message(), Some("authentication required"));
}
fn started_rebalance_meta(id: &str) -> RebalanceMeta { fn started_rebalance_meta(id: &str) -> RebalanceMeta {
RebalanceMeta { RebalanceMeta {
id: id.to_string(), id: id.to_string(),
@@ -1928,4 +1918,74 @@ mod rebalance_handler_tests {
vec!["peer node-b load_rebalance_meta(start=false) failed: timeout"] vec!["peer node-b load_rebalance_meta(start=false) failed: timeout"]
); );
} }
/// The rebalance handlers pre-check credentials before delegating to the
/// shared admin gate, so a credential-less request keeps returning
/// `InvalidRequest: authentication required` rather than the gate's own
/// "get cred failed" wording (rustfs/backlog#1829).
#[tokio::test]
async fn rebalance_handlers_keep_their_missing_credentials_response() {
assert_missing_credentials(&RebalanceStart {}, Method::POST, "/rustfs/admin/v3/rebalance/start").await;
assert_missing_credentials(&RebalanceStatus {}, Method::GET, "/rustfs/admin/v3/rebalance/status").await;
assert_missing_credentials(&RebalanceStop {}, Method::POST, "/rustfs/admin/v3/rebalance/stop").await;
}
fn source_block<'a>(production: &'a str, marker: &str) -> &'a str {
let block = production
.split_once(marker)
.unwrap_or_else(|| panic!("{marker} should exist"))
.1;
let end = [
"\npub struct ",
"\nasync fn ",
"\npub(crate) async fn ",
"\nmod ",
"\n#[cfg(test)]",
]
.into_iter()
.filter_map(|boundary| block.find(boundary))
.min()
.unwrap_or(block.len());
&block[..end]
}
/// All three rebalance handlers authorize through the single shared gate with
/// the same `RebalanceAdminAction` vector they used before the deduplication,
/// and none of them binds the returned credentials (rustfs/backlog#1829).
#[test]
fn rebalance_handlers_use_the_shared_admin_gate_with_their_actions() {
let production = include_str!("rebalance.rs")
.split("\n#[cfg(test)]\nmod ")
.next()
.expect("production source must precede the test module");
for handler in ["RebalanceStart", "RebalanceStatus", "RebalanceStop"] {
let block = source_block(production, &format!("impl Operation for {handler}"));
assert_eq!(
block.matches("authorize_admin_request(").count(),
1,
"{handler} must use exactly one shared gate"
);
assert_eq!(
block.matches("Action::AdminAction(").count(),
1,
"{handler} must preserve its exact action-vector length"
);
assert!(
block.contains("AdminAction::RebalanceAdminAction"),
"{handler} must authorize with RebalanceAdminAction"
);
assert!(
!block.contains("let cred = authorize_admin_request("),
"{handler} does not consume the authenticated credentials"
);
assert!(
block.contains("let actor = MaskedAccessKey(&input_cred.access_key).to_string();"),
"{handler} must keep masking the caller access key for its audit trail"
);
}
assert!(!production.contains("check_key_valid(get_session_token"));
assert!(!production.contains("validate_admin_request("));
}
} }
rustfs/src/admin/handlers/system.rs
+147 -85
View File
@@ -13,7 +13,7 @@
// limitations under the License. // limitations under the License.
use super::{cluster_snapshot, metrics}; use super::{cluster_snapshot, metrics};
use crate::admin::auth::validate_admin_request; use crate::admin::auth::authorize_admin_request;
use crate::admin::handlers::account::{ACCOUNT_INFO_ROUTE, ACCOUNT_PASSWORD_ROUTE}; use crate::admin::handlers::account::{ACCOUNT_INFO_ROUTE, ACCOUNT_PASSWORD_ROUTE};
use crate::admin::handlers::mfa::{ACCOUNT_MFA_ROUTE, MFA_CHALLENGE_ROUTE, USER_MFA_ROUTE}; use crate::admin::handlers::mfa::{ACCOUNT_MFA_ROUTE, MFA_CHALLENGE_ROUTE, USER_MFA_ROUTE};
use crate::admin::route_policy::{ use crate::admin::route_policy::{
@@ -31,9 +31,8 @@ use crate::admin::storage_api::cluster::{
CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider, CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider,
}; };
use crate::admin::storage_api::storageclass as storage_class_contract; use crate::admin::storage_api::storageclass as storage_class_contract;
use crate::auth::{check_key_valid, get_session_token};
use crate::runtime_capabilities::{EndpointTopologySnapshotProvider, RustFsObservabilitySnapshotProvider}; use crate::runtime_capabilities::{EndpointTopologySnapshotProvider, RustFsObservabilitySnapshotProvider};
use crate::server::{ADMIN_PREFIX, RemoteAddr}; use crate::server::ADMIN_PREFIX;
use crate::workload_admission::workload_admission_registry_snapshot; use crate::workload_admission::workload_admission_registry_snapshot;
use http::{HeaderMap, HeaderValue}; use http::{HeaderMap, HeaderValue};
use hyper::{Method, StatusCode}; use hyper::{Method, StatusCode};
@@ -248,10 +247,10 @@ fn request_graceful_shutdown() {}
#[async_trait::async_trait] #[async_trait::async_trait]
impl Operation for ServiceHandle { impl Operation for ServiceHandle {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials.as_ref() else { if req.credentials.is_none() {
log_system_request_rejected!("service_handle", "missing_credentials"); log_system_request_rejected!("service_handle", "missing_credentials");
return Err(s3_error!(InvalidRequest, "get cred failed")); return Err(s3_error!(InvalidRequest, "get cred failed"));
}; }
let Some(action) = service_action_from_uri(&req.uri) else { let Some(action) = service_action_from_uri(&req.uri) else {
log_system_request_rejected!("service_handle", "invalid_action"); log_system_request_rejected!("service_handle", "invalid_action");
@@ -265,10 +264,7 @@ impl Operation for ServiceHandle {
ServiceAction::Freeze | ServiceAction::Unfreeze => AdminAction::ServiceFreezeAdminAction, ServiceAction::Freeze | ServiceAction::Unfreeze => AdminAction::ServiceFreezeAdminAction,
}; };
let (cred, owner) = authorize_admin_request(&req, vec![Action::AdminAction(admin_action)]).await?;
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(admin_action)], remote_addr).await?;
let response = match action { let response = match action {
ServiceAction::Restart => { ServiceAction::Restart => {
@@ -362,22 +358,11 @@ struct ServerUpdateStatus {
#[async_trait::async_trait] #[async_trait::async_trait]
impl Operation for UpdateHandler { impl Operation for UpdateHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials.as_ref() else { if req.credentials.is_none() {
log_system_request_rejected!("server_update", "missing_credentials"); log_system_request_rejected!("server_update", "missing_credentials");
return Err(s3_error!(InvalidRequest, "get cred failed")); return Err(s3_error!(InvalidRequest, "get cred failed"));
}; }
let (cred, owner) = authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::ServerUpdateAdminAction)]).await?;
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::ServerUpdateAdminAction)],
remote_addr,
)
.await?;
// MinIO's server-update downloads and swaps the binary in place. RustFS // MinIO's server-update downloads and swaps the binary in place. RustFS
// intentionally does not implement in-process self-update: binaries are // intentionally does not implement in-process self-update: binaries are
@@ -459,24 +444,12 @@ fn bitrot_selftest_status_str() -> &'static str {
#[async_trait::async_trait] #[async_trait::async_trait]
impl Operation for ServerInfoHandler { impl Operation for ServerInfoHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else { if req.credentials.is_none() {
log_system_request_rejected!("query_server_info", "missing_credentials"); log_system_request_rejected!("query_server_info", "missing_credentials");
return Err(s3_error!(InvalidRequest, "get cred failed")); return Err(s3_error!(InvalidRequest, "get cred failed"));
}; }
let (cred, owner) = authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
remote_addr,
)
.await?;
let usecase = default_admin_usecase(); let usecase = default_admin_usecase();
let info = usecase let info = usecase
@@ -536,22 +509,11 @@ impl Operation for InspectDataHandler {
use crate::admin::storage_api::object::StorageObjectOptions; use crate::admin::storage_api::object::StorageObjectOptions;
use tokio::io::AsyncReadExt; use tokio::io::AsyncReadExt;
let Some(input_cred) = req.credentials.as_ref() else { if req.credentials.is_none() {
log_system_request_rejected!("inspect_data", "missing_credentials"); log_system_request_rejected!("inspect_data", "missing_credentials");
return Err(s3_error!(InvalidRequest, "get cred failed")); return Err(s3_error!(InvalidRequest, "get cred failed"));
}; }
let (cred, owner) = authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::InspectDataAction)]).await?;
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::InspectDataAction)],
remote_addr,
)
.await?;
// MinIO's inspect-data exports a signed archive of raw drive files for a // MinIO's inspect-data exports a signed archive of raw drive files for a
// `volume`/`file` glob. RustFS erasure-codes and (optionally) encrypts // `volume`/`file` glob. RustFS erasure-codes and (optionally) encrypts
@@ -608,24 +570,12 @@ pub struct StorageInfoHandler {}
#[async_trait::async_trait] #[async_trait::async_trait]
impl Operation for StorageInfoHandler { impl Operation for StorageInfoHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else { if req.credentials.is_none() {
log_system_request_rejected!("query_storage_info", "missing_credentials"); log_system_request_rejected!("query_storage_info", "missing_credentials");
return Err(s3_error!(InvalidRequest, "get cred failed")); return Err(s3_error!(InvalidRequest, "get cred failed"));
}; }
let (cred, owner) = authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::StorageInfoAdminAction)]).await?;
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::StorageInfoAdminAction)],
remote_addr,
)
.await?;
let usecase = default_admin_usecase(); let usecase = default_admin_usecase();
let info = usecase.execute_query_storage_info().await.map_err(S3Error::from)?; let info = usecase.execute_query_storage_info().await.map_err(S3Error::from)?;
@@ -1178,16 +1128,12 @@ fn summarize_named_capability_statuses<const N: usize>(
#[async_trait::async_trait] #[async_trait::async_trait]
impl Operation for RuntimeCapabilitiesHandler { impl Operation for RuntimeCapabilitiesHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else { if req.credentials.is_none() {
log_system_request_rejected!("runtime_capabilities", "missing_credentials"); log_system_request_rejected!("runtime_capabilities", "missing_credentials");
return Err(s3_error!(InvalidRequest, "get cred failed")); return Err(s3_error!(InvalidRequest, "get cred failed"));
}; }
let (cred, owner) = authorize_admin_request(&req, runtime_capabilities_gate_actions()).await?;
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
validate_admin_request(&req.headers, &cred, owner, false, runtime_capabilities_gate_actions(), remote_addr).await?;
let response = build_runtime_capabilities_response().await.map_err(|err| { let response = build_runtime_capabilities_response().await.map_err(|err| {
log_system_request_failed!("runtime_capabilities", "build_runtime_capabilities_failed", err); log_system_request_failed!("runtime_capabilities", "build_runtime_capabilities_failed", err);
@@ -1220,16 +1166,12 @@ pub(crate) fn data_usage_info_gate_actions() -> Vec<Action> {
#[async_trait::async_trait] #[async_trait::async_trait]
impl Operation for DataUsageInfoHandler { impl Operation for DataUsageInfoHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else { if req.credentials.is_none() {
log_system_request_rejected!("query_data_usage_info", "missing_credentials"); log_system_request_rejected!("query_data_usage_info", "missing_credentials");
return Err(s3_error!(InvalidRequest, "get cred failed")); return Err(s3_error!(InvalidRequest, "get cred failed"));
}; }
let (cred, owner) = authorize_admin_request(&req, data_usage_info_gate_actions()).await?;
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
validate_admin_request(&req.headers, &cred, owner, false, data_usage_info_gate_actions(), remote_addr).await?;
let usecase = default_admin_usecase(); let usecase = default_admin_usecase();
let info = usecase.execute_query_data_usage_info().await.map_err(S3Error::from)?; let info = usecase.execute_query_data_usage_info().await.map_err(S3Error::from)?;
@@ -1250,10 +1192,11 @@ impl Operation for DataUsageInfoHandler {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{ use super::{
MANUAL_TRANSITION_JOB_ROUTE, MANUAL_TRANSITION_RUN_ROUTE, OBSERVABILITY_SUMMARY_RESOLVED, RuntimeCapabilitiesHandler, DataUsageInfoHandler, InspectDataHandler, MANUAL_TRANSITION_JOB_ROUTE, MANUAL_TRANSITION_RUN_ROUTE,
SITE_REPLICATION_EDIT_ROUTE, SITE_REPLICATION_INFO_ROUTE, SITE_REPLICATION_REPAIR_ROUTE, OBSERVABILITY_SUMMARY_RESOLVED, RuntimeCapabilitiesHandler, SITE_REPLICATION_EDIT_ROUTE, SITE_REPLICATION_INFO_ROUTE,
SITE_REPLICATION_REPAIR_STATUS_ROUTE, SITE_REPLICATION_RESYNC_ROUTE, ServerInfoResponse, TOPOLOGY_SNAPSHOT_NOT_AVAILABLE, SITE_REPLICATION_REPAIR_ROUTE, SITE_REPLICATION_REPAIR_STATUS_ROUTE, SITE_REPLICATION_RESYNC_ROUTE, ServerInfoHandler,
TOPOLOGY_SUMMARY_RESOLVED, admin_route_capability_from_inventory, build_runtime_capabilities_response, ServerInfoResponse, ServiceHandle, StorageInfoHandler, TOPOLOGY_SNAPSHOT_NOT_AVAILABLE, TOPOLOGY_SUMMARY_RESOLVED,
UpdateHandler, admin_route_capability_from_inventory, build_runtime_capabilities_response,
build_runtime_capabilities_summary, data_usage_info_gate_actions, runtime_capabilities_gate_actions, build_runtime_capabilities_summary, data_usage_info_gate_actions, runtime_capabilities_gate_actions,
system_admin_discovery, system_admin_discovery,
}; };
@@ -1908,4 +1851,123 @@ mod tests {
.contains(TOPOLOGY_SUMMARY_RESOLVED) .contains(TOPOLOGY_SUMMARY_RESOLVED)
); );
} }
fn credential_less_request(method: Method, uri: &'static str) -> S3Request<Body> {
S3Request {
input: Body::empty(),
method,
uri: Uri::from_static(uri),
headers: HeaderMap::new(),
extensions: Extensions::new(),
credentials: None,
region: None,
service: None,
trailing_headers: None,
}
}
async fn assert_missing_credentials(operation: &dyn Operation, method: Method, uri: &'static str) {
let err = operation
.call(credential_less_request(method, uri), Params::new())
.await
.expect_err("a system admin request without credentials must fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(err.message(), Some("get cred failed"));
}
/// Every system handler pre-checks credentials before delegating to the
/// shared admin gate, so the credential-less response stays byte-identical to
/// what it was before the deduplication. `ServiceHandle` in particular must
/// keep rejecting on missing credentials *before* it parses the requested
/// service action (rustfs/backlog#1829).
#[tokio::test]
async fn system_handlers_keep_their_missing_credentials_response() {
assert_missing_credentials(&ServiceHandle {}, Method::POST, "/rustfs/admin/v3/service?action=restart").await;
assert_missing_credentials(&ServiceHandle {}, Method::POST, "/rustfs/admin/v3/service").await;
assert_missing_credentials(&UpdateHandler {}, Method::POST, "/rustfs/admin/v3/update").await;
assert_missing_credentials(&ServerInfoHandler {}, Method::GET, "/rustfs/admin/v3/info").await;
assert_missing_credentials(&InspectDataHandler {}, Method::GET, "/rustfs/admin/v3/inspect-data").await;
assert_missing_credentials(&StorageInfoHandler {}, Method::GET, "/rustfs/admin/v3/storageinfo").await;
assert_missing_credentials(&RuntimeCapabilitiesHandler {}, Method::GET, "/rustfs/admin/v4/runtime/capabilities").await;
assert_missing_credentials(&DataUsageInfoHandler {}, Method::GET, "/rustfs/admin/v3/datausageinfo").await;
}
fn source_block<'a>(production: &'a str, marker: &str) -> &'a str {
let block = production
.split_once(marker)
.unwrap_or_else(|| panic!("{marker} should exist"))
.1;
let end = [
"\npub struct ",
"\nasync fn ",
"\npub(crate) async fn ",
"\npub fn ",
"\npub(crate) fn ",
"\nfn ",
"\nmod ",
"\n#[cfg(test)]",
]
.into_iter()
.filter_map(|boundary| block.find(boundary))
.min()
.unwrap_or(block.len());
&block[..end]
}
/// Pins the gate wiring: each system handler authorizes through exactly one
/// `authorize_admin_request` call carrying the same action vector it used
/// before the deduplication. The two gate-action helpers must be passed
/// through by name so the vectors pinned by the tests above keep governing
/// the live gate (rustfs/backlog#1829).
#[test]
fn system_handlers_use_the_shared_admin_gate_with_their_actions() {
let production = include_str!("system.rs")
.split("\n#[cfg(test)]\nmod ")
.next()
.expect("production source must precede the test module");
let service_tokens = [
"AdminAction::ServiceRestartAdminAction",
"AdminAction::ServiceStopAdminAction",
"AdminAction::ServiceFreezeAdminAction",
];
let update_tokens = ["AdminAction::ServerUpdateAdminAction"];
let server_info_tokens = ["AdminAction::ServerInfoAdminAction"];
let inspect_data_tokens = ["AdminAction::InspectDataAction"];
let storage_info_tokens = ["AdminAction::StorageInfoAdminAction"];
let runtime_capabilities_tokens = ["runtime_capabilities_gate_actions()"];
let data_usage_info_tokens = ["data_usage_info_gate_actions()"];
for (handler, inline_admin_actions, tokens) in [
("ServiceHandle", 1usize, service_tokens.as_slice()),
("UpdateHandler", 1, update_tokens.as_slice()),
("ServerInfoHandler", 1, server_info_tokens.as_slice()),
("InspectDataHandler", 1, inspect_data_tokens.as_slice()),
("StorageInfoHandler", 1, storage_info_tokens.as_slice()),
("RuntimeCapabilitiesHandler", 0, runtime_capabilities_tokens.as_slice()),
("DataUsageInfoHandler", 0, data_usage_info_tokens.as_slice()),
] {
let block = source_block(production, &format!("impl Operation for {handler}"));
assert_eq!(
block.matches("authorize_admin_request(").count(),
1,
"{handler} must use exactly one shared gate"
);
assert_eq!(
block.matches("Action::AdminAction(").count(),
inline_admin_actions,
"{handler} must preserve its exact inline action-vector length"
);
for token in tokens {
assert!(block.contains(token), "{handler} must authorize with {token}");
}
assert!(
!block.contains("let cred = authorize_admin_request("),
"{handler} does not consume the authenticated credentials"
);
}
assert!(!production.contains("check_key_valid(get_session_token"));
assert!(!production.contains("validate_admin_request("));
}
} }