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

fix(iam): align OIDC parent IDs with MinIO (#5290)

Browse Source 
* fix(iam): align OIDC parent IDs with MinIO

* test(iam): cover OIDC STS binding policy lookup

* test(admin): use runtime facade for AppContext setup

* test(ilm): wait for lifecycle backfill before manual failure

* test(ilm): make overlapping admission check concurrent
This commit is contained in:
GatewayJ
2026-07-27 15:16:22 +08:00
committed by GitHub
parent e076e8cc6e
commit 24cf2cdb78
7 changed files with 259 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/reliant/tiering.rs
+94 -26
View File
@@ -382,6 +382,7 @@ struct ManualTransitionRunReport {
skipped_delete_marker: u64,
skipped_directory: u64,
skipped_replication: u64,
skipped_already_transitioned: u64,
skipped_already_in_flight: u64,
skipped_queue_full: u64,
skipped_queue_closed: u64,
@@ -411,6 +412,23 @@ struct ManualTransitionQueueSnapshot {
compensation_running: u64,
}
#[derive(Debug, Deserialize)]
struct ScannerStatusResponse {
metrics: ScannerStatusMetrics,
}
#[derive(Debug, Deserialize)]
struct ScannerStatusMetrics {
lifecycle_transition: ScannerTransitionQueueState,
}
#[derive(Debug, Deserialize)]
struct ScannerTransitionQueueState {
current_queued: u64,
current_active: u64,
failed: u64,
}
#[derive(Debug, Deserialize)]
struct ManualTransitionJobStatusResponse {
job_id: String,
@@ -621,6 +639,41 @@ async fn wait_for_manual_transition_job_running(
}
}
async fn scanner_transition_queue_state(
hot: &RustFSTestEnvironment,
) -> Result<ScannerTransitionQueueState, Box<dyn std::error::Error + Send + Sync>> {
let path = "/rustfs/admin/v3/scanner/status";
let (status, body) = signed_admin_request(&hot.url, Method::GET, path, None, &hot.access_key, &hot.secret_key).await?;
if !status.is_success() {
return Err(format!("scanner status failed: status={status}, body={body}").into());
}
Ok(serde_json::from_str::<ScannerStatusResponse>(&body)?
.metrics
.lifecycle_transition)
}
async fn wait_for_transition_failure_and_idle(
hot: &RustFSTestEnvironment,
failed_before: u64,
deadline: StdDuration,
) -> TestResult {
let start = Instant::now();
loop {
let state = scanner_transition_queue_state(hot).await?;
if state.failed > failed_before && state.current_queued == 0 && state.current_active == 0 {
return Ok(());
}
if start.elapsed() >= deadline {
return Err(format!(
"transition queue did not report a new failure and become idle within {}s; failed_before={failed_before}, last={state:#?}",
deadline.as_secs()
)
.into());
}
tokio::time::sleep(StdDuration::from_millis(100)).await;
}
}
/// Number of objects currently stored in the cold-tier bucket.
async fn cold_tier_object_count(cold_client: &Client) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
let resp = cold_client.list_objects_v2().bucket(TIER_BUCKET).send().await?;
@@ -1200,14 +1253,34 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_
.await?;
let before_remote_count = cold_tier_object_count(&cold_client).await?;
let accepted = manual_transition_async_run(
&hot,
MANUAL_ASYNC_CONFLICT_BUCKET,
MANUAL_ASYNC_CONFLICT_PREFIX,
false,
MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
)
.await?;
let (parent, nested) = tokio::join!(
manual_transition_async_run_raw(
&hot,
MANUAL_ASYNC_CONFLICT_BUCKET,
MANUAL_ASYNC_CONFLICT_PREFIX,
false,
MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
),
manual_transition_async_run_raw(
&hot,
MANUAL_ASYNC_CONFLICT_BUCKET,
MANUAL_ASYNC_CONFLICT_NESTED_PREFIX,
false,
MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
)
);
let responses = [parent?, nested?];
let accepted = responses
.iter()
.find(|(status, _)| *status == reqwest::StatusCode::ACCEPTED)
.ok_or("one concurrent overlapping-scope async run must be accepted")?;
let conflict = responses
.iter()
.find(|(status, _)| *status == reqwest::StatusCode::CONFLICT)
.ok_or("one concurrent overlapping-scope async run must report conflict")?;
let accepted: ManualTransitionRunResponse = serde_json::from_str(&accepted.1)?;
let conflict: ManualTransitionJobConflictResponse = serde_json::from_str(&conflict.1)?;
let job_id = accepted
.job_id
.as_deref()
@@ -1224,22 +1297,14 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_
assert_eq!(accepted.state, "accepted");
assert_eq!(accepted.mode, "durable_job");
let active = wait_for_manual_transition_job_running(&hot, status_endpoint, MANUAL_ACTIVE_CANCEL_RUNNING_TIMEOUT).await?;
assert_eq!(active.job_id, job_id);
let (conflict_status, conflict_body) = manual_transition_async_run_raw(
&hot,
MANUAL_ASYNC_CONFLICT_BUCKET,
MANUAL_ASYNC_CONFLICT_NESTED_PREFIX,
false,
MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
)
.await?;
assert_eq!(
conflict_status,
reqwest::StatusCode::CONFLICT,
"overlapping async run response: {conflict_body}"
assert_eq!(accepted.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET);
assert!(
matches!(
accepted.report.prefix.as_str(),
MANUAL_ASYNC_CONFLICT_PREFIX | MANUAL_ASYNC_CONFLICT_NESTED_PREFIX
),
"overlapping async winner prefix: {accepted:#?}"
);
let conflict: ManualTransitionJobConflictResponse = serde_json::from_str(&conflict_body)?;
assert_eq!(conflict.state, "conflict");
assert_eq!(conflict.mode, "durable_job");
assert_eq!(conflict.active_job_id, job_id);
@@ -1252,19 +1317,20 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_
assert_eq!(terminal.status, "completed", "terminal conflict winner response: {terminal:#?}");
assert!(!terminal.report.dry_run);
assert_eq!(terminal.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET);
assert_eq!(terminal.report.prefix, MANUAL_ASYNC_CONFLICT_PREFIX);
assert_eq!(terminal.report.prefix, accepted.report.prefix);
assert_eq!(
terminal.report.scanned, MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
"terminal conflict winner response: {terminal:#?}"
);
assert_eq!(
terminal.report.eligible, MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
terminal.report.eligible + terminal.report.skipped_already_transitioned,
MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
"terminal conflict winner response: {terminal:#?}"
);
assert_eq!(terminal.report.dry_run_eligible, 0, "terminal conflict winner response: {terminal:#?}");
assert_eq!(
terminal.report.enqueued + terminal.report.skipped_already_in_flight,
MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
terminal.report.eligible,
"terminal conflict winner response: {terminal:#?}"
);
assert_eq!(
@@ -1647,6 +1713,7 @@ async fn test_manual_transition_async_worker_failure_reports_terminal_partial()
due_mtime,
)
.await?;
let transition_failures_before = scanner_transition_queue_state(&hot).await?.failed;
put_lifecycle_transition_rule(
&hot_client,
MANUAL_WORKER_FAILURE_BUCKET,
@@ -1655,6 +1722,7 @@ async fn test_manual_transition_async_worker_failure_reports_terminal_partial()
0,
)
.await?;
wait_for_transition_failure_and_idle(&hot, transition_failures_before, StdDuration::from_secs(30)).await?;
let accepted =
manual_transition_async_run(&hot, MANUAL_WORKER_FAILURE_BUCKET, MANUAL_WORKER_FAILURE_PREFIX, false, 10).await?;
crates/iam/src/federation/model.rs
+7 -31
View File
@@ -18,7 +18,6 @@ use serde_json::Value;
use std::collections::HashMap;
pub const OIDC_VIRTUAL_PARENT_CLAIM: &str = "x-rustfs-internal-oidc-parent";
const OIDC_VIRTUAL_PARENT_PREFIX: &str = "openid=";
#[derive(Debug, Clone)]
pub struct FederatedClaims {
@@ -65,18 +64,13 @@ impl FederatedAuthorization {
return None;
}
let subject_len = u64::try_from(subject.len()).ok()?;
let issuer_len = u64::try_from(issuer.len()).ok()?;
let mut source = Vec::with_capacity(16 + subject.len() + issuer.len());
source.extend_from_slice(&subject_len.to_be_bytes());
let mut source = Vec::with_capacity(8 + subject.len() + issuer.len());
source.extend_from_slice(b"openid:");
source.extend_from_slice(subject.as_bytes());
source.extend_from_slice(&issuer_len.to_be_bytes());
source.push(b':');
source.extend_from_slice(issuer.as_bytes());
let digest = HashAlgorithm::SHA256.hash_encode(&source);
Some(format!(
"{OIDC_VIRTUAL_PARENT_PREFIX}{}",
base64_simd::URL_SAFE_NO_PAD.encode_to_string(digest.as_ref())
))
Some(base64_simd::URL_SAFE_NO_PAD.encode_to_string(digest.as_ref()))
}
}
@@ -151,7 +145,7 @@ mod tests {
}
#[test]
fn oidc_virtual_parent_is_stable_and_issuer_scoped() {
fn oidc_virtual_parent_matches_minio_and_is_issuer_scoped() {
let first = authorization(Vec::new(), Vec::new());
let mut second = first.clone();
second
@@ -161,9 +155,9 @@ mod tests {
assert_eq!(
first.oidc_virtual_parent().as_deref(),
Some("openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I")
Some("TwyekekG2eMes0qk9Tgh7KXEitwGi1z2W1f2KccrXGA")
);
assert!(rustfs_policy::auth::contains_reserved_chars(
assert!(!rustfs_policy::auth::contains_reserved_chars(
first.oidc_virtual_parent().as_deref().expect("virtual parent")
));
assert_ne!(first.oidc_virtual_parent(), second.oidc_virtual_parent());
@@ -188,22 +182,4 @@ mod tests {
assert_ne!(plain.oidc_virtual_parent(), padded.oidc_virtual_parent());
}
#[test]
fn oidc_virtual_parent_encoding_is_unambiguous() {
let mut first = authorization(Vec::new(), Vec::new());
first.claims.sub = "subject".to_string();
first.claims.raw.insert(
"iss".to_string(),
Value::String("https://issuer.example/path:https://other.example".to_string()),
);
let mut second = authorization(Vec::new(), Vec::new());
second.claims.sub = "subject:https://issuer.example/path".to_string();
second
.claims
.raw
.insert("iss".to_string(), Value::String("https://other.example".to_string()));
assert_ne!(first.oidc_virtual_parent(), second.oidc_virtual_parent());
}
}
crates/iam/src/manager.rs
+62 -1
View File
@@ -907,7 +907,41 @@ where
return Err(Error::InvalidArgument);
}
let (mut policies, _) = self.policy_db_get_internal(name, false, false).await?;
let (policies, _) = self.policy_db_get_internal(name, false, false).await?;
self.policy_db_get_with_groups(policies, groups).await
}
pub async fn sts_policy_db_get(&self, name: &str, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
if name.is_empty() {
return Err(Error::InvalidArgument);
}
let cache = self.cache.snapshot();
let sts_policies = Arc::clone(&cache.sts_policies);
drop(cache);
let mapped_policy = match sts_policies.get(name) {
Some(policy) => policy.clone(),
None => {
let mut policies = HashMap::new();
if let Err(err) = self.api.load_mapped_policy(name, UserType::Sts, false, &mut policies).await
&& !is_err_no_such_policy(&err)
{
return Err(err);
}
match policies.get(name) {
Some(policy) => {
self.cache.add_or_update_sts_policy(name, policy, OffsetDateTime::now_utc());
policy.clone()
}
None => MappedPolicy::default(),
}
}
};
self.policy_db_get_with_groups(mapped_policy.to_slice(), groups).await
}
async fn policy_db_get_with_groups(&self, mut policies: Vec<String>, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
let present = !policies.is_empty();
if let Some(groups) = groups {
@@ -2585,6 +2619,33 @@ mod tests {
}
}
#[tokio::test]
async fn sts_policy_lookup_ignores_colliding_regular_user_mapping() {
let iam = build_test_iam_cache(FailingInitialLoadStore);
let parent = "TwyekekG2eMes0qk9Tgh7KXEitwGi1z2W1f2KccrXGA";
let user = UserIdentity::new(Credentials {
access_key: parent.to_string(),
secret_key: "regular-user-secret".to_string(),
status: STATUS_ENABLED.to_string(),
..Default::default()
});
let now = OffsetDateTime::now_utc();
iam.cache.add_or_update_user(parent, &user, now);
iam.cache
.add_or_update_user_policy(parent, &MappedPolicy::new("regular-policy"), now);
iam.cache
.add_or_update_sts_policy(parent, &MappedPolicy::new("oidc-policy"), now);
assert_eq!(
iam.policy_db_get(parent, &None).await.expect("regular lookup should succeed"),
vec!["regular-policy"]
);
assert_eq!(
iam.sts_policy_db_get(parent, &None).await.expect("STS lookup should succeed"),
vec!["oidc-policy"]
);
}
#[tokio::test]
async fn set_temp_user_retries_until_sts_identity_becomes_visible() {
let store = DelayedTempUserVisibilityStore::new(2);
crates/iam/src/sys.rs
+17 -1
View File
@@ -872,6 +872,10 @@ impl<T: Store> IamSys<T> {
self.store.policy_db_get(name, groups).await
}
pub async fn sts_policy_db_get(&self, name: &str, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
self.store.sts_policy_db_get(name, groups).await
}
/// Check whether a policy name from a JWT claim is safe to resolve against the IAM store.
///
/// Allowed characters: `[a-zA-Z0-9_:.-]`
@@ -2776,7 +2780,7 @@ mod tests {
async fn oidc_service_account_uses_verified_policy_and_persisted_boundary() {
ensure_test_global_credentials();
let iam_sys = IamSys::new(IamCache::new(StsTestMockStore::new(true)).await.unwrap());
let parent_user = "openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I";
let parent_user = "TwyekekG2eMes0qk9Tgh7KXEitwGi1z2W1f2KccrXGA";
let mut oidc_claims = HashMap::from([
("iss".to_string(), Value::String("rustfs-oidc".to_string())),
("oidc_provider".to_string(), Value::String("default".to_string())),
@@ -3466,6 +3470,18 @@ mod tests {
);
}
#[tokio::test]
async fn test_sts_policy_lookup_loads_missing_mapping_without_regular_user() {
let store = StsTestMockStore::new(false);
let cache_manager = IamCache::new(store).await.unwrap();
let iam_sys = IamSys::new(cache_manager);
let policies = iam_sys.sts_policy_db_get("notify-sts-parent", &None).await.unwrap();
assert_eq!(policies, vec!["readwrite"]);
assert!(iam_sys.store.cache.snapshot().sts_policies.contains_key("notify-sts-parent"));
}
#[tokio::test]
async fn test_missing_user_notification_cleans_related_cache_state() {
let store = StsTestMockStore::new(false);