diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 544d1748c..620937fed 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -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> { + 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::(&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> { 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?; diff --git a/crates/iam/src/federation/model.rs b/crates/iam/src/federation/model.rs index e0ce0222d..97fe3cb69 100644 --- a/crates/iam/src/federation/model.rs +++ b/crates/iam/src/federation/model.rs @@ -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()); - } } diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index 9a1afff18..aa83a1b80 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -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>) -> Result> { + 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, groups: &Option>) -> Result> { 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); diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 76e0dfde3..47b58add2 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -872,6 +872,10 @@ impl IamSys { self.store.policy_db_get(name, groups).await } + pub async fn sts_policy_db_get(&self, name: &str, groups: &Option>) -> Result> { + 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); diff --git a/rustfs/src/admin/runtime_sources.rs b/rustfs/src/admin/runtime_sources.rs index 4d112e86d..20828a43a 100644 --- a/rustfs/src/admin/runtime_sources.rs +++ b/rustfs/src/admin/runtime_sources.rs @@ -32,6 +32,7 @@ pub(crate) use crate::runtime_sources::{ #[cfg(test)] pub(crate) use crate::runtime_sources::{ IamInterface, KmsInterface, NotificationSystemInterface, ServerConfigInterface, StorageClassInterface, + publish_test_app_context, }; use rustfs_config::server_config::Config; use rustfs_kms::KmsServiceManager; diff --git a/rustfs/src/admin/service/federated_identity.rs b/rustfs/src/admin/service/federated_identity.rs index 36f059cf4..69914c3cd 100644 --- a/rustfs/src/admin/service/federated_identity.rs +++ b/rustfs/src/admin/service/federated_identity.rs @@ -266,7 +266,7 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding { FederatedSessionBindingError::InvalidRequest("verified OIDC identity is missing issuer or subject".to_string()) })?; let selected_policy_names = match iam_store - .policy_db_get(&parent_user, &Some(authorization.groups.clone())) + .sts_policy_db_get(&parent_user, &Some(authorization.groups.clone())) .await .map_err(|_| FederatedSessionBindingError::Internal("failed to resolve OIDC policy mapping".to_string()))? { @@ -310,7 +310,13 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding { #[cfg(test)] mod tests { use super::*; + use crate::admin::runtime_sources::{AppContext, publish_test_app_context}; use rustfs_iam::federation::{FederatedAuthorization, FederatedClaims}; + use rustfs_iam::store::{Store, UserType, object::IAM_CONFIG_PREFIX}; + use rustfs_kms::KmsServiceManager; + use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq}; + use serial_test::serial; + use std::sync::Arc; fn transaction() -> FederatedSessionTransaction { FederatedSessionTransaction { @@ -348,14 +354,14 @@ mod tests { } #[test] - fn issued_credentials_and_replication_item_preserve_existing_shape() { + fn issued_credentials_and_replication_item_use_minio_parent_shape() { let transaction = transaction(); let secret = "federated-session-test-signing-secret"; let selected_policy_names = vec!["readonly".to_string()]; let credentials = issue_credentials(&transaction, &selected_policy_names, Some(secret)).expect("credential issuance should succeed"); - assert_eq!(credentials.parent_user, "openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I"); + assert_eq!(credentials.parent_user, "TwyekekG2eMes0qk9Tgh7KXEitwGi1z2W1f2KccrXGA"); assert_eq!(credentials.groups, Some(vec!["devs".to_string()])); let claims = rustfs_iam::sys::get_claims_from_token_with_secret(&credentials.session_token, secret) @@ -364,11 +370,11 @@ mod tests { assert_eq!(claims.get("oidc_provider"), Some(&serde_json::json!("default"))); assert_eq!( claims.get("parent"), - Some(&serde_json::json!("openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I")) + Some(&serde_json::json!("TwyekekG2eMes0qk9Tgh7KXEitwGi1z2W1f2KccrXGA")) ); assert_eq!( claims.get(OIDC_VIRTUAL_PARENT_CLAIM), - Some(&serde_json::json!("openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I")) + Some(&serde_json::json!("TwyekekG2eMes0qk9Tgh7KXEitwGi1z2W1f2KccrXGA")) ); assert_eq!(claims.get("policy"), Some(&serde_json::json!("readonly"))); assert_eq!(claims.get("groups"), Some(&serde_json::json!(["devs"]))); @@ -384,13 +390,77 @@ mod tests { assert_eq!(replicated.access_key, credentials.access_key); assert_eq!(replicated.secret_key, credentials.secret_key); assert_eq!(replicated.session_token, credentials.session_token); - assert_eq!(replicated.parent_user, "openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I"); + assert_eq!(replicated.parent_user, "TwyekekG2eMes0qk9Tgh7KXEitwGi1z2W1f2KccrXGA"); assert_eq!(replicated.parent_policy_mapping, OIDC_STS_REQUIRES_VIRTUAL_PARENT_RECEIVER_POLICY); assert!(replicated.parent_policy_mapping.trim().is_empty()); assert!(MappedPolicy::new(&replicated.parent_policy_mapping).to_slice().is_empty()); assert_eq!(replicated.api_version.as_deref(), Some(SITE_REPL_API_VERSION)); } + #[tokio::test] + #[serial] + async fn binding_uses_sts_policy_when_regular_mapping_collides() { + let _ = rustfs_credentials::init_global_action_credentials( + Some("TESTROOTACCESSKEY".to_string()), + Some("TESTROOTSECRET123".to_string()), + ); + if current_ready_iam_handle().is_err() { + let env = rustfs_test_utils::TestECStoreEnv::builder() + .prefix("federated_binding_sts_policy") + .disk_count(1) + .init_bucket_metadata(false) + .build() + .await; + ObjectStore::new(Arc::clone(&env.ecstore)) + .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)) + .await + .expect("seed IAM format"); + let iam = rustfs_iam::build_iam_sys(Arc::clone(&env.ecstore)) + .await + .expect("build test IAM"); + publish_test_app_context(Arc::new(AppContext::with_default_interfaces( + env.ecstore, + iam, + Arc::new(KmsServiceManager::new()), + ))); + } + + let iam = current_ready_iam_handle().expect("test IAM should be ready"); + let mut transaction = transaction(); + transaction.authorization.claims.sub = "binding-sts-policy-subject".to_string(); + transaction.authorization.policies = vec!["readwrite".to_string()]; + let parent = transaction + .authorization + .oidc_virtual_parent() + .expect("test authorization should have a virtual parent"); + iam.create_user( + &parent, + &AddOrUpdateUserReq { + secret_key: "regular-user-secret".to_string(), + policy: None, + status: AccountStatus::Enabled, + }, + ) + .await + .expect("create colliding regular user"); + iam.policy_db_set(&parent, UserType::Reg, false, "readonly") + .await + .expect("store colliding regular policy mapping"); + iam.policy_db_set(&parent, UserType::Sts, false, "writeonly") + .await + .expect("store STS policy mapping"); + + let credentials = DefaultFederatedSessionBinding + .bind(&transaction) + .await + .expect("production binding should issue credentials"); + let signing_key = current_token_signing_key().expect("test signing key should be initialized"); + let claims = rustfs_iam::sys::get_claims_from_token_with_secret(&credentials.session_token, &signing_key) + .expect("issued session token should verify"); + + assert_eq!(claims.get("policy"), Some(&serde_json::json!("writeonly"))); + } + #[test] fn invalid_session_policy_precedes_missing_signing_key() { let mut transaction = transaction(); diff --git a/rustfs/src/runtime_sources.rs b/rustfs/src/runtime_sources.rs index 9f8445535..1b8394686 100644 --- a/rustfs/src/runtime_sources.rs +++ b/rustfs/src/runtime_sources.rs @@ -59,6 +59,8 @@ pub(crate) fn set_test_outbound_tls_generation(generation: u64) { #[cfg(test)] pub(crate) use context::install_test_app_context; +#[cfg(test)] +pub(crate) use context::publish_global_app_context as publish_test_app_context; #[cfg(test)] pub(crate) use context::{IamInterface, KmsInterface, NotificationSystemInterface, ServerConfigInterface, StorageClassInterface};