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

refactor(admin): route IAM policy and group auth through the shared gate (#6685)

Browse Source
This commit is contained in:
Zhengchao An
2026-08-27 03:15:50 +08:00
committed by GitHub
parent a42046b79c
commit 4154a3b7ca
2 changed files with 232 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/group.rs
+98 -73
View File
@@ -16,13 +16,13 @@ use super::iam_error::iam_error_to_s3_error;
use crate::{
admin::runtime_sources::current_action_credentials,
admin::{
auth::validate_admin_request,
auth::authorize_admin_request,
handlers::site_replication::site_replication_iam_change_hook,
router::{AdminOperation, Operation, S3Router},
utils::has_space_be,
},
auth::{check_key_valid, constant_time_eq, get_session_token},
server::{ADMIN_PREFIX, RemoteAddr},
auth::constant_time_eq,
server::ADMIN_PREFIX,
};
use http::{HeaderMap, StatusCode};
use hyper::Method;
@@ -98,22 +98,11 @@ impl Operation for ListGroups {
"admin group state"
);
let Some(input_cred) = req.credentials else {
if req.credentials.is_none() {
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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::ListGroupsAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::ListGroupsAdminAction)]).await?;
let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
return Err(s3_error!(InternalError, "iam is not initialized"));
@@ -154,22 +143,11 @@ impl Operation for GetGroup {
"admin group state"
);
let Some(input_cred) = req.credentials else {
if req.credentials.is_none() {
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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::GetGroupAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::GetGroupAdminAction)]).await?;
let query = {
if let Some(query) = req.uri.query() {
@@ -237,22 +215,11 @@ impl Operation for DeleteGroup {
"admin group state"
);
let Some(input_cred) = req.credentials else {
if req.credentials.is_none() {
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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::RemoveUserFromGroupAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::RemoveUserFromGroupAdminAction)]).await?;
let group = decode_delete_group_name(&params)?;
@@ -363,22 +330,11 @@ impl Operation for SetGroupStatus {
"admin group state"
);
let Some(input_cred) = req.credentials else {
if req.credentials.is_none() {
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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::EnableGroupAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::EnableGroupAdminAction)]).await?;
let query = {
if let Some(query) = req.uri.query() {
@@ -488,22 +444,11 @@ impl Operation for UpdateGroupMembers {
"admin group state"
);
let Some(input_cred) = req.credentials else {
if req.credentials.is_none() {
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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::AddUserToGroupAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::AddUserToGroupAdminAction)]).await?;
let mut input = req.input;
let body = match input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await {
@@ -673,8 +618,32 @@ impl Operation for UpdateGroupMembers {
#[cfg(test)]
mod tests {
use super::*;
use http::Uri;
use matchit::Router;
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 group admin request without credentials must fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(err.message(), Some("authentication required"));
}
fn with_delete_group_params<T>(path: &str, f: impl FnOnce(&Params<'_, '_>) -> T) -> T {
let mut router = Router::new();
router
@@ -727,4 +696,60 @@ mod tests {
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
assert_eq!(err.message(), Some("group name contains invalid characters"));
}
#[tokio::test]
async fn group_handlers_keep_their_missing_credentials_response() {
assert_missing_credentials(&ListGroups {}, Method::GET, "/rustfs/admin/v3/groups").await;
assert_missing_credentials(&GetGroup {}, Method::GET, "/rustfs/admin/v3/group").await;
assert_missing_credentials(&DeleteGroup {}, Method::DELETE, "/rustfs/admin/v3/group/dev").await;
assert_missing_credentials(&SetGroupStatus {}, Method::PUT, "/rustfs/admin/v3/set-group-status").await;
assert_missing_credentials(&UpdateGroupMembers {}, Method::PUT, "/rustfs/admin/v3/update-group-members").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]
}
#[test]
fn group_handlers_use_the_shared_admin_gate_with_their_actions() {
let production = include_str!("group.rs")
.split("\n#[cfg(test)]\n")
.next()
.expect("production source must precede tests");
for (handler, action) in [
("ListGroups", "ListGroupsAdminAction"),
("GetGroup", "GetGroupAdminAction"),
("DeleteGroup", "RemoveUserFromGroupAdminAction"),
("SetGroupStatus", "EnableGroupAdminAction"),
("UpdateGroupMembers", "AddUserToGroupAdminAction"),
] {
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 request exactly one admin action"
);
assert!(
block.contains(&format!("AdminAction::{action}")),
"{handler} must authorize with {action}"
);
}
assert!(!production.contains("check_key_valid(get_session_token"));
}
}
rustfs/src/admin/handlers/policies.rs
+134 -107
View File
@@ -16,13 +16,12 @@ use super::iam_error::iam_error_to_s3_error;
use crate::{
admin::runtime_sources::current_action_credentials,
admin::{
auth::validate_admin_request,
auth::authorize_admin_request,
handlers::site_replication::site_replication_iam_change_hook,
router::{AdminOperation, Operation, S3Router},
utils::{encode_compatible_admin_payload, has_space_be, read_compatible_admin_body},
},
auth::{check_key_valid, get_session_token},
server::{ADMIN_PREFIX, RemoteAddr},
server::ADMIN_PREFIX,
};
use http::{HeaderMap, StatusCode};
use hyper::Method;
@@ -117,22 +116,11 @@ pub struct ListCannedPolicies {}
#[async_trait::async_trait]
impl Operation for ListCannedPolicies {
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(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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::ListUserPoliciesAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::ListUserPoliciesAdminAction)]).await?;
let query = {
if let Some(query) = req.uri.query() {
@@ -184,22 +172,11 @@ pub struct AddCannedPolicy {}
#[async_trait::async_trait]
impl Operation for AddCannedPolicy {
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(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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::CreatePolicyAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::CreatePolicyAdminAction)]).await?;
let query = {
if let Some(query) = req.uri.query() {
@@ -304,22 +281,11 @@ pub struct InfoCannedPolicy {}
#[async_trait::async_trait]
impl Operation for InfoCannedPolicy {
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(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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::GetPolicyAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::GetPolicyAdminAction)]).await?;
let query = {
if let Some(query) = req.uri.query() {
@@ -370,22 +336,11 @@ pub struct RemoveCannedPolicy {}
#[async_trait::async_trait]
impl Operation for RemoveCannedPolicy {
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(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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::DeletePolicyAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::DeletePolicyAdminAction)]).await?;
let query = {
if let Some(query) = req.uri.query() {
@@ -461,22 +416,11 @@ pub struct SetPolicyForUserOrGroup {}
#[async_trait::async_trait]
impl Operation for SetPolicyForUserOrGroup {
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(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?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::AttachPolicyAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::AttachPolicyAdminAction)]).await?;
let query = {
if let Some(query) = req.uri.query() {
@@ -802,24 +746,13 @@ async fn collect_group_policy_mappings(
}
async fn handle_builtin_policy_entities(req: S3Request<Body>) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
let cred = authorize_admin_request(
&req,
vec![
Action::AdminAction(AdminAction::ListGroupsAdminAction),
Action::AdminAction(AdminAction::ListUsersAdminAction),
Action::AdminAction(AdminAction::ListUserPoliciesAdminAction),
],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
@@ -941,22 +874,7 @@ pub(crate) async fn handle_builtin_policy_association(
req: S3Request<Body>,
is_attach: bool,
) -> S3Result<S3Response<(StatusCode, Body)>> {
let Some(input_cred) = req.credentials else {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
validate_admin_request(
&req.headers,
&cred,
owner,
false,
vec![Action::AdminAction(AdminAction::AttachPolicyAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
)
.await?;
let cred = authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::AttachPolicyAdminAction)]).await?;
let req_path = req.uri.path().to_string();
let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, &req_path, &cred.secret_key).await?;
@@ -1129,13 +1047,33 @@ impl Operation for ListPolicyEntitiesBuiltin {
#[cfg(test)]
mod tests {
use super::{
GroupPolicyEntities, PolicyAssociationReq, SetPolicyForUserOrGroupQuery, UserPolicyEntities, attach_policy_names,
build_policy_mappings, detach_policy_names, direct_user_policy_names, parse_policy_entities_query,
validate_policy_association_req,
};
use super::*;
use http::Uri;
use rustfs_madmin::UserInfo;
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("an IAM policy request without credentials must fail");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(err.message(), Some("authentication required"));
}
#[test]
fn set_policy_query_supports_external_parameter_names() {
let query: SetPolicyForUserOrGroupQuery =
@@ -1244,4 +1182,93 @@ mod tests {
vec!["readonly".to_string(), "writeonly".to_string()]
);
}
#[tokio::test]
async fn policy_handlers_keep_their_missing_credentials_response() {
assert_missing_credentials(&ListCannedPolicies {}, Method::GET, "/rustfs/admin/v3/list-canned-policies").await;
assert_missing_credentials(&AddCannedPolicy {}, Method::PUT, "/rustfs/admin/v3/add-canned-policy").await;
assert_missing_credentials(&InfoCannedPolicy {}, Method::GET, "/rustfs/admin/v3/info-canned-policy").await;
assert_missing_credentials(&RemoveCannedPolicy {}, Method::DELETE, "/rustfs/admin/v3/remove-canned-policy").await;
assert_missing_credentials(&SetPolicyForUserOrGroup {}, Method::PUT, "/rustfs/admin/v3/set-user-or-group-policy").await;
for result in [
handle_builtin_policy_entities(credential_less_request(Method::GET, "/rustfs/admin/v3/idp/builtin/policy-entities"))
.await,
handle_builtin_policy_association(
credential_less_request(Method::POST, "/rustfs/admin/v3/idp/builtin/policy-association"),
true,
)
.await,
] {
let err = result.expect_err("the shared gate must reject missing credentials");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(err.message(), Some("get cred failed"));
}
}
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"
);
}
#[test]
fn policy_handlers_use_the_shared_admin_gate_with_their_actions() {
let production = include_str!("policies.rs")
.split("\n#[cfg(test)]\n")
.next()
.expect("production source must precede tests");
for (handler, action) in [
("ListCannedPolicies", "ListUserPoliciesAdminAction"),
("AddCannedPolicy", "CreatePolicyAdminAction"),
("InfoCannedPolicy", "GetPolicyAdminAction"),
("RemoveCannedPolicy", "DeletePolicyAdminAction"),
("SetPolicyForUserOrGroup", "AttachPolicyAdminAction"),
] {
let block = source_block(production, &format!("impl Operation for {handler}"));
assert_shared_gate_wiring(block, handler, &[action], false);
}
let entities = source_block(production, "async fn handle_builtin_policy_entities");
assert_shared_gate_wiring(
entities,
"handle_builtin_policy_entities",
&["ListGroupsAdminAction", "ListUsersAdminAction", "ListUserPoliciesAdminAction"],
true,
);
let association = source_block(production, "pub(crate) async fn handle_builtin_policy_association");
assert_shared_gate_wiring(association, "handle_builtin_policy_association", &["AttachPolicyAdminAction"], true);
assert!(!production.contains("check_key_valid(get_session_token"));
}
}