diff --git a/crates/iam/src/cache.rs b/crates/iam/src/cache.rs index f0ff24d30..271786c19 100644 --- a/crates/iam/src/cache.rs +++ b/crates/iam/src/cache.rs @@ -16,7 +16,7 @@ use std::{ collections::{HashMap, HashSet}, ops::{Deref, DerefMut}, ptr, - sync::{Arc, Mutex}, + sync::{Arc, Mutex, Weak}, }; use arc_swap::{ArcSwap, Guard}; @@ -65,6 +65,29 @@ pub struct Cache { state: ArcSwap, write_lock: Mutex<()>, service_account_mutation_lock: AsyncMutex<()>, + sts_account_mutation_locks: Arc, +} + +struct StsMutationLockRegistry { + locks: Mutex>>>, +} + +pub(crate) struct StsMutationLockState { + access_key: String, + registry: Weak, + lock: Weak>, +} + +impl Drop for StsMutationLockState { + fn drop(&mut self) { + let Some(registry) = self.registry.upgrade() else { + return; + }; + let mut locks = registry.locks.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if locks.get(&self.access_key).is_some_and(|current| current.ptr_eq(&self.lock)) { + locks.remove(&self.access_key); + } + } } impl Default for Cache { @@ -73,6 +96,9 @@ impl Default for Cache { state: ArcSwap::new(Arc::new(CacheState::default())), write_lock: Mutex::new(()), service_account_mutation_lock: AsyncMutex::new(()), + sts_account_mutation_locks: Arc::new(StsMutationLockRegistry { + locks: Mutex::new(HashMap::new()), + }), } } } @@ -84,6 +110,29 @@ impl Cache { &self.service_account_mutation_lock } + pub(crate) fn sts_account_mutation_lock(&self, access_key: &str) -> Arc> { + let mut locks = self + .sts_account_mutation_locks + .locks + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(lock) = locks.get(access_key).and_then(Weak::upgrade) { + return lock; + } + + let registry = Arc::downgrade(&self.sts_account_mutation_locks); + let access_key_owned = access_key.to_string(); + let lock = Arc::new_cyclic(|lock| { + AsyncMutex::new(StsMutationLockState { + access_key: access_key_owned, + registry, + lock: lock.clone(), + }) + }); + locks.insert(access_key.to_string(), Arc::downgrade(&lock)); + lock + } + pub(crate) fn snapshot(&self) -> CacheSnapshot { self.state.load() } @@ -445,6 +494,30 @@ mod tests { use crate::cache::Cache; use crate::store::MappedPolicy; + #[test] + fn sts_mutation_locks_are_keyed_and_prune_unused_entries() { + let cache = Cache::default(); + let first = cache.sts_account_mutation_lock("first"); + let same = cache.sts_account_mutation_lock("first"); + let different = cache.sts_account_mutation_lock("different"); + + assert!(Arc::ptr_eq(&first, &same)); + assert!(!Arc::ptr_eq(&first, &different)); + drop(first); + drop(same); + drop(different); + + let _next = cache.sts_account_mutation_lock("next"); + let locks = cache + .sts_account_mutation_locks + .locks + .lock() + .expect("STS mutation lock registry mutex poisoned"); + assert!(!locks.contains_key("first")); + assert!(!locks.contains_key("different")); + assert!(locks.contains_key("next")); + } + #[tokio::test] async fn test_cache_entity_add() { let owner = Arc::new(Cache::default()); diff --git a/crates/iam/src/lib.rs b/crates/iam/src/lib.rs index e4ac0662a..a487942d0 100644 --- a/crates/iam/src/lib.rs +++ b/crates/iam/src/lib.rs @@ -102,7 +102,49 @@ pub(crate) async fn notify_iam_delete_user(access_key: &str) -> Vec>, + pub(crate) remaining_failures: std::sync::atomic::AtomicUsize, + pub(crate) attempts: std::sync::atomic::AtomicUsize, + pub(crate) panic: bool, + pub(crate) started: tokio::sync::Notify, + pub(crate) release: Option, + pub(crate) completed: tokio::sync::Notify, +} + +#[cfg(test)] +tokio::task_local! { + pub(crate) static LOAD_USER_NOTIFICATION_PROBE: std::sync::Arc; +} + pub(crate) async fn notify_iam_load_user(access_key: &str, temp: bool) -> Vec { + #[cfg(test)] + if let Ok(probe) = LOAD_USER_NOTIFICATION_PROBE.try_with(std::sync::Arc::clone) { + probe.attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + *probe.observed.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some((access_key.to_string(), temp)); + probe.started.notify_one(); + if let Some(release) = &probe.release { + release.notified().await; + } + assert!(!probe.panic, "notification probe panic"); + let should_fail = probe + .remaining_failures + .fetch_update(std::sync::atomic::Ordering::SeqCst, std::sync::atomic::Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok(); + let result = if should_fail { + vec![IamNotificationPeerErr { + err: Some(IamEcstoreError::other("peer notification failed")), + }] + } else { + Vec::new() + }; + probe.completed.notify_one(); + return result; + } + match runtime_sources::notification_sys() { Some(notification_sys) => notification_sys .load_user(access_key, temp) diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index 3acd8bdab..d7831fc89 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -293,6 +293,8 @@ where } pub async fn load_user(&self, access_key: &str) -> Result<()> { + let sts_mutation_lock = self.cache.sts_account_mutation_lock(access_key); + let _sts_mutation_guard = sts_mutation_lock.lock().await; let mut users_map: HashMap = HashMap::new(); let mut user_policy_map = HashMap::new(); let mut sts_users_map = HashMap::new(); @@ -1207,6 +1209,9 @@ where return Err(Error::InvalidArgument); } + let mutation_lock = self.cache.sts_account_mutation_lock(access_key); + let _mutation_guard = mutation_lock.lock().await; + let sts_policy_update = if let Some(policy) = policy_name { let mp = MappedPolicy::new(policy); let (_, combined_policy_stmt) = filter_policies(&self.cache, &mp.policies, "temp"); @@ -1433,6 +1438,11 @@ where return Err(Error::InvalidArgument); } + let sts_mutation_lock = (utype == UserType::Sts).then(|| self.cache.sts_account_mutation_lock(access_key)); + let _sts_mutation_guard = match &sts_mutation_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; let _service_account_guard = if utype == UserType::Svc { Some(self.cache.service_account_mutation_lock().lock().await) } else { @@ -1493,9 +1503,13 @@ where }); } - let _ = self.api.delete_mapped_policy(access_key, utype, false).await; + if utype != UserType::Sts { + let _ = self.api.delete_mapped_policy(access_key, utype, false).await; + } - self.cache.delete_user_policy(access_key, OffsetDateTime::now_utc()); + if utype != UserType::Sts { + self.cache.delete_user_policy(access_key, OffsetDateTime::now_utc()); + } if let Err(err) = self.api.delete_user_identity(access_key, utype).await && !is_err_no_such_user(&err) @@ -1507,8 +1521,17 @@ where self.cache.with_write_lock(|cache| { if utype == UserType::Sts { cache.delete_sts_account(access_key, deleted_at); + if cache + .state() + .users + .get(access_key) + .is_some_and(|identity| identity.credentials.is_temp()) + { + cache.delete_user(access_key, deleted_at); + } + } else { + cache.delete_user(access_key, deleted_at); } - cache.delete_user(access_key, deleted_at); }); Ok(deleted_at) @@ -2032,6 +2055,11 @@ where Ok(()) } pub async fn user_notification_handler(&self, name: &str, user_type: UserType) -> Result<()> { + let sts_mutation_lock = (user_type == UserType::Sts).then(|| self.cache.sts_account_mutation_lock(name)); + let _sts_mutation_guard = match &sts_mutation_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; let _service_account_guard = if user_type == UserType::Svc { Some(self.cache.service_account_mutation_lock().lock().await) } else { @@ -2077,7 +2105,9 @@ where UserType::Reg | UserType::Svc => cache.delete_user(name, now), UserType::None => {} } - self.remove_user_from_cached_groups(cache, name, now); + if user_type != UserType::Sts { + self.remove_user_from_cached_groups(cache, name, now); + } if user_type == UserType::Reg { for access_key in service_accounts_to_delete.iter() { cache.delete_user(access_key, now); @@ -2087,7 +2117,9 @@ where cache.delete_user(access_key, now); } } - cache.delete_user_policy(name, now); + if user_type != UserType::Sts { + cache.delete_user_policy(name, now); + } }); return Ok(()); @@ -2446,12 +2478,12 @@ mod tests { saved_user: Arc>>, load_attempts: Arc, visible_after_attempt: usize, - block_service_save: Arc, - service_save_started: Arc, - release_service_save: Arc, - block_service_load: Arc, - service_load_started: Arc, - release_service_load: Arc, + block_account_save: Arc, + account_save_started: Arc, + release_account_save: Arc, + block_account_load: Arc, + account_load_started: Arc, + release_account_load: Arc, } impl DelayedTempUserVisibilityStore { @@ -2460,12 +2492,12 @@ mod tests { saved_user: Arc::new(Mutex::new(None)), load_attempts: Arc::new(AtomicUsize::new(0)), visible_after_attempt, - block_service_save: Arc::new(AtomicBool::new(false)), - service_save_started: Arc::new(Notify::new()), - release_service_save: Arc::new(Notify::new()), - block_service_load: Arc::new(AtomicBool::new(false)), - service_load_started: Arc::new(Notify::new()), - release_service_load: Arc::new(Notify::new()), + block_account_save: Arc::new(AtomicBool::new(false)), + account_save_started: Arc::new(Notify::new()), + release_account_save: Arc::new(Notify::new()), + block_account_load: Arc::new(AtomicBool::new(false)), + account_load_started: Arc::new(Notify::new()), + release_account_load: Arc::new(Notify::new()), } } } @@ -2495,9 +2527,9 @@ mod tests { item: UserIdentity, _ttl: Option, ) -> Result<()> { - if user_type == UserType::Svc && self.block_service_save.load(Ordering::SeqCst) { - self.service_save_started.notify_one(); - self.release_service_save.notified().await; + if matches!(user_type, UserType::Svc | UserType::Sts) && self.block_account_save.load(Ordering::SeqCst) { + self.account_save_started.notify_one(); + self.release_account_save.notified().await; } *self.saved_user.lock().expect("saved_user mutex poisoned") = Some(item); Ok(()) @@ -2528,9 +2560,18 @@ mod tests { .expect("saved_user mutex poisoned") .clone() .ok_or_else(|| Error::NoSuchUser(name.to_string()))?; - if user_type == UserType::Svc && self.block_service_load.load(Ordering::SeqCst) { - self.service_load_started.notify_one(); - self.release_service_load.notified().await; + let matches_user_type = match user_type { + UserType::Sts => loaded.credentials.is_temp(), + UserType::Svc => loaded.credentials.is_service_account(), + UserType::Reg => !loaded.credentials.is_temp() && !loaded.credentials.is_service_account(), + UserType::None => false, + }; + if !matches_user_type { + return Err(Error::NoSuchUser(name.to_string())); + } + if self.block_account_load.load(Ordering::SeqCst) { + self.account_load_started.notify_one(); + self.release_account_load.notified().await; } m.insert(name.to_string(), loaded); Ok(()) @@ -2746,12 +2787,12 @@ mod tests { }; cache.add_service_account(credentials).await.expect("seed service account"); - store.block_service_load.store(true, Ordering::SeqCst); + store.block_account_load.store(true, Ordering::SeqCst); let notification = { let cache = Arc::clone(&cache); tokio::spawn(async move { cache.user_notification_handler(access_key, UserType::Svc).await }) }; - store.service_load_started.notified().await; + store.account_load_started.notified().await; let update = { let cache = Arc::clone(&cache); @@ -2776,7 +2817,7 @@ mod tests { tokio::task::yield_now().await; assert!(!update.is_finished(), "update must wait for the in-flight cache refresh"); - store.release_service_load.notify_one(); + store.release_account_load.notify_one(); notification.await.expect("notification task").expect("notification refresh"); update.await.expect("update task").expect("service account update"); @@ -2793,7 +2834,7 @@ mod tests { #[tokio::test] async fn concurrent_service_account_create_cannot_overwrite_first_writer() { let store = DelayedTempUserVisibilityStore::new(0); - store.block_service_save.store(true, Ordering::SeqCst); + store.block_account_save.store(true, Ordering::SeqCst); let cache = Arc::new(build_test_iam_cache(store.clone())); let access_key = "SERIALIZEDSERVICE00"; let credentials = |secret_key: &str| Credentials { @@ -2808,7 +2849,7 @@ mod tests { let cache = Arc::clone(&cache); tokio::spawn(async move { cache.add_service_account(credentials("firstServiceSecret123")).await }) }; - store.service_save_started.notified().await; + store.account_save_started.notified().await; let second = { let cache = Arc::clone(&cache); @@ -2817,8 +2858,8 @@ mod tests { tokio::task::yield_now().await; assert!(!second.is_finished(), "second create must wait for the first writer"); - store.block_service_save.store(false, Ordering::SeqCst); - store.release_service_save.notify_waiters(); + store.block_account_save.store(false, Ordering::SeqCst); + store.release_account_save.notify_waiters(); first.await.expect("first create task").expect("first create"); let err = second .await @@ -2862,12 +2903,12 @@ mod tests { }; cache.add_service_account(credentials).await.expect("seed service account"); - store.block_service_load.store(true, Ordering::SeqCst); + store.block_account_load.store(true, Ordering::SeqCst); let notification = { let cache = Arc::clone(&cache); tokio::spawn(async move { cache.user_notification_handler(access_key, UserType::Svc).await }) }; - store.service_load_started.notified().await; + store.account_load_started.notified().await; let delete = { let cache = Arc::clone(&cache); @@ -2876,7 +2917,7 @@ mod tests { tokio::task::yield_now().await; assert!(!delete.is_finished(), "delete must wait for the in-flight cache refresh"); - store.release_service_load.notify_one(); + store.release_account_load.notify_one(); notification.await.expect("notification task").expect("notification refresh"); delete.await.expect("delete task").expect("service account delete"); @@ -2884,6 +2925,130 @@ mod tests { assert!(store.saved_user.lock().expect("saved_user mutex poisoned").is_none()); } + #[tokio::test] + async fn sts_notification_cannot_restore_concurrent_delete() { + let store = DelayedTempUserVisibilityStore::new(0); + let cache = Arc::new(build_test_iam_cache(store.clone())); + let credentials = build_test_temp_credentials(); + let access_key = credentials.access_key.clone(); + cache + .set_temp_user(&access_key, &credentials, None) + .await + .expect("seed temporary account"); + + store.block_account_load.store(true, Ordering::SeqCst); + let notification = { + let cache = Arc::clone(&cache); + let access_key = access_key.clone(); + tokio::spawn(async move { cache.user_notification_handler(&access_key, UserType::Sts).await }) + }; + store.account_load_started.notified().await; + + let delete_started = Arc::new(Notify::new()); + let delete = { + let cache = Arc::clone(&cache); + let access_key = access_key.clone(); + let delete_started = Arc::clone(&delete_started); + tokio::spawn(async move { + delete_started.notify_one(); + cache.delete_user(&access_key, UserType::Sts).await + }) + }; + delete_started.notified().await; + tokio::task::yield_now().await; + let delete_waited_for_notification = !delete.is_finished(); + + store.release_account_load.notify_one(); + notification.await.expect("notification task").expect("notification refresh"); + delete.await.expect("delete task").expect("temporary account delete"); + + assert!(delete_waited_for_notification, "delete must wait for the in-flight STS cache refresh"); + assert!(!cache.cache.snapshot().sts_accounts.contains_key(&access_key)); + assert!(store.saved_user.lock().expect("saved_user mutex poisoned").is_none()); + } + + #[tokio::test] + async fn sts_auth_reload_cannot_restore_concurrent_delete() { + let store = DelayedTempUserVisibilityStore::new(0); + let cache = Arc::new(build_test_iam_cache(store.clone())); + let credentials = build_test_temp_credentials(); + let access_key = credentials.access_key.clone(); + cache + .set_temp_user(&access_key, &credentials, None) + .await + .expect("seed temporary account"); + cache.cache.delete_sts_account(&access_key, OffsetDateTime::now_utc()); + + store.block_account_load.store(true, Ordering::SeqCst); + let reload = { + let cache = Arc::clone(&cache); + let access_key = access_key.clone(); + tokio::spawn(async move { cache.load_user(&access_key).await }) + }; + store.account_load_started.notified().await; + + let delete_started = Arc::new(Notify::new()); + let delete = { + let cache = Arc::clone(&cache); + let access_key = access_key.clone(); + let delete_started = Arc::clone(&delete_started); + tokio::spawn(async move { + delete_started.notify_one(); + cache.delete_user(&access_key, UserType::Sts).await + }) + }; + delete_started.notified().await; + tokio::task::yield_now().await; + let delete_waited_for_reload = !delete.is_finished(); + + store.release_account_load.notify_one(); + reload.await.expect("reload task").expect("authentication cache reload"); + delete.await.expect("delete task").expect("temporary account delete"); + + assert!(delete_waited_for_reload, "delete must wait for the in-flight authentication reload"); + assert!(!cache.cache.snapshot().sts_accounts.contains_key(&access_key)); + assert!(store.saved_user.lock().expect("saved_user mutex poisoned").is_none()); + } + + #[tokio::test] + async fn sts_create_cannot_restore_concurrent_delete() { + let store = DelayedTempUserVisibilityStore::new(0); + store.block_account_save.store(true, Ordering::SeqCst); + let cache = Arc::new(build_test_iam_cache(store.clone())); + let credentials = build_test_temp_credentials(); + let access_key = credentials.access_key.clone(); + + let create = { + let cache = Arc::clone(&cache); + let access_key = access_key.clone(); + tokio::spawn(async move { cache.set_temp_user(&access_key, &credentials, None).await }) + }; + store.account_save_started.notified().await; + + let delete_started = Arc::new(Notify::new()); + let delete = { + let cache = Arc::clone(&cache); + let access_key = access_key.clone(); + let delete_started = Arc::clone(&delete_started); + tokio::spawn(async move { + delete_started.notify_one(); + cache.delete_user(&access_key, UserType::Sts).await + }) + }; + delete_started.notified().await; + tokio::task::yield_now().await; + let delete_waited_for_create = !delete.is_finished(); + + store.block_account_save.store(false, Ordering::SeqCst); + store.release_account_save.notify_one(); + create.await.expect("create task").expect("temporary account create"); + delete.await.expect("delete task").expect("temporary account delete"); + + assert!(delete_waited_for_create, "delete must wait for the in-flight STS create"); + assert!(!cache.cache.snapshot().sts_accounts.contains_key(&access_key)); + assert!(store.saved_user.lock().expect("saved_user mutex poisoned").is_none()); + } + #[tokio::test] async fn test_init_keeps_error_state_when_initial_load_fails() { let (sender, receiver) = mpsc::channel::(1); diff --git a/crates/iam/src/store/object.rs b/crates/iam/src/store/object.rs index 03530084e..e6b023a4a 100644 --- a/crates/iam/src/store/object.rs +++ b/crates/iam/src/store/object.rs @@ -28,7 +28,10 @@ use crate::{ use futures::future::join_all; use rustfs_io_metrics::record_system_path_failure; use rustfs_policy::{auth::UserIdentity, policy::PolicyDoc}; -use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; +use rustfs_utils::{ + MaskedAccessKey, + path::{SLASH_SEPARATOR, path_join_buf}, +}; use serde::{Serialize, de::DeserializeOwned}; use std::sync::{LazyLock, Mutex}; use std::time::{Duration, Instant}; @@ -600,10 +603,10 @@ impl ObjectStore { .await .map_err(|err| { if is_err_config_not_found(&err) { - warn!(name, user_type = ?user_type, "IAM user identity missing"); + debug!(name = %MaskedAccessKey(name), user_type = ?user_type, "IAM user identity missing"); Error::NoSuchUser(name.to_owned()) } else { - warn!(name, user_type = ?user_type, error = ?err, "IAM user identity load failed"); + warn!(name = %MaskedAccessKey(name), user_type = ?user_type, error = ?err, "IAM user identity load failed"); err } })?; @@ -611,7 +614,7 @@ impl ObjectStore { if u.credentials.is_expired() { let _ = self.delete_iam_config(get_user_identity_path(name, user_type)).await; let _ = self.delete_iam_config(get_mapped_policy_path(name, user_type, false)).await; - warn!(name, user_type = ?user_type, "IAM user identity expired and was removed"); + warn!(name = %MaskedAccessKey(name), user_type = ?user_type, "IAM user identity expired and was removed"); return Err(Error::NoSuchUser(name.to_owned())); } @@ -635,7 +638,7 @@ impl ObjectStore { let _ = self.delete_iam_config(get_user_identity_path(name, user_type)).await; let _ = self.delete_iam_config(get_mapped_policy_path(name, user_type, false)).await; } - warn!(name, user_type = ?user_type, error = ?err, "IAM JWT claim extraction failed"); + warn!(name = %MaskedAccessKey(name), user_type = ?user_type, error = ?err, "IAM JWT claim extraction failed"); return Err(Error::NoSuchUser(name.to_owned())); } } @@ -873,13 +876,7 @@ impl Store for ObjectStore { async fn delete_user_identity(&self, name: &str, user_type: UserType) -> Result<()> { self.delete_iam_config(get_user_identity_path(name, user_type)) .await - .map_err(|err| { - if is_err_config_not_found(&err) { - Error::NoSuchPolicy - } else { - err - } - })?; + .map_err(|err| map_delete_user_identity_error(name, err))?; Ok(()) } async fn load_user_identity(&self, name: &str, user_type: UserType) -> Result { @@ -1327,9 +1324,18 @@ impl Store for ObjectStore { } } +fn map_delete_user_identity_error(name: &str, err: Error) -> Error { + if is_err_config_not_found(&err) { + Error::NoSuchUser(name.to_owned()) + } else { + err + } +} + #[cfg(test)] mod tests { - use super::{DecryptSource, LoadMode, ObjectStore}; + use super::{DecryptSource, LoadMode, ObjectStore, map_delete_user_identity_error}; + use crate::error::Error; use crate::keyring; use rustfs_credentials::{Credentials, init_global_action_credentials}; use serial_test::serial; @@ -1352,6 +1358,12 @@ mod tests { assert!(!LoadMode::Locked.read_opts().no_lock); } + #[test] + fn missing_user_identity_delete_maps_to_no_such_user() { + let err = map_delete_user_identity_error("missing-sts", Error::ConfigNotFound); + assert!(matches!(err, Error::NoSuchUser(name) if name == "missing-sts")); + } + fn test_cred() -> Credentials { if let Some(cred) = crate::root_credentials::credentials() { return cred; diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 07a6aa81f..da02dd8bc 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -48,6 +48,16 @@ use time::OffsetDateTime; use tokio::sync::RwLock; use tracing::{error, info, warn}; +#[cfg(not(test))] +const STS_INVALIDATION_RETRY_INITIAL_DELAY: std::time::Duration = std::time::Duration::from_secs(1); +#[cfg(test)] +const STS_INVALIDATION_RETRY_INITIAL_DELAY: std::time::Duration = std::time::Duration::from_millis(1); +const STS_INVALIDATION_MAX_ATTEMPTS: usize = 3; +#[cfg(not(test))] +const STS_INVALIDATION_ATTEMPT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +#[cfg(test)] +const STS_INVALIDATION_ATTEMPT_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(20); + pub const MAX_SVCSESSION_POLICY_SIZE: usize = 4096; pub const SITE_REPLICATOR_SERVICE_ACCOUNT: &str = "site-replicator-0"; @@ -393,17 +403,74 @@ impl IamSys { /// associated session token. This is the primitive used by the admin /// `revoke-tokens` endpoint to revoke STS credentials for a parent user. pub async fn delete_temp_account(&self, access_key: &str, notify: bool) -> Result<()> { - self.store.delete_user(access_key, UserType::Sts).await?; - - if notify && !self.has_watcher() { - for r in notify_iam_delete_user(access_key).await { - if let Some(err) = r.err { - warn!("notify delete_temp_account failed: {}", err); - } - } + if !notify || self.has_watcher() { + return self.store.delete_user(access_key, UserType::Sts).await; } - Ok(()) + let runtime = tokio::runtime::Handle::try_current().map_err(Error::other)?; + #[cfg(test)] + let notification_probe = crate::LOAD_USER_NOTIFICATION_PROBE.try_with(Arc::clone).ok(); + #[cfg(test)] + let notification_available = notification_probe.is_some() || crate::runtime_sources::notification_sys().is_some(); + #[cfg(not(test))] + let notification_available = crate::runtime_sources::notification_sys().is_some(); + if !notification_available { + return Err(Error::other("IAM peer notification system is unavailable")); + } + + let store = Arc::clone(&self.store); + let access_key = access_key.to_string(); + + let operation = async move { + store.delete_user(&access_key, UserType::Sts).await?; + + let mut delay = STS_INVALIDATION_RETRY_INITIAL_DELAY; + for attempt in 1..=STS_INVALIDATION_MAX_ATTEMPTS { + let attempt_error = + match tokio::time::timeout(STS_INVALIDATION_ATTEMPT_TIMEOUT, notify_iam_load_user(&access_key, true)).await { + Ok(results) => results.into_iter().find_map(|result| result.err).map(Error::other), + Err(_) => Some(Error::other("peer STS invalidation timed out")), + }; + let Some(err) = attempt_error else { + return Ok(()); + }; + if attempt == STS_INVALIDATION_MAX_ATTEMPTS { + return Err(Error::other(err)); + } + tokio::time::sleep(delay).await; + delay = delay.saturating_mul(2); + } + unreachable!("STS invalidation retry loop always returns") + }; + + #[cfg(test)] + let task = runtime.spawn(async move { + if let Some(probe) = notification_probe { + return crate::LOAD_USER_NOTIFICATION_PROBE.scope(probe, operation).await; + } + operation.await + }); + #[cfg(not(test))] + let task = runtime.spawn(operation); + + task.await.map_err(Error::other)? + } + + #[cfg(test)] + fn load_user_notification_probe( + failures_before_success: usize, + block: bool, + panic: bool, + ) -> Arc { + Arc::new(crate::LoadUserNotificationProbe { + observed: std::sync::Mutex::new(None), + remaining_failures: std::sync::atomic::AtomicUsize::new(failures_before_success), + attempts: std::sync::atomic::AtomicUsize::new(0), + panic, + started: tokio::sync::Notify::new(), + release: block.then(tokio::sync::Notify::new), + completed: tokio::sync::Notify::new(), + }) } async fn notify_for_user(&self, name: &str, is_temp: bool) { @@ -1855,6 +1922,11 @@ mod tests { empty_policies: bool, saved_sts_users: Arc>>, saved_service_account_count: Arc>, + fail_delete: Arc, + deleted_mapped_policies: Arc>>, + block_delete: Arc, + delete_started: Arc, + release_delete: Arc, } impl StsTestMockStore { @@ -1863,6 +1935,11 @@ mod tests { empty_policies, saved_sts_users: Arc::new(Mutex::new(HashMap::new())), saved_service_account_count: Arc::new(Mutex::new(0)), + fail_delete: Arc::new(std::sync::atomic::AtomicBool::new(false)), + deleted_mapped_policies: Arc::new(Mutex::new(Vec::new())), + block_delete: Arc::new(std::sync::atomic::AtomicBool::new(false)), + delete_started: Arc::new(tokio::sync::Notify::new()), + release_delete: Arc::new(tokio::sync::Notify::new()), } } @@ -1913,6 +1990,13 @@ mod tests { } async fn delete_user_identity(&self, name: &str, _user_type: UserType) -> Result<()> { + if self.block_delete.load(std::sync::atomic::Ordering::SeqCst) { + self.delete_started.notify_one(); + self.release_delete.notified().await; + } + if self.fail_delete.load(std::sync::atomic::Ordering::SeqCst) { + return Err(Error::Io(std::io::Error::other("delete temporary account failed"))); + } self.saved_sts_users .lock() .expect("saved_sts_users mutex poisoned") @@ -1930,7 +2014,7 @@ mod tests { } async fn load_user(&self, name: &str, user_type: UserType, m: &mut HashMap) -> Result<()> { - if name == "deleted-notify-user" { + if matches!(name, "deleted-notify-user" | "deleted-notify-sts") { return Err(Error::NoSuchUser(name.to_string())); } @@ -2008,7 +2092,11 @@ mod tests { Err(Error::InvalidArgument) } - async fn delete_mapped_policy(&self, _name: &str, _user_type: UserType, _is_group: bool) -> Result<()> { + async fn delete_mapped_policy(&self, name: &str, user_type: UserType, _is_group: bool) -> Result<()> { + self.deleted_mapped_policies + .lock() + .expect("deleted_mapped_policies mutex poisoned") + .push((name.to_string(), user_type)); Err(Error::InvalidArgument) } @@ -3867,6 +3955,363 @@ mod tests { assert!(iam_sys.store.cache.snapshot().sts_policies.contains_key("notify-sts-parent")); } + #[tokio::test] + async fn delete_temp_account_notifies_peers_as_sts_user() { + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + let probe = IamSys::::load_user_notification_probe(0, false, false); + + crate::LOAD_USER_NOTIFICATION_PROBE + .scope(Arc::clone(&probe), async { + iam_sys + .delete_temp_account("deleted-notify-sts", true) + .await + .expect("delete temporary account"); + }) + .await; + assert_eq!( + probe + .observed + .lock() + .expect("notification probe mutex poisoned") + .as_ref() + .map(|(access_key, temp)| (access_key.as_str(), *temp)), + Some(("deleted-notify-sts", true)), + "peer notification must retain the STS access key and user type" + ); + } + + #[tokio::test] + async fn delete_temp_account_reports_peer_invalidation_failure() { + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + let probe = IamSys::::load_user_notification_probe(STS_INVALIDATION_MAX_ATTEMPTS, false, false); + + crate::LOAD_USER_NOTIFICATION_PROBE + .scope(Arc::clone(&probe), async { + let result = iam_sys.delete_temp_account("deleted-notify-sts", true).await; + let err = result.expect_err("failed peer invalidation must fail STS revocation"); + assert!(err.to_string().contains("peer notification failed")); + }) + .await; + assert_eq!( + probe + .observed + .lock() + .expect("notification probe mutex poisoned") + .as_ref() + .map(|(access_key, temp)| (access_key.as_str(), *temp)), + Some(("deleted-notify-sts", true)), + "failed notification must retain the STS access key and user type" + ); + assert_eq!(probe.attempts.load(std::sync::atomic::Ordering::SeqCst), STS_INVALIDATION_MAX_ATTEMPTS); + } + + #[tokio::test] + async fn transient_peer_invalidation_retries_after_local_delete() { + const ACCESS_KEY: &str = "retryable-revoked-sts"; + + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + + let probe = IamSys::::load_user_notification_probe(2, false, false); + crate::LOAD_USER_NOTIFICATION_PROBE + .scope(Arc::clone(&probe), iam_sys.delete_temp_account(ACCESS_KEY, true)) + .await + .expect("transient peer invalidation should converge within the retry budget"); + + assert_eq!( + probe.attempts.load(std::sync::atomic::Ordering::SeqCst), + STS_INVALIDATION_MAX_ATTEMPTS, + "peer invalidation must retry until it succeeds" + ); + assert_eq!( + probe + .observed + .lock() + .expect("notification probe mutex poisoned") + .as_ref() + .map(|(access_key, temp)| (access_key.as_str(), *temp)), + Some((ACCESS_KEY, true)) + ); + } + + #[tokio::test] + async fn stalled_peer_invalidation_is_bounded_by_attempt_timeouts() { + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + let probe = IamSys::::load_user_notification_probe(0, true, false); + + let err = crate::LOAD_USER_NOTIFICATION_PROBE + .scope(Arc::clone(&probe), async { + iam_sys + .delete_temp_account("stalled-peer-sts", true) + .await + .expect_err("stalled peer invalidation must fail after bounded attempts") + }) + .await; + + assert!(err.to_string().contains("peer STS invalidation timed out")); + assert_eq!(probe.attempts.load(std::sync::atomic::Ordering::SeqCst), STS_INVALIDATION_MAX_ATTEMPTS); + } + + #[tokio::test] + async fn delete_temp_account_reports_local_deletion_failure_without_notifying() { + let store = StsTestMockStore::new(false); + store.fail_delete.store(true, std::sync::atomic::Ordering::SeqCst); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + let probe = IamSys::::load_user_notification_probe(0, false, false); + + let err = crate::LOAD_USER_NOTIFICATION_PROBE + .scope(Arc::clone(&probe), async { + iam_sys + .delete_temp_account("deleted-notify-sts", true) + .await + .expect_err("local deletion failure must fail STS revocation") + }) + .await; + + assert!(err.to_string().contains("delete temporary account failed")); + assert!( + probe.observed.lock().expect("notification probe mutex poisoned").is_none(), + "peer invalidation must not run after local deletion fails" + ); + } + + #[tokio::test] + async fn delete_temp_account_reports_notification_task_panic() { + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + let probe = IamSys::::load_user_notification_probe(0, false, true); + + let err = crate::LOAD_USER_NOTIFICATION_PROBE + .scope(Arc::clone(&probe), async { + iam_sys + .delete_temp_account("deleted-notify-sts", true) + .await + .expect_err("notification task panic must fail STS revocation") + }) + .await; + + assert!(err.to_string().contains("panicked")); + } + + #[tokio::test] + async fn delete_temp_account_notification_survives_caller_cancellation() { + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = Arc::new(IamSys::new(cache_manager)); + let probe = IamSys::::load_user_notification_probe(0, true, false); + let call = { + let iam_sys = Arc::clone(&iam_sys); + crate::LOAD_USER_NOTIFICATION_PROBE.scope(Arc::clone(&probe), async move { + iam_sys.delete_temp_account("deleted-notify-sts", true).await + }) + }; + let call = tokio::spawn(call); + probe.started.notified().await; + + call.abort(); + assert!(call.await.expect_err("caller task should be cancelled").is_cancelled()); + probe + .release + .as_ref() + .expect("blocking probe must have a release signal") + .notify_one(); + probe.completed.notified().await; + + assert_eq!( + probe + .observed + .lock() + .expect("notification probe mutex poisoned") + .as_ref() + .map(|(access_key, temp)| (access_key.as_str(), *temp)), + Some(("deleted-notify-sts", true)), + "background peer invalidation must complete with the STS access key and user type" + ); + } + + #[tokio::test] + async fn delete_temp_account_local_delete_survives_caller_cancellation() { + const ACCESS_KEY: &str = "cancelled-during-local-delete"; + + let store = StsTestMockStore::new(false); + store.saved_sts_users.lock().expect("saved_sts_users mutex poisoned").insert( + ACCESS_KEY.to_string(), + UserIdentity::from(Credentials { + access_key: ACCESS_KEY.to_string(), + secret_key: "temporary-user-secret".to_string(), + session_token: "session-token".to_string(), + status: ACCOUNT_ON.to_string(), + ..Default::default() + }), + ); + store.block_delete.store(true, std::sync::atomic::Ordering::SeqCst); + let store_probe = store.clone(); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = Arc::new(IamSys::new(cache_manager)); + let probe = IamSys::::load_user_notification_probe(0, false, false); + let call = { + let iam_sys = Arc::clone(&iam_sys); + crate::LOAD_USER_NOTIFICATION_PROBE + .scope(Arc::clone(&probe), async move { iam_sys.delete_temp_account(ACCESS_KEY, true).await }) + }; + let call = tokio::spawn(call); + store_probe.delete_started.notified().await; + + call.abort(); + assert!(call.await.expect_err("caller task should be cancelled").is_cancelled()); + store_probe.release_delete.notify_one(); + probe.completed.notified().await; + + assert!( + !store_probe + .saved_sts_users + .lock() + .expect("saved_sts_users mutex poisoned") + .contains_key(ACCESS_KEY) + ); + } + + #[test] + fn notified_delete_without_tokio_runtime_returns_error() { + let runtime = tokio::runtime::Runtime::new().expect("create test runtime"); + let iam_sys = runtime.block_on(async { + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + IamSys::new(cache_manager) + }); + drop(runtime); + + let result = futures::executor::block_on(iam_sys.delete_temp_account("deleted-notify-sts", true)); + let err = result.expect_err("notified deletion without a Tokio runtime must return an error"); + assert!(err.to_string().contains("Tokio")); + } + + #[tokio::test] + async fn notified_delete_without_notification_system_returns_error() { + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + + let err = iam_sys + .delete_temp_account("deleted-notify-sts", true) + .await + .expect_err("missing peer notification system must fail STS revocation"); + assert!(err.to_string().contains("peer notification system is unavailable")); + } + + #[tokio::test] + async fn missing_sts_notification_evicts_only_sts_cache_entry() { + const ACCESS_KEY: &str = "deleted-notify-sts"; + const GROUP: &str = "deleted-notify-sts-group"; + + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + let regular_user = UserIdentity::from(Credentials { + access_key: ACCESS_KEY.to_string(), + secret_key: "regular-user-secret".to_string(), + status: ACCOUNT_ON.to_string(), + ..Default::default() + }); + let sts_user = UserIdentity::from(Credentials { + access_key: ACCESS_KEY.to_string(), + secret_key: "temporary-user-secret".to_string(), + session_token: "session-token".to_string(), + status: ACCOUNT_ON.to_string(), + ..Default::default() + }); + let mapped_policy = MappedPolicy::new("readwrite"); + let membership = HashSet::from([GROUP.to_string()]); + let group = GroupInfo::new(vec![ACCESS_KEY.to_string()]); + iam_sys.store.cache.with_write_lock(|cache| { + let now = OffsetDateTime::now_utc(); + cache.add_or_update_user(ACCESS_KEY, ®ular_user, now); + cache.add_or_update_user_policy(ACCESS_KEY, &mapped_policy, now); + cache.add_or_update_group(GROUP, &group, now); + cache.add_or_update_user_group_membership(ACCESS_KEY, &membership, now); + cache.add_or_update_sts_account(ACCESS_KEY, &sts_user, now); + }); + + iam_sys + .load_user(ACCESS_KEY, UserType::Sts) + .await + .expect("process missing STS user notification"); + + let cache = iam_sys.store.cache.snapshot(); + assert!(!cache.sts_accounts.contains_key(ACCESS_KEY)); + assert!( + cache.users.contains_key(ACCESS_KEY), + "STS invalidation must not evict a same-name regular user" + ); + assert!(cache.user_policies.contains_key(ACCESS_KEY)); + assert!(cache.user_group_memberships.contains_key(ACCESS_KEY)); + assert!( + cache + .groups + .get(GROUP) + .is_some_and(|group| group.members.contains(&ACCESS_KEY.to_string())), + "STS invalidation must preserve same-name regular-user group membership" + ); + } + + #[tokio::test] + async fn delete_temp_account_preserves_same_name_regular_cache_state() { + const ACCESS_KEY: &str = "deleted-notify-sts"; + + let store = StsTestMockStore::new(false); + let cache_manager = IamCache::new(store).await.expect("initialize IAM cache"); + let iam_sys = IamSys::new(cache_manager); + let regular_user = UserIdentity::from(Credentials { + access_key: ACCESS_KEY.to_string(), + secret_key: "regular-user-secret".to_string(), + status: ACCOUNT_ON.to_string(), + ..Default::default() + }); + let sts_user = UserIdentity::from(Credentials { + access_key: ACCESS_KEY.to_string(), + secret_key: "temporary-user-secret".to_string(), + session_token: "session-token".to_string(), + status: ACCOUNT_ON.to_string(), + ..Default::default() + }); + let mapped_policy = MappedPolicy::new("readwrite"); + iam_sys.store.cache.with_write_lock(|cache| { + let now = OffsetDateTime::now_utc(); + cache.add_or_update_user(ACCESS_KEY, ®ular_user, now); + cache.add_or_update_user_policy(ACCESS_KEY, &mapped_policy, now); + cache.add_or_update_sts_account(ACCESS_KEY, &sts_user, now); + }); + + iam_sys + .delete_temp_account(ACCESS_KEY, false) + .await + .expect("delete temporary account without peer notification"); + + let cache = iam_sys.store.cache.snapshot(); + assert!(!cache.sts_accounts.contains_key(ACCESS_KEY)); + assert!(cache.users.contains_key(ACCESS_KEY)); + assert!(cache.user_policies.contains_key(ACCESS_KEY)); + assert!( + iam_sys + .store + .api + .deleted_mapped_policies + .lock() + .expect("deleted_mapped_policies mutex poisoned") + .is_empty(), + "deleting one STS identity must not delete a parent-scoped STS policy mapping" + ); + } + #[tokio::test] async fn test_missing_user_notification_cleans_related_cache_state() { let store = StsTestMockStore::new(false); diff --git a/rustfs/src/admin/handlers/idp_compat.rs b/rustfs/src/admin/handlers/idp_compat.rs index d29a68c9a..4138e0a62 100644 --- a/rustfs/src/admin/handlers/idp_compat.rs +++ b/rustfs/src/admin/handlers/idp_compat.rs @@ -59,10 +59,11 @@ use rustfs_madmin::{ ServiceAccountInfo, }; use rustfs_policy::policy::action::{Action, AdminAction}; +use rustfs_utils::MaskedAccessKey; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use serde::Serialize; -use std::collections::HashMap; +use std::{collections::HashMap, sync::LazyLock}; use time::OffsetDateTime; use tracing::warn; use url::form_urlencoded; @@ -210,6 +211,59 @@ struct RevokeTokensResp { /// provider. This limitation is reported as COMPAT-SEMANTICS-ONLY. pub struct RevokeTokens {} +const MAX_CONCURRENT_STS_REVOCATIONS: usize = 16; +const MAX_GLOBAL_STS_REVOCATIONS: usize = 64; +static STS_REVOCATION_PERMITS: LazyLock = + LazyLock::new(|| tokio::sync::Semaphore::new(MAX_GLOBAL_STS_REVOCATIONS)); + +async fn revoke_matching_sts_accounts( + accounts: Vec, + user_provider: String, + revoke: F, +) -> Result, tokio::task::JoinError> +where + E: Send + 'static, + F: Fn(String) -> Fut + Clone + Send + Sync + 'static, + Fut: std::future::Future> + Send + 'static, +{ + tokio::spawn(async move { + use futures::StreamExt; + + let access_keys = accounts + .into_iter() + .filter(|credential| guess_user_provider(credential) == user_provider) + .map(|credential| credential.access_key) + .collect::>(); + let results = futures::stream::iter(access_keys) + .map(|access_key| { + let revoke = revoke.clone(); + async move { revoke(access_key).await } + }) + .buffer_unordered(MAX_CONCURRENT_STS_REVOCATIONS) + .collect::>() + .await; + + let mut revoked = 0usize; + let mut first_error = None; + for result in results { + match result { + Ok(()) => revoked += 1, + Err(err) => { + if first_error.is_none() { + first_error = Some(err); + } + } + } + } + + match first_error { + Some(err) => Err(err), + None => Ok(revoked), + } + }) + .await +} + #[async_trait::async_trait] impl Operation for RevokeTokens { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { @@ -248,7 +302,7 @@ impl Operation for RevokeTokens { subsystem = LOG_SUBSYSTEM_IDP, event = EVENT_ADMIN_IDP_STATE, action = "revoke_tokens", - target_user = %target_user, + target_user = %MaskedAccessKey(&target_user), result = "list_sts_failed", error = ?e, "admin idp state" @@ -260,31 +314,49 @@ impl Operation for RevokeTokens { // not persist that claim. Only an empty type / full revoke deletes. let revoke_all = query.full_revoke || query.token_revoke_type.is_empty(); - let mut revoked = 0usize; - if revoke_all { - for sts in &sts_accounts { - // Provider scoping: only revoke STS credentials that were issued - // through the requested identity provider. - if guess_user_provider(sts) != user_provider { - continue; + let revoked = if revoke_all { + let batch_result = revoke_matching_sts_accounts(sts_accounts, user_provider.clone(), move |access_key| { + let iam_store = iam_store.clone(); + async move { + let _permit = STS_REVOCATION_PERMITS + .acquire() + .await + .map_err(rustfs_iam::error::Error::other)?; + iam_store.delete_temp_account(&access_key, true).await } - - iam_store.delete_temp_account(&sts.access_key, true).await.map_err(|e| { + }) + .await + .map_err(|err| { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_IDP, + event = EVENT_ADMIN_IDP_STATE, + action = "revoke_tokens", + result = "batch_task_failed", + error = ?err, + "admin idp state" + ); + s3_error!(InternalError, "revoke token batch failed") + })?; + match batch_result { + Ok(revoked) => revoked, + Err(err) => { warn!( component = LOG_COMPONENT_ADMIN, subsystem = LOG_SUBSYSTEM_IDP, event = EVENT_ADMIN_IDP_STATE, action = "revoke_tokens", - access_key = %sts.access_key, + target_user = %MaskedAccessKey(&target_user), result = "delete_failed", - error = ?e, + error = ?err, "admin idp state" ); - s3_error!(InternalError, "revoke token failed") - })?; - revoked += 1; + return Err(s3_error!(InternalError, "revoke token failed")); + } } - } + } else { + 0 + }; let _ = owner; json_response( @@ -949,6 +1021,95 @@ mod tests { assert!(q.full_revoke || q.token_revoke_type.is_empty()); } + #[tokio::test] + async fn revoke_tokens_attempts_all_matching_credentials_before_returning_error() { + let credentials = ["first-sts", "second-sts"].map(|access_key| StoredCredentials { + access_key: access_key.to_string(), + secret_key: "temporary-user-secret".to_string(), + session_token: "session-token".to_string(), + expiration: Some(OffsetDateTime::now_utc() + time::Duration::hours(1)), + parent_user: "parent-user".to_string(), + ..Default::default() + }); + let attempted = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + + let result = revoke_matching_sts_accounts(credentials.to_vec(), "builtin".to_string(), { + let attempted = std::sync::Arc::clone(&attempted); + move |access_key| { + let attempted = std::sync::Arc::clone(&attempted); + async move { + attempted + .lock() + .expect("attempted revocations mutex poisoned") + .push(access_key.clone()); + if access_key == "first-sts" { + Err("first revoke failed") + } else { + Ok(()) + } + } + } + }) + .await + .expect("revocation batch task"); + + assert_eq!(result, Err("first revoke failed")); + assert_eq!( + *attempted.lock().expect("attempted revocations mutex poisoned"), + vec!["first-sts".to_string(), "second-sts".to_string()] + ); + } + + #[tokio::test] + async fn revoke_tokens_batch_survives_caller_cancellation() { + let credentials = (0..32) + .map(|index| StoredCredentials { + access_key: format!("sts-{index}"), + secret_key: "temporary-user-secret".to_string(), + session_token: "session-token".to_string(), + expiration: Some(OffsetDateTime::now_utc() + time::Duration::hours(1)), + parent_user: "parent-user".to_string(), + ..Default::default() + }) + .collect::>(); + let attempts = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let progress = std::sync::Arc::new(tokio::sync::Notify::new()); + let first_window_started = std::sync::Arc::new(tokio::sync::Barrier::new(17)); + let release_first_window = std::sync::Arc::new(tokio::sync::Barrier::new(17)); + + let caller = tokio::spawn(revoke_matching_sts_accounts(credentials, "builtin".to_string(), { + let attempts = std::sync::Arc::clone(&attempts); + let progress = std::sync::Arc::clone(&progress); + let first_window_started = std::sync::Arc::clone(&first_window_started); + let release_first_window = std::sync::Arc::clone(&release_first_window); + move |_| { + let attempts = std::sync::Arc::clone(&attempts); + let progress = std::sync::Arc::clone(&progress); + let first_window_started = std::sync::Arc::clone(&first_window_started); + let release_first_window = std::sync::Arc::clone(&release_first_window); + async move { + let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + progress.notify_one(); + if attempt < MAX_CONCURRENT_STS_REVOCATIONS { + first_window_started.wait().await; + release_first_window.wait().await; + } + Ok::<(), ()>(()) + } + } + })); + first_window_started.wait().await; + + caller.abort(); + assert!(caller.await.expect_err("caller task should be cancelled").is_cancelled()); + release_first_window.wait().await; + while attempts.load(std::sync::atomic::Ordering::SeqCst) < 32 { + progress.notified().await; + } + + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 32); + } + #[test] fn list_type_flags_parse_known_values() { assert!(matches!(ListTypeFlags::parse(""), Ok(ListTypeFlags { sts: true, svc: true }))); diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 5aec4de94..9511c0636 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -671,6 +671,25 @@ for pattern in "${forbidden_patterns[@]}"; do fi done +if rg -n -F -- 'warn!(name = %MaskedAccessKey(name), user_type = ?user_type, "IAM user identity missing")' crates/iam/src/store/object.rs >/dev/null; then + echo "❌ logging guardrail violation: missing IAM identity is an expected debug event, not a warning" >&2 + exit 1 +fi + +unmasked_iam_identity_logs="$(rg -n 'IAM (user identity|JWT claim)' crates/iam/src/store/object.rs | rg -v 'MaskedAccessKey' || true)" +if [[ -n "$unmasked_iam_identity_logs" ]]; then + echo "❌ logging guardrail violation: IAM identity log omits MaskedAccessKey" >&2 + echo "$unmasked_iam_identity_logs" >&2 + exit 1 +fi + +unmasked_revoke_fields="$(rg -n '(access_key\s*=\s*%\s*&?sts\.access_key|target_user\s*=\s*%\s*&?target_user)' rustfs/src/admin/handlers/idp_compat.rs || true)" +if [[ -n "$unmasked_revoke_fields" ]]; then + echo "❌ logging guardrail violation: revoke-tokens log exposes an unmasked credential identifier" >&2 + echo "$unmasked_revoke_fields" >&2 + exit 1 +fi + # Secret material must never be interpolated into log or error strings. # Error messages are log content: they propagate via `?` and are printed by # startup/error logging far from the construction site, and a value that fails