gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-28 17:18:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(iam): serialize IAM cache writes (#3105)

Browse Source 
* fix(iam): serialize IAM cache writes

* fix(iam): timestamp rebuilt group memberships

* fix(iam): publish cache updates atomically

* fix(iam): reuse policy cache snapshots

* fix(iam): commit missing user notification cache updates atomically

* fix(iam): remove unused cache membership rebuild wrapper

---------

Co-authored-by: 季宏伟 <jihongwei@jihongweis-MacBook-Pro.local>
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
GatewayJ
2026-05-29 16:02:42 +08:00
committed by GitHub
parent ede813070f
commit c257043b63
4 changed files with 1015 additions and 695 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/iam/src/cache.rs
+349 -132
View File
@@ -16,10 +16,10 @@ use std::{
collections::{HashMap, HashSet},
ops::{Deref, DerefMut},
ptr,
sync::Arc,
sync::{Arc, Mutex},
};
use arc_swap::{ArcSwap, AsRaw, Guard};
use arc_swap::{ArcSwap, Guard};
use rustfs_policy::{
auth::UserIdentity,
policy::{Args, PolicyDoc},
@@ -29,80 +29,302 @@ use tracing::warn;
use crate::store::{GroupInfo, MappedPolicy};
/// Immutable IAM cache snapshot published atomically by [`Cache`].
///
/// Readers should load one `CacheState` and read all related maps from it. Writers
/// must go through `Cache`/`LockedCache` so multi-map updates publish as one state.
#[derive(Clone)]
pub(crate) struct CacheState {
pub(crate) policy_docs: Arc<CacheEntity<PolicyDoc>>,
pub(crate) users: Arc<CacheEntity<UserIdentity>>,
pub(crate) user_policies: Arc<CacheEntity<MappedPolicy>>,
pub(crate) sts_accounts: Arc<CacheEntity<UserIdentity>>,
pub(crate) sts_policies: Arc<CacheEntity<MappedPolicy>>,
pub(crate) groups: Arc<CacheEntity<GroupInfo>>,
pub(crate) user_group_memberships: Arc<CacheEntity<HashSet<String>>>,
pub(crate) group_policies: Arc<CacheEntity<MappedPolicy>>,
}
impl Default for CacheState {
fn default() -> Self {
Self {
policy_docs: Arc::new(CacheEntity::default()),
users: Arc::new(CacheEntity::default()),
user_policies: Arc::new(CacheEntity::default()),
sts_accounts: Arc::new(CacheEntity::default()),
sts_policies: Arc::new(CacheEntity::default()),
groups: Arc::new(CacheEntity::default()),
user_group_memberships: Arc::new(CacheEntity::default()),
group_policies: Arc::new(CacheEntity::default()),
}
}
}
pub struct Cache {
pub policy_docs: ArcSwap<CacheEntity<PolicyDoc>>,
pub users: ArcSwap<CacheEntity<UserIdentity>>,
pub user_policies: ArcSwap<CacheEntity<MappedPolicy>>,
pub sts_accounts: ArcSwap<CacheEntity<UserIdentity>>,
pub sts_policies: ArcSwap<CacheEntity<MappedPolicy>>,
pub groups: ArcSwap<CacheEntity<GroupInfo>>,
pub user_group_memberships: ArcSwap<CacheEntity<HashSet<String>>>,
pub group_policies: ArcSwap<CacheEntity<MappedPolicy>>,
state: ArcSwap<CacheState>,
write_lock: Mutex<()>,
}
impl Default for Cache {
fn default() -> Self {
Self {
policy_docs: ArcSwap::new(Arc::new(CacheEntity::default())),
users: ArcSwap::new(Arc::new(CacheEntity::default())),
user_policies: ArcSwap::new(Arc::new(CacheEntity::default())),
sts_accounts: ArcSwap::new(Arc::new(CacheEntity::default())),
sts_policies: ArcSwap::new(Arc::new(CacheEntity::default())),
groups: ArcSwap::new(Arc::new(CacheEntity::default())),
user_group_memberships: ArcSwap::new(Arc::new(CacheEntity::default())),
group_policies: ArcSwap::new(Arc::new(CacheEntity::default())),
state: ArcSwap::new(Arc::new(CacheState::default())),
write_lock: Mutex::new(()),
}
}
}
pub(crate) type CacheSnapshot = Guard<Arc<CacheState>>;
impl Cache {
pub fn ptr_eq<Base, A, B>(a: A, b: B) -> bool
where
A: AsRaw<Base>,
B: AsRaw<Base>,
{
let a = a.as_raw();
let b = b.as_raw();
ptr::eq(a, b)
pub(crate) fn snapshot(&self) -> CacheSnapshot {
self.state.load()
}
fn exec<T: Clone>(target: &ArcSwap<CacheEntity<T>>, t: OffsetDateTime, mut op: impl FnMut(&mut CacheEntity<T>)) {
let mut cur = target.load();
loop {
// If the current update time is later than the execution time,
// the background task is loaded and the current operation does not need to be performed.
if cur.load_time >= t {
return;
}
let mut new = CacheEntity::clone(&cur);
op(&mut new);
// Replace content with CAS atoms
let prev = target.compare_and_swap(&*cur, Arc::new(new));
let swapped = Self::ptr_eq(&*cur, &*prev);
if swapped {
return;
} else {
cur = prev;
}
pub(crate) fn with_write_lock<R>(&self, f: impl FnOnce(&mut LockedCache) -> R) -> R {
let _guard = self.write_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let current = self.state.load_full();
let mut locked = LockedCache {
state: CacheState::clone(&current),
current_ptr: Arc::as_ptr(&current),
dirty: false,
};
let ret = f(&mut locked);
if locked.dirty {
self.state.store(Arc::new(locked.state));
}
ret
}
pub fn add_or_update<T: Clone>(target: &ArcSwap<CacheEntity<T>>, key: &str, value: &T, t: OffsetDateTime) {
Self::exec(target, t, |map: &mut CacheEntity<T>| {
pub fn add_or_update_policy_doc(&self, key: &str, value: &PolicyDoc, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_policy_doc(key, value, t));
}
pub fn add_or_update_user(&self, key: &str, value: &UserIdentity, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_user(key, value, t));
}
pub fn add_or_update_user_policy(&self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_user_policy(key, value, t));
}
pub fn add_or_update_sts_account(&self, key: &str, value: &UserIdentity, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_sts_account(key, value, t));
}
pub fn add_or_update_sts_policy(&self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_sts_policy(key, value, t));
}
pub fn add_or_update_group(&self, key: &str, value: &GroupInfo, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_group(key, value, t));
}
pub fn add_or_update_user_group_membership(&self, key: &str, value: &HashSet<String>, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_user_group_membership(key, value, t));
}
pub fn add_or_update_group_policy(&self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.add_or_update_group_policy(key, value, t));
}
pub fn delete_policy_doc(&self, key: &str, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.delete_policy_doc(key, t));
}
pub fn delete_user(&self, key: &str, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.delete_user(key, t));
}
pub fn delete_user_policy(&self, key: &str, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.delete_user_policy(key, t));
}
pub fn delete_sts_account(&self, key: &str, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.delete_sts_account(key, t));
}
pub fn delete_sts_policy(&self, key: &str, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.delete_sts_policy(key, t));
}
pub fn delete_group(&self, key: &str, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.delete_group(key, t));
}
pub fn delete_group_policy(&self, key: &str, t: OffsetDateTime) {
self.with_write_lock(|cache| cache.delete_group_policy(key, t));
}
}
pub(crate) struct LockedCache {
state: CacheState,
current_ptr: *const CacheState,
dirty: bool,
}
impl LockedCache {
pub(crate) fn state(&self) -> &CacheState {
&self.state
}
pub(crate) fn matches_snapshot(&self, snapshot: &CacheSnapshot) -> bool {
ptr::eq(self.current_ptr, Arc::as_ptr(snapshot))
}
fn exec<T: Clone>(target: &mut Arc<CacheEntity<T>>, t: OffsetDateTime, mut op: impl FnMut(&mut CacheEntity<T>)) -> bool {
if target.load_time >= t {
return false;
}
let mut new = CacheEntity::clone(target);
op(&mut new);
*target = Arc::new(new);
true
}
fn replaced<T>(value: CacheEntity<T>) -> Arc<CacheEntity<T>> {
Arc::new(value.update_load_time())
}
pub(crate) fn replace_policy_docs(&mut self, value: CacheEntity<PolicyDoc>) {
self.state.policy_docs = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_users(&mut self, value: CacheEntity<UserIdentity>) {
self.state.users = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_user_policies(&mut self, value: CacheEntity<MappedPolicy>) {
self.state.user_policies = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_sts_accounts(&mut self, value: CacheEntity<UserIdentity>) {
self.state.sts_accounts = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_sts_policies(&mut self, value: CacheEntity<MappedPolicy>) {
self.state.sts_policies = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_groups(&mut self, value: CacheEntity<GroupInfo>) {
self.state.groups = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_group_policies(&mut self, value: CacheEntity<MappedPolicy>) {
self.state.group_policies = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_user_group_memberships(&mut self, value: CacheEntity<HashSet<String>>) {
self.state.user_group_memberships = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn add_or_update_policy_doc(&mut self, key: &str, value: &PolicyDoc, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.policy_docs, t, |map| {
map.insert(key.to_string(), value.clone());
})
});
}
pub fn delete<T: Clone>(target: &ArcSwap<CacheEntity<T>>, key: &str, t: OffsetDateTime) {
Self::exec(target, t, |map: &mut CacheEntity<T>| {
pub(crate) fn add_or_update_user(&mut self, key: &str, value: &UserIdentity, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.users, t, |map| {
map.insert(key.to_string(), value.clone());
});
}
pub(crate) fn add_or_update_user_policy(&mut self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.user_policies, t, |map| {
map.insert(key.to_string(), value.clone());
});
}
pub(crate) fn add_or_update_sts_account(&mut self, key: &str, value: &UserIdentity, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.sts_accounts, t, |map| {
map.insert(key.to_string(), value.clone());
});
}
pub(crate) fn add_or_update_sts_policy(&mut self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.sts_policies, t, |map| {
map.insert(key.to_string(), value.clone());
});
}
pub(crate) fn add_or_update_group(&mut self, key: &str, value: &GroupInfo, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.groups, t, |map| {
map.insert(key.to_string(), value.clone());
});
}
pub(crate) fn add_or_update_user_group_membership(&mut self, key: &str, value: &HashSet<String>, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.user_group_memberships, t, |map| {
map.insert(key.to_string(), value.clone());
});
}
pub(crate) fn add_or_update_group_policy(&mut self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.group_policies, t, |map| {
map.insert(key.to_string(), value.clone());
});
}
pub(crate) fn delete_policy_doc(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.policy_docs, t, |map| {
map.remove(key);
})
});
}
pub fn build_user_group_memberships(&self) {
let groups = self.groups.load();
pub(crate) fn delete_user(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.users, t, |map| {
map.remove(key);
});
}
pub(crate) fn delete_user_policy(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.user_policies, t, |map| {
map.remove(key);
});
}
pub(crate) fn delete_user_group_membership(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.user_group_memberships, t, |map| {
map.remove(key);
});
}
pub(crate) fn delete_sts_account(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.sts_accounts, t, |map| {
map.remove(key);
});
}
pub(crate) fn delete_sts_policy(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.sts_policies, t, |map| {
map.remove(key);
});
}
pub(crate) fn delete_group(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.groups, t, |map| {
map.remove(key);
});
}
pub(crate) fn delete_group_policy(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.group_policies, t, |map| {
map.remove(key);
});
}
pub(crate) fn build_user_group_memberships(&mut self) {
let groups = Arc::clone(&self.state.groups);
let mut user_group_memberships = HashMap::new();
for (group_name, group) in groups.iter() {
for user_name in &group.members {
@@ -112,49 +334,19 @@ impl Cache {
.insert(group_name.clone());
}
}
self.user_group_memberships
.store(Arc::new(CacheEntity::new(user_group_memberships)));
self.replace_user_group_memberships(CacheEntity::new(user_group_memberships));
}
}
impl CacheInner {
#[inline]
pub fn get_user(&self, user_name: &str) -> Option<&UserIdentity> {
self.users.get(user_name).or_else(|| self.sts_accounts.get(user_name))
self.snapshot
.users
.get(user_name)
.or_else(|| self.snapshot.sts_accounts.get(user_name))
}
// fn get_policy(&self, _name: &str, _groups: &[String]) -> crate::Result<Vec<Policy>> {
// todo!()
// }
// /// Return Ok(Some(parent_name)) when the user is temporary.
// /// Return Ok(None) for non-temporary users.
// fn is_temp_user(&self, user_name: &str) -> crate::Result<Option<&str>> {
// let user = self
// .get_user(user_name)
// .ok_or_else(|| Error::NoSuchUser(user_name.to_owned()))?;
// if user.credentials.is_temp() {
// Ok(Some(&user.credentials.parent_user))
// } else {
// Ok(None)
// }
// }
// /// Return Ok(Some(parent_name)) when the user is a temporary identity.
// /// Return Ok(None) when the user is not temporary.
// fn is_service_account(&self, user_name: &str) -> crate::Result<Option<&str>> {
// let user = self
// .get_user(user_name)
// .ok_or_else(|| Error::NoSuchUser(user_name.to_owned()))?;
// if user.credentials.is_service_account() {
// Ok(Some(&user.credentials.parent_user))
// } else {
// Ok(None)
// }
// }
// todo
pub fn is_allowed_sts(&self, _args: &Args, _parent: &str) -> bool {
warn!("policy cache STS check path is not implemented");
@@ -223,30 +415,14 @@ impl<T> CacheEntity<T> {
}
}
pub type G<T> = Guard<Arc<CacheEntity<T>>>;
pub struct CacheInner {
pub policy_docs: G<PolicyDoc>,
pub users: G<UserIdentity>,
pub user_policies: G<MappedPolicy>,
pub sts_accounts: G<UserIdentity>,
pub sts_policies: G<MappedPolicy>,
pub groups: G<GroupInfo>,
pub user_group_memberships: G<HashSet<String>>,
pub group_policies: G<MappedPolicy>,
snapshot: CacheSnapshot,
}
impl From<&Cache> for CacheInner {
fn from(value: &Cache) -> Self {
Self {
policy_docs: value.policy_docs.load(),
users: value.users.load(),
user_policies: value.user_policies.load(),
sts_accounts: value.sts_accounts.load(),
sts_policies: value.sts_policies.load(),
groups: value.groups.load(),
user_group_memberships: value.user_group_memberships.load(),
group_policies: value.group_policies.load(),
snapshot: value.snapshot(),
}
}
}
@@ -255,98 +431,139 @@ impl From<&Cache> for CacheInner {
mod tests {
use std::sync::Arc;
use arc_swap::ArcSwap;
use futures::future::join_all;
use rustfs_policy::auth::UserIdentity;
use time::OffsetDateTime;
use super::CacheEntity;
use crate::cache::Cache;
use crate::store::MappedPolicy;
#[tokio::test]
async fn test_cache_entity_add() {
let cache = ArcSwap::new(Arc::new(CacheEntity::<usize>::default()));
let owner = Arc::new(Cache::default());
let mut f = vec![];
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
let c = &cache;
let owner = Arc::clone(&owner);
f.push(async move {
Cache::add_or_update(c, &key, &index, OffsetDateTime::now_utc());
let user = UserIdentity {
version: index as i64,
..Default::default()
};
owner.add_or_update_user(&key, &user, OffsetDateTime::now_utc());
});
}
join_all(f).await;
let cache = cache.load();
let cache = owner.snapshot();
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
assert_eq!(cache.get(&key), Some(&index));
assert_eq!(cache.users.get(&key).map(|user| user.version), Some(index as i64));
}
}
#[tokio::test]
async fn test_cache_entity_update() {
let cache = ArcSwap::new(Arc::new(CacheEntity::<usize>::default()));
let owner = Arc::new(Cache::default());
let mut f = vec![];
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
let c = &cache;
let owner = Arc::clone(&owner);
f.push(async move {
Cache::add_or_update(c, &key, &index, OffsetDateTime::now_utc());
let user = UserIdentity {
version: index as i64,
..Default::default()
};
owner.add_or_update_user(&key, &user, OffsetDateTime::now_utc());
});
}
join_all(f).await;
let cache_load = cache.load();
let cache_load = owner.snapshot();
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
assert_eq!(cache_load.get(&key), Some(&index));
assert_eq!(cache_load.users.get(&key).map(|user| user.version), Some(index as i64));
}
let mut f = vec![];
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
let c = &cache;
let owner = Arc::clone(&owner);
f.push(async move {
Cache::add_or_update(c, &key, &(index * 1000), OffsetDateTime::now_utc());
let user = UserIdentity {
version: (index * 1000) as i64,
..Default::default()
};
owner.add_or_update_user(&key, &user, OffsetDateTime::now_utc());
});
}
join_all(f).await;
let cache_load = cache.load();
let cache_load = owner.snapshot();
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
assert_eq!(cache_load.get(&key), Some(&(index * 1000)));
assert_eq!(cache_load.users.get(&key).map(|user| user.version), Some((index * 1000) as i64));
}
}
#[tokio::test]
async fn test_cache_entity_delete() {
let cache = ArcSwap::new(Arc::new(CacheEntity::<usize>::default()));
let owner = Arc::new(Cache::default());
let mut f = vec![];
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
let c = &cache;
let owner = Arc::clone(&owner);
f.push(async move {
Cache::add_or_update(c, &key, &index, OffsetDateTime::now_utc());
let user = UserIdentity {
version: index as i64,
..Default::default()
};
owner.add_or_update_user(&key, &user, OffsetDateTime::now_utc());
});
}
join_all(f).await;
let cache_load = cache.load();
let cache_load = owner.snapshot();
for (index, key) in (0..100).map(|x| x.to_string()).enumerate() {
assert_eq!(cache_load.get(&key), Some(&index));
assert_eq!(cache_load.users.get(&key).map(|user| user.version), Some(index as i64));
}
drop(cache_load);
let mut f = vec![];
for key in (0..100).map(|x| x.to_string()) {
let c = &cache;
let owner = Arc::clone(&owner);
f.push(async move {
Cache::delete(c, &key, OffsetDateTime::now_utc());
owner.delete_user(&key, OffsetDateTime::now_utc());
});
}
join_all(f).await;
let cache_load = cache.load();
assert!(cache_load.is_empty());
let cache_load = owner.snapshot();
assert!(cache_load.users.is_empty());
}
#[tokio::test]
async fn test_cache_snapshot_reads_one_published_state() {
let cache = Cache::default();
let before = cache.snapshot();
let user = UserIdentity {
version: 7,
..Default::default()
};
let policy = MappedPolicy::new("readwrite");
cache.with_write_lock(|cache| {
let now = OffsetDateTime::now_utc();
cache.add_or_update_user("snapshot-user", &user, now);
cache.add_or_update_user_policy("snapshot-user", &policy, now);
});
assert!(!before.users.contains_key("snapshot-user"));
assert!(!before.user_policies.contains_key("snapshot-user"));
let after = cache.snapshot();
assert!(after.users.contains_key("snapshot-user"));
assert!(after.user_policies.contains_key("snapshot-user"));
}
}
crates/iam/src/manager.rs
+483 -334
View File
File diff suppressed because it is too large Load Diff
crates/iam/src/store/object.rs
+20 -195
View File
@@ -1025,14 +1025,7 @@ impl Store for ObjectStore {
}
async fn load_all(&self, cache: &Cache) -> Result<()> {
let policy_docs_snapshot = cache.policy_docs.load();
let users_snapshot = cache.users.load();
let user_policies_snapshot = cache.user_policies.load();
let groups_snapshot = cache.groups.load();
let user_group_memberships_snapshot = cache.user_group_memberships.load();
let group_policies_snapshot = cache.group_policies.load();
let sts_accounts_snapshot = cache.sts_accounts.load();
let sts_policies_snapshot = cache.sts_policies.load();
let cache_snapshot = cache.snapshot();
let listed_config_items = self.list_all_iamconfig_items().await?;
let mut policy_docs_cache = CacheEntity::new(get_default_policyes());
@@ -1080,8 +1073,6 @@ impl Store for ObjectStore {
if let Some(item_name_list) = listed_config_items.get(USERS_LIST_KEY) {
let mut item_name_list = item_name_list.clone();
// let mut items_cache = CacheEntity::default();
loop {
if item_name_list.len() < 32 {
let items = self.load_user_concurrent(&item_name_list, UserType::Reg).await?;
@@ -1112,8 +1103,6 @@ impl Store for ObjectStore {
item_name_list = item_name_list.split_off(32);
}
// cache.users.store(Arc::new(items_cache.update_load_time()));
}
// groups
@@ -1228,8 +1217,6 @@ impl Store for ObjectStore {
// Merge items_cache to user_items_cache
user_items_cache.extend(items_cache);
// cache.users.store(Arc::new(items_cache.update_load_time()));
}
let mut sts_items_cache = CacheEntity::default();
@@ -1260,191 +1247,29 @@ impl Store for ObjectStore {
}
}
let policy_docs_current = cache.policy_docs.load();
let users_current = cache.users.load();
let user_policies_current = cache.user_policies.load();
let groups_current = cache.groups.load();
let user_group_memberships_current = cache.user_group_memberships.load();
let group_policies_current = cache.group_policies.load();
let sts_accounts_current = cache.sts_accounts.load();
let sts_policies_current = cache.sts_policies.load();
if Cache::ptr_eq(&*policy_docs_snapshot, &*policy_docs_current)
&& Cache::ptr_eq(&*users_snapshot, &*users_current)
&& Cache::ptr_eq(&*user_policies_snapshot, &*user_policies_current)
&& Cache::ptr_eq(&*groups_snapshot, &*groups_current)
&& Cache::ptr_eq(&*user_group_memberships_snapshot, &*user_group_memberships_current)
&& Cache::ptr_eq(&*group_policies_snapshot, &*group_policies_current)
&& Cache::ptr_eq(&*sts_accounts_snapshot, &*sts_accounts_current)
&& Cache::ptr_eq(&*sts_policies_snapshot, &*sts_policies_current)
{
cache.policy_docs.store(Arc::new(policy_docs_cache.update_load_time()));
if let Some(groups_cache) = groups_cache {
cache.groups.store(Arc::new(groups_cache.update_load_time()));
cache.with_write_lock(|cache| {
if cache.matches_snapshot(&cache_snapshot) {
cache.replace_policy_docs(policy_docs_cache);
if let Some(groups_cache) = groups_cache {
cache.replace_groups(groups_cache);
}
if let Some(user_policies_cache) = user_policies_cache {
cache.replace_user_policies(user_policies_cache);
}
if let Some(group_policies_cache) = group_policies_cache {
cache.replace_group_policies(group_policies_cache);
}
cache.replace_users(user_items_cache);
cache.replace_sts_accounts(sts_items_cache);
cache.replace_sts_policies(sts_policies_cache);
cache.build_user_group_memberships();
} else {
warn!("skip IAM full reload cache commit because one or more IAM caches changed during reload");
}
if let Some(user_policies_cache) = user_policies_cache {
cache.user_policies.store(Arc::new(user_policies_cache.update_load_time()));
}
if let Some(group_policies_cache) = group_policies_cache {
cache.group_policies.store(Arc::new(group_policies_cache.update_load_time()));
}
cache.users.store(Arc::new(user_items_cache.update_load_time()));
cache.sts_accounts.store(Arc::new(sts_items_cache.update_load_time()));
cache.sts_policies.store(Arc::new(sts_policies_cache.update_load_time()));
cache.build_user_group_memberships();
} else {
warn!("skip IAM full reload cache commit because one or more IAM caches changed during reload");
}
});
Ok(())
}
// /// load all and make a new cache.
// async fn load_all(&self, cache: &Cache) -> Result<()> {
// let _items = &[
// "policydb/",
// "policies/",
// "groups/",
// "policydb/users/",
// "policydb/groups/",
// "service-accounts/",
// "policydb/sts-users/",
// "sts/",
// ];
// let items = self.list_iam_config_items("config/iam/").await?;
// debug!("all iam items: {items:?}");
// let (policy_docs, users, user_policies, sts_policies, sts_accounts) = (
// Arc::new(tokio::sync::Mutex::new(CacheEntity::new(Self::get_default_policyes()))),
// Arc::new(tokio::sync::Mutex::new(CacheEntity::default())),
// Arc::new(tokio::sync::Mutex::new(CacheEntity::default())),
// Arc::new(tokio::sync::Mutex::new(CacheEntity::default())),
// Arc::new(tokio::sync::Mutex::new(CacheEntity::default())),
// );
// // Read 32 elements at a time
// let iter = items
// .iter()
// .map(|item| item.trim_start_matches("config/iam/"))
// .map(|item| split_path(item, item.starts_with("policydb/")))
// .filter_map(|(list_key, trimmed_item)| {
// debug!("list_key: {list_key}, trimmed_item: {trimmed_item}");
// if list_key == "format.json" {
// return None;
// }
// let (policy_docs, users, user_policies, sts_policies, sts_accounts) = (
// policy_docs.clone(),
// users.clone(),
// user_policies.clone(),
// sts_policies.clone(),
// sts_accounts.clone(),
// );
// Some(async move {
// match list_key {
// "policies/" => {
// let trimmed_item = dir(trimmed_item);
// let name = trimmed_item.trim_end_matches('/');
// let policy_doc = self.load_policy(name).await?;
// policy_docs.lock().await.insert(name.to_owned(), policy_doc);
// }
// "users/" => {
// let name = dir(trimmed_item);
// if let Some(user) = self.load_user_identity(UserType::Reg, &name).await? {
// users.lock().await.insert(name.to_owned(), user);
// };
// }
// "groups/" => {}
// "policydb/users/" | "policydb/groups/" => {
// let name = trimmed_item.strip_suffix(".json").unwrap_or(trimmed_item);
// let mapped_policy = self
// .load_mapped_policy(UserType::Reg, name, list_key == "policydb/groups/")
// .await?;
// if !mapped_policy.policies.is_empty() {
// user_policies.lock().await.insert(name.to_owned(), mapped_policy);
// }
// }
// "service-accounts/" => {
// let trimmed_item = dir(trimmed_item);
// let name = trimmed_item.trim_end_matches('/');
// let Some(user) = self.load_user_identity(UserType::Svc, name).await? else {
// return Ok(());
// };
// let parent = user.credentials.parent_user.clone();
// {
// users.lock().await.insert(name.to_owned(), user);
// }
// if users.lock().await.get(&parent).is_some() {
// return Ok(());
// }
// match self.load_mapped_policy(UserType::Sts, parent.as_str(), false).await {
// Ok(m) => sts_policies.lock().await.insert(name.to_owned(), m),
// Err(Error::EcstoreError(e)) if is_err_config_not_found(&e) => return Ok(()),
// Err(e) => return Err(e),
// };
// }
// "sts/" => {
// let name = dir(trimmed_item);
// if let Some(user) = self.load_user_identity(UserType::Sts, &name).await? {
// warn!("sts_accounts insert {}, user {:?}", name, &user.credentials.access_key);
// sts_accounts.lock().await.insert(name.to_owned(), user);
// };
// }
// "policydb/sts-users/" => {
// let name = trimmed_item.strip_suffix(".json").unwrap_or(trimmed_item);
// let mapped_policy = self.load_mapped_policy(UserType::Sts, name, false).await?;
// if !mapped_policy.policies.is_empty() {
// sts_policies.lock().await.insert(name.to_owned(), mapped_policy);
// }
// }
// _ => {}
// }
// Result::Ok(())
// })
// });
// let mut all_futures = Vec::with_capacity(32);
// for f in iter {
// all_futures.push(f);
// if all_futures.len() == 32 {
// try_join_all(all_futures).await?;
// all_futures = Vec::with_capacity(32);
// }
// }
// if !all_futures.is_empty() {
// try_join_all(all_futures).await?;
// }
// if let Some(x) = Arc::into_inner(users) {
// cache.users.store(Arc::new(x.into_inner().update_load_time()))
// }
// if let Some(x) = Arc::into_inner(policy_docs) {
// cache.policy_docs.store(Arc::new(x.into_inner().update_load_time()))
// }
// if let Some(x) = Arc::into_inner(user_policies) {
// cache.user_policies.store(Arc::new(x.into_inner().update_load_time()))
// }
// if let Some(x) = Arc::into_inner(sts_policies) {
// cache.sts_policies.store(Arc::new(x.into_inner().update_load_time()))
// }
// if let Some(x) = Arc::into_inner(sts_accounts) {
// cache.sts_accounts.store(Arc::new(x.into_inner().update_load_time()))
// }
// Ok(())
// }
}
#[cfg(test)]
crates/iam/src/sys.rs
+163 -34
View File
@@ -1325,7 +1325,7 @@ mod tests {
use rustfs_policy::policy::action::{Action, AdminAction, S3Action};
use rustfs_policy::policy::policy_uses_existing_object_tag_conditions;
use serde_json::Value;
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use time::OffsetDateTime;
#[test]
@@ -1410,6 +1410,10 @@ mod tests {
}
async fn load_user(&self, name: &str, user_type: UserType, m: &mut HashMap<String, UserIdentity>) -> Result<()> {
if name == "deleted-notify-user" {
return Err(Error::NoSuchUser(name.to_string()));
}
if user_type == UserType::Reg && name == "load-failure-user" {
return Err(Error::Io(std::io::Error::other("load user failed")));
}
@@ -1442,7 +1446,10 @@ mod tests {
Err(Error::InvalidArgument)
}
async fn load_group(&self, _name: &str, _m: &mut HashMap<String, GroupInfo>) -> Result<()> {
async fn load_group(&self, name: &str, m: &mut HashMap<String, GroupInfo>) -> Result<()> {
if name == "notify-group" {
m.insert(name.to_string(), GroupInfo::new(vec!["notify-user".to_string()]));
}
Ok(())
}
@@ -1495,6 +1502,9 @@ mod tests {
if user_type == UserType::Reg && !is_group && name == "notify-user" {
m.insert(name.to_string(), MappedPolicy::new("readwrite"));
}
if user_type == UserType::Sts && !is_group && name == "notify-sts-parent" {
m.insert(name.to_string(), MappedPolicy::new("readwrite"));
}
Ok(())
}
@@ -1512,9 +1522,6 @@ mod tests {
let custom_claim_policy =
Policy::parse_config(CUSTOM_STS_CLAIM_POLICY_JSON.as_bytes()).expect("custom STS claim policy should parse");
policy_docs.insert(CUSTOM_STS_CLAIM_POLICY.to_string(), PolicyDoc::new(custom_claim_policy));
cache
.policy_docs
.store(Arc::new(CacheEntity::new(policy_docs).update_load_time()));
if self.empty_policies {
const PARENT_USER: &str = "sts-empty-parent-policy-test";
@@ -1537,16 +1544,17 @@ mod tests {
};
let mut users = HashMap::new();
users.insert(PARENT_USER.to_string(), parent_identity);
cache.users.store(Arc::new(CacheEntity::new(users).update_load_time()));
cache.groups.store(Arc::new(CacheEntity::default().update_load_time()));
cache
.group_policies
.store(Arc::new(CacheEntity::default().update_load_time()));
cache.user_policies.store(Arc::new(CacheEntity::default().update_load_time()));
cache.sts_accounts.store(Arc::new(CacheEntity::default().update_load_time()));
cache.sts_policies.store(Arc::new(CacheEntity::default().update_load_time()));
cache.build_user_group_memberships();
cache.with_write_lock(|cache| {
cache.replace_policy_docs(CacheEntity::new(policy_docs));
cache.replace_users(CacheEntity::new(users));
cache.replace_groups(CacheEntity::default());
cache.replace_group_policies(CacheEntity::default());
cache.replace_user_policies(CacheEntity::default());
cache.replace_sts_accounts(CacheEntity::default());
cache.replace_sts_policies(CacheEntity::default());
cache.build_user_group_memberships();
});
return Ok(());
}
@@ -1572,24 +1580,25 @@ mod tests {
};
let mut users = HashMap::new();
users.insert(PARENT_USER.to_string(), parent_identity);
cache.users.store(Arc::new(CacheEntity::new(users).update_load_time()));
let group = GroupInfo::new(vec![PARENT_USER.to_string()]);
let mut groups = HashMap::new();
groups.insert(GROUP_NAME.to_string(), group);
cache.groups.store(Arc::new(CacheEntity::new(groups).update_load_time()));
let group_policy = MappedPolicy::new("readwrite");
let mut group_policies = HashMap::new();
group_policies.insert(GROUP_NAME.to_string(), group_policy);
cache
.group_policies
.store(Arc::new(CacheEntity::new(group_policies).update_load_time()));
cache.user_policies.store(Arc::new(CacheEntity::default().update_load_time()));
cache.sts_accounts.store(Arc::new(CacheEntity::default().update_load_time()));
cache.sts_policies.store(Arc::new(CacheEntity::default().update_load_time()));
cache.build_user_group_memberships();
cache.with_write_lock(|cache| {
cache.replace_policy_docs(CacheEntity::new(policy_docs));
cache.replace_users(CacheEntity::new(users));
cache.replace_groups(CacheEntity::new(groups));
cache.replace_group_policies(CacheEntity::new(group_policies));
cache.replace_user_policies(CacheEntity::default());
cache.replace_sts_accounts(CacheEntity::default());
cache.replace_sts_policies(CacheEntity::default());
cache.build_user_group_memberships();
});
Ok(())
}
@@ -1950,7 +1959,10 @@ mod tests {
parent_user: parent_user.to_string(),
..Default::default()
});
Cache::add_or_update(&iam_sys.store.cache.sts_accounts, sts_access_key, &sts_user, OffsetDateTime::now_utc());
iam_sys
.store
.cache
.add_or_update_sts_account(sts_access_key, &sts_user, OffsetDateTime::now_utc());
let mut claims = HashMap::new();
claims.insert(POLICYNAME.to_string(), Value::String(CUSTOM_STS_CLAIM_POLICY.to_string()));
@@ -1991,7 +2003,10 @@ mod tests {
parent_user: parent_user.to_string(),
..Default::default()
});
Cache::add_or_update(&iam_sys.store.cache.sts_accounts, sts_access_key, &sts_user, OffsetDateTime::now_utc());
iam_sys
.store
.cache
.add_or_update_sts_account(sts_access_key, &sts_user, OffsetDateTime::now_utc());
let mut claims = HashMap::new();
claims.insert(
@@ -2035,7 +2050,10 @@ mod tests {
parent_user: parent_user.to_string(),
..Default::default()
});
Cache::add_or_update(&iam_sys.store.cache.sts_accounts, sts_access_key, &sts_user, OffsetDateTime::now_utc());
iam_sys
.store
.cache
.add_or_update_sts_account(sts_access_key, &sts_user, OffsetDateTime::now_utc());
let mut claims = HashMap::new();
claims.insert(POLICYNAME.to_string(), Value::String(CUSTOM_STS_CLAIM_POLICY.to_string()));
@@ -2227,6 +2245,113 @@ mod tests {
);
}
#[tokio::test]
async fn test_group_notification_populates_new_membership_entry() {
let store = StsTestMockStore { empty_policies: false };
let cache_manager = IamCache::new(store).await.unwrap();
let iam_sys = IamSys::new(cache_manager);
iam_sys.store.cache.with_write_lock(|cache| {
cache.replace_user_group_memberships(CacheEntity::default());
});
iam_sys.load_group("notify-group").await.unwrap();
let cache = iam_sys.store.cache.snapshot();
let memberships = &cache.user_group_memberships;
assert!(
memberships
.get("notify-user")
.is_some_and(|groups| groups.contains("notify-group")),
"group notification must create a reverse membership entry for first-time members"
);
}
#[tokio::test]
async fn test_sts_policy_mapping_notification_updates_sts_policy_cache() {
let store = StsTestMockStore { empty_policies: false };
let cache_manager = IamCache::new(store).await.unwrap();
let iam_sys = IamSys::new(cache_manager);
iam_sys
.load_policy_mapping("notify-sts-parent", UserType::Sts, false)
.await
.unwrap();
let cache = iam_sys.store.cache.snapshot();
let sts_policies = &cache.sts_policies;
assert!(
sts_policies.contains_key("notify-sts-parent"),
"STS policy mapping notifications must update sts_policies instead of deleting them"
);
}
#[tokio::test]
async fn test_missing_user_notification_cleans_related_cache_state() {
let store = StsTestMockStore { empty_policies: false };
let cache_manager = IamCache::new(store).await.unwrap();
let iam_sys = IamSys::new(cache_manager);
const USER: &str = "deleted-notify-user";
const GROUP: &str = "deleted-notify-group";
const SVC_CHILD: &str = "deleted-notify-service-child";
const STS_CHILD: &str = "deleted-notify-sts-child";
const OTHER_USER: &str = "deleted-notify-other-user";
let user = UserIdentity::from(Credentials {
access_key: USER.to_string(),
secret_key: "longenoughsecret".to_string(),
status: ACCOUNT_ON.to_string(),
..Default::default()
});
let mut service_claims = HashMap::new();
service_claims.insert(iam_policy_claim_name_sa(), Value::String(INHERITED_POLICY_TYPE.to_string()));
let service_child = UserIdentity::from(Credentials {
access_key: SVC_CHILD.to_string(),
secret_key: "longenoughsecret".to_string(),
status: ACCOUNT_ON.to_string(),
parent_user: USER.to_string(),
claims: Some(service_claims),
..Default::default()
});
let sts_child = UserIdentity::from(Credentials {
access_key: STS_CHILD.to_string(),
secret_key: "longenoughsecret".to_string(),
session_token: "session-token".to_string(),
status: ACCOUNT_ON.to_string(),
parent_user: USER.to_string(),
..Default::default()
});
let membership = HashSet::from([GROUP.to_string()]);
let group = GroupInfo::new(vec![USER.to_string(), OTHER_USER.to_string()]);
let mapped_policy = MappedPolicy::new("readwrite");
iam_sys.store.cache.with_write_lock(|cache| {
let now = OffsetDateTime::now_utc();
cache.add_or_update_user(USER, &user, now);
cache.add_or_update_user_policy(USER, &mapped_policy, now);
cache.add_or_update_group(GROUP, &group, now);
cache.add_or_update_user_group_membership(USER, &membership, now);
cache.add_or_update_user(SVC_CHILD, &service_child, now);
cache.add_or_update_sts_account(STS_CHILD, &sts_child, now);
});
iam_sys.load_user(USER, UserType::Reg).await.unwrap();
let cache = iam_sys.store.cache.snapshot();
assert!(!cache.users.contains_key(USER));
assert!(!cache.user_policies.contains_key(USER));
assert!(!cache.user_group_memberships.contains_key(USER));
assert!(!cache.users.contains_key(SVC_CHILD));
assert!(!cache.sts_accounts.contains_key(STS_CHILD));
let group = cache.groups.get(GROUP).expect("group should remain after member removal");
assert!(!group.members.contains(&USER.to_string()));
assert!(group.members.contains(&OTHER_USER.to_string()));
}
#[tokio::test]
async fn test_check_key_propagates_cache_miss_load_failure() {
let store = StsTestMockStore { empty_policies: false };
@@ -2281,7 +2406,10 @@ mod tests {
parent_user: "sts-empty-parent-policy-test".to_string(),
..Default::default()
});
Cache::add_or_update(&iam_sys.store.cache.sts_accounts, sts_access_key, &sts_user, OffsetDateTime::now_utc());
iam_sys
.store
.cache
.add_or_update_sts_account(sts_access_key, &sts_user, OffsetDateTime::now_utc());
let mut claims = HashMap::new();
claims.insert(
@@ -2395,7 +2523,10 @@ mod tests {
parent_user: "sts-empty-parent-policy-test".to_string(),
..Default::default()
});
Cache::add_or_update(&iam_sys.store.cache.sts_accounts, sts_access_key, &sts_user, OffsetDateTime::now_utc());
iam_sys
.store
.cache
.add_or_update_sts_account(sts_access_key, &sts_user, OffsetDateTime::now_utc());
let session_policy_json = r#"{
"Version":"2012-10-17",
@@ -2445,12 +2576,10 @@ mod tests {
claims: Some(service_account_claims),
..Default::default()
});
Cache::add_or_update(
&iam_sys.store.cache.users,
service_account_access_key,
&service_identity,
OffsetDateTime::now_utc(),
);
iam_sys
.store
.cache
.add_or_update_user(service_account_access_key, &service_identity, OffsetDateTime::now_utc());
let mut request_claims = HashMap::new();
request_claims.insert("parent".to_string(), Value::String(parent_user.to_string()));