fix(iam): close identity, session and admin-authorization gaps (#9163)

Scan findings in the IAM crate and the admin identity handlers, grouped
because they share the IamCache mutation lock, the STS session sweep and
the admin client-address plumbing:

- Admin and STS authorization used the socket peer as aws:SourceIp
  behind a trusted proxy; they now read the verified client address the
  S3 path already uses (backlog#3025), and the admin usage route passes
  the bucket named in the path to the s3:ListBucket check (backlog#3033).
- AssumeRoleWithWebIdentity accepted HS256 ID tokens MACed with the OIDC
  client secret; only provider-signed tokens verify (backlog#3028), and
  JWKS refresh is single-flight with a failure cool-down (backlog#3103).
- A disabled group emptied the member's whole policy list, dropping
  every Deny; direct and other-group policies are preserved (backlog#3042).
- Identity rewrites (secret change, add-user, status) ran without a lock
  and without re-reading the record, so a user's password change could
  undo an admin disable or reset (backlog#3048). add-user on an existing
  user and in-flight AssumeRole both left sessions minted under the old
  secret valid (backlog#3071, backlog#3079); rotation now revokes under
  the per-key mutation lock and derived-credential publication is ordered
  with it.
- aws:groups never carried a built-in user's IAM group membership and
  jwt:iss was always the local marker, so group- and issuer-scoped
  Denies never matched; both now come from live IAM state
  (backlog#3095, backlog#3096).
- OPA mode skipped the account-status gates for service accounts and STS
  sessions of a disabled user (backlog#3057).
- The periodic IAM reload published nothing when any cache write raced
  it and kept stale partitions when a prefix listed empty, so a missed
  revocation never applied (backlog#3064, backlog#3114).
- update-group-members never evaluated admin:RemoveUserFromGroup
  (backlog#3098); ExportIAM compared the site-replicator account against
  a name it never has (backlog#3045); add-user accepted the access key of
  a live STS session or service account (backlog#3134); a service account
  could create or extend a lifetime beyond its own (backlog#3112);
  sts:DurationSeconds and svc:DurationSeconds were never populated
  (backlog#3136); temporary-account-info returned service accounts
  (backlog#3137); an access key the object layer refuses no longer holds
  the service-account lock through the write retries (backlog#3133).
This commit is contained in:
Chris An
2026-11-04 18:44:50 +08:00
committed by GitHub
parent e0f26a540d
commit 73e7afa01a
33 changed files with 3011 additions and 380 deletions
+387 -81
View File
@@ -15,7 +15,6 @@
use std::{
collections::{HashMap, HashSet},
ops::{Deref, DerefMut},
ptr,
sync::{Arc, Mutex, Weak},
};
@@ -64,21 +63,46 @@ impl Default for CacheState {
pub struct Cache {
state: ArcSwap<CacheState>,
write_lock: Mutex<()>,
reload_lock: AsyncMutex<()>,
reload_changes: Mutex<Weak<Mutex<ReloadChanges>>>,
service_account_mutation_lock: AsyncMutex<()>,
sts_account_mutation_locks: Arc<StsMutationLockRegistry>,
sts_account_mutation_locks: Arc<MutationLockRegistry>,
parent_mutation_locks: Arc<MutationLockRegistry>,
}
struct StsMutationLockRegistry {
locks: Mutex<HashMap<String, Weak<AsyncMutex<StsMutationLockState>>>>,
#[derive(Default)]
struct MutationLockRegistry {
locks: Mutex<HashMap<String, Weak<AsyncMutex<MutationLockState>>>>,
}
pub(crate) struct StsMutationLockState {
pub(crate) struct MutationLockState {
access_key: String,
registry: Weak<StsMutationLockRegistry>,
lock: Weak<AsyncMutex<StsMutationLockState>>,
registry: Weak<MutationLockRegistry>,
lock: Weak<AsyncMutex<MutationLockState>>,
}
impl Drop for StsMutationLockState {
impl MutationLockRegistry {
fn lock_for(self: &Arc<Self>, access_key: &str) -> Arc<AsyncMutex<MutationLockState>> {
let mut locks = self.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);
let access_key_owned = access_key.to_string();
let lock = Arc::new_cyclic(|lock| {
AsyncMutex::new(MutationLockState {
access_key: access_key_owned,
registry,
lock: lock.clone(),
})
});
locks.insert(access_key.to_string(), Arc::downgrade(&lock));
lock
}
}
impl Drop for MutationLockState {
fn drop(&mut self) {
let Some(registry) = self.registry.upgrade() else {
return;
@@ -95,10 +119,52 @@ impl Default for Cache {
Self {
state: ArcSwap::new(Arc::new(CacheState::default())),
write_lock: Mutex::new(()),
reload_lock: AsyncMutex::new(()),
reload_changes: Mutex::new(Weak::new()),
service_account_mutation_lock: AsyncMutex::new(()),
sts_account_mutation_locks: Arc::new(StsMutationLockRegistry {
locks: Mutex::new(HashMap::new()),
}),
sts_account_mutation_locks: Arc::new(MutationLockRegistry::default()),
parent_mutation_locks: Arc::new(MutationLockRegistry::default()),
}
}
}
#[derive(Clone, Copy, Hash, PartialEq, Eq)]
pub(crate) enum CachePartition {
PolicyDocs,
Users,
UserPolicies,
StsAccounts,
StsPolicies,
Groups,
UserGroupMemberships,
GroupPolicies,
}
type ReloadChanges = HashMap<CachePartition, HashSet<String>>;
/// A reload journals only keys changed during its own lifetime, including
/// deletions. Unrelated STS traffic cannot suppress a persisted revocation.
pub(crate) struct CacheReload<'a> {
_guard: tokio::sync::MutexGuard<'a, ()>,
changes: Arc<Mutex<ReloadChanges>>,
}
impl CacheReload<'_> {
pub(crate) fn merge_partition<T: Clone>(
&self,
partition: CachePartition,
loaded: &mut CacheEntity<T>,
current: &CacheEntity<T>,
) {
let changes = self.changes.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(keys) = changes.get(&partition) {
for key in keys {
if let Some(value) = current.get(key) {
loaded.insert(key.clone(), value.clone());
} else {
loaded.remove(key);
}
}
}
}
}
@@ -106,31 +172,27 @@ impl Default for Cache {
pub(crate) type CacheSnapshot = Guard<Arc<CacheState>>;
impl Cache {
pub(crate) async fn begin_reload(&self) -> CacheReload<'_> {
// Registration and publication both take write_lock before the journal
// lock, so no mutation can fall between the reload and its journal.
let guard = self.reload_lock.lock().await;
let changes = Arc::new(Mutex::new(HashMap::new()));
let _write = self.write_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
*self.reload_changes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = Arc::downgrade(&changes);
CacheReload { _guard: guard, changes }
}
pub(crate) fn service_account_mutation_lock(&self) -> &AsyncMutex<()> {
&self.service_account_mutation_lock
}
pub(crate) fn sts_account_mutation_lock(&self, access_key: &str) -> Arc<AsyncMutex<StsMutationLockState>> {
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;
}
pub(crate) fn sts_account_mutation_lock(&self, access_key: &str) -> Arc<AsyncMutex<MutationLockState>> {
self.sts_account_mutation_locks.lock_for(access_key)
}
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
/// Orders derived-credential publication with changes to its regular parent.
pub(crate) fn parent_mutation_lock(&self, access_key: &str) -> Arc<AsyncMutex<MutationLockState>> {
self.parent_mutation_locks.lock_for(access_key)
}
pub(crate) fn snapshot(&self) -> CacheSnapshot {
@@ -142,7 +204,13 @@ impl Cache {
let current = self.state.load_full();
let mut locked = LockedCache {
state: CacheState::clone(&current),
#[cfg(test)]
current_ptr: Arc::as_ptr(&current),
reload_changes: self
.reload_changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.upgrade(),
dirty: false,
};
let ret = f(&mut locked);
@@ -215,7 +283,9 @@ impl Cache {
pub(crate) struct LockedCache {
state: CacheState,
#[cfg(test)]
current_ptr: *const CacheState,
reload_changes: Option<Arc<Mutex<ReloadChanges>>>,
dirty: bool,
}
@@ -224,11 +294,19 @@ impl LockedCache {
&self.state
}
#[cfg(test)]
pub(crate) fn matches_snapshot(&self, snapshot: &CacheSnapshot) -> bool {
ptr::eq(self.current_ptr, Arc::as_ptr(snapshot))
std::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 {
fn exec<T: Clone>(
target: &mut Arc<CacheEntity<T>>,
t: OffsetDateTime,
changes: Option<&Arc<Mutex<ReloadChanges>>>,
partition: CachePartition,
key: &str,
mut op: impl FnMut(&mut CacheEntity<T>),
) -> bool {
if target.load_time >= t {
return false;
}
@@ -236,6 +314,14 @@ impl LockedCache {
let mut new = CacheEntity::clone(target);
op(&mut new);
*target = Arc::new(new);
if let Some(changes) = changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(partition)
.or_default()
.insert(key.to_string());
}
true
}
@@ -244,139 +330,315 @@ impl LockedCache {
}
pub(crate) fn replace_policy_docs(&mut self, value: CacheEntity<PolicyDoc>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::PolicyDocs)
.or_default()
.extend(self.state.policy_docs.keys().chain(value.keys()).cloned());
}
self.state.policy_docs = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_users(&mut self, value: CacheEntity<UserIdentity>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::Users)
.or_default()
.extend(self.state.users.keys().chain(value.keys()).cloned());
}
self.state.users = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_user_policies(&mut self, value: CacheEntity<MappedPolicy>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::UserPolicies)
.or_default()
.extend(self.state.user_policies.keys().chain(value.keys()).cloned());
}
self.state.user_policies = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_sts_accounts(&mut self, value: CacheEntity<UserIdentity>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::StsAccounts)
.or_default()
.extend(self.state.sts_accounts.keys().chain(value.keys()).cloned());
}
self.state.sts_accounts = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_sts_policies(&mut self, value: CacheEntity<MappedPolicy>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::StsPolicies)
.or_default()
.extend(self.state.sts_policies.keys().chain(value.keys()).cloned());
}
self.state.sts_policies = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_groups(&mut self, value: CacheEntity<GroupInfo>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::Groups)
.or_default()
.extend(self.state.groups.keys().chain(value.keys()).cloned());
}
self.state.groups = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_group_policies(&mut self, value: CacheEntity<MappedPolicy>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::GroupPolicies)
.or_default()
.extend(self.state.group_policies.keys().chain(value.keys()).cloned());
}
self.state.group_policies = Self::replaced(value);
self.dirty = true;
}
pub(crate) fn replace_user_group_memberships(&mut self, value: CacheEntity<HashSet<String>>) {
if let Some(changes) = &self.reload_changes {
changes
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.entry(CachePartition::UserGroupMemberships)
.or_default()
.extend(self.state.user_group_memberships.keys().chain(value.keys()).cloned());
}
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());
});
self.dirty |= Self::exec(
&mut self.state.policy_docs,
t,
self.reload_changes.as_ref(),
CachePartition::PolicyDocs,
key,
|map| {
map.insert(key.to_string(), value.clone());
},
);
}
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());
});
self.dirty |= Self::exec(
&mut self.state.users,
t,
self.reload_changes.as_ref(),
CachePartition::Users,
key,
|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());
});
self.dirty |= Self::exec(
&mut self.state.user_policies,
t,
self.reload_changes.as_ref(),
CachePartition::UserPolicies,
key,
|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());
});
self.dirty |= Self::exec(
&mut self.state.sts_accounts,
t,
self.reload_changes.as_ref(),
CachePartition::StsAccounts,
key,
|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());
});
self.dirty |= Self::exec(
&mut self.state.sts_policies,
t,
self.reload_changes.as_ref(),
CachePartition::StsPolicies,
key,
|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());
});
self.dirty |= Self::exec(
&mut self.state.groups,
t,
self.reload_changes.as_ref(),
CachePartition::Groups,
key,
|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());
});
self.dirty |= Self::exec(
&mut self.state.user_group_memberships,
t,
self.reload_changes.as_ref(),
CachePartition::UserGroupMemberships,
key,
|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());
});
self.dirty |= Self::exec(
&mut self.state.group_policies,
t,
self.reload_changes.as_ref(),
CachePartition::GroupPolicies,
key,
|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);
});
self.dirty |= Self::exec(
&mut self.state.policy_docs,
t,
self.reload_changes.as_ref(),
CachePartition::PolicyDocs,
key,
|map| {
map.remove(key);
},
);
}
pub(crate) fn delete_user(&mut self, key: &str, t: OffsetDateTime) {
self.dirty |= Self::exec(&mut self.state.users, t, |map| {
map.remove(key);
});
self.dirty |= Self::exec(
&mut self.state.users,
t,
self.reload_changes.as_ref(),
CachePartition::Users,
key,
|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);
});
self.dirty |= Self::exec(
&mut self.state.user_policies,
t,
self.reload_changes.as_ref(),
CachePartition::UserPolicies,
key,
|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);
});
self.dirty |= Self::exec(
&mut self.state.user_group_memberships,
t,
self.reload_changes.as_ref(),
CachePartition::UserGroupMemberships,
key,
|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);
});
self.dirty |= Self::exec(
&mut self.state.sts_accounts,
t,
self.reload_changes.as_ref(),
CachePartition::StsAccounts,
key,
|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);
});
self.dirty |= Self::exec(
&mut self.state.sts_policies,
t,
self.reload_changes.as_ref(),
CachePartition::StsPolicies,
key,
|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);
});
self.dirty |= Self::exec(
&mut self.state.groups,
t,
self.reload_changes.as_ref(),
CachePartition::Groups,
key,
|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);
});
self.dirty |= Self::exec(
&mut self.state.group_policies,
t,
self.reload_changes.as_ref(),
CachePartition::GroupPolicies,
key,
|map| {
map.remove(key);
},
);
}
pub(crate) fn build_user_group_memberships(&mut self) {
@@ -494,6 +756,50 @@ mod tests {
use crate::cache::Cache;
use crate::store::MappedPolicy;
#[tokio::test]
async fn reload_journal_preserves_same_key_mutations_and_expires_on_cancellation() {
use super::{CacheEntity, CachePartition};
let cache = Cache::default();
let stale = UserIdentity {
version: 1,
..Default::default()
};
let current = UserIdentity {
version: 2,
..Default::default()
};
let reload = cache.begin_reload().await;
cache.add_or_update_user("updated", &current, OffsetDateTime::now_utc());
cache.add_or_update_user("removed", &current, OffsetDateTime::now_utc());
cache.delete_user("removed", OffsetDateTime::now_utc());
// Even an absent key's deletion is recorded: storage may have returned
// an entry created and deleted entirely during the reload's reads.
cache.delete_user("absent", OffsetDateTime::now_utc());
let mut loaded = CacheEntity::new(std::collections::HashMap::from([
("updated".to_string(), stale.clone()),
("removed".to_string(), stale.clone()),
("absent".to_string(), stale.clone()),
("disk-only".to_string(), stale),
]));
cache.with_write_lock(|locked| {
reload.merge_partition(CachePartition::Users, &mut loaded, &locked.state().users);
locked.replace_users(loaded);
});
let snapshot = cache.snapshot();
assert_eq!(snapshot.users["updated"].version, 2);
assert!(!snapshot.users.contains_key("removed"));
assert!(!snapshot.users.contains_key("absent"));
assert!(snapshot.users.contains_key("disk-only"));
drop(reload);
assert!(cache.reload_changes.lock().expect("journal lock").upgrade().is_none());
let cancelled = cache.begin_reload().await;
cache.delete_user("disk-only", OffsetDateTime::now_utc());
drop(cancelled);
assert!(cache.reload_changes.lock().expect("journal lock").upgrade().is_none());
let next = cache.begin_reload().await;
assert!(next.changes.lock().expect("new journal").is_empty());
}
#[test]
fn sts_mutation_locks_are_keyed_and_prune_unused_entries() {
let cache = Cache::default();
+10
View File
@@ -16,6 +16,7 @@ use super::{FederatedClaims, VerifiedFederatedIdentity};
use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
use rustfs_utils::HashAlgorithm;
use serde_json::Value;
use std::{collections::HashMap, sync::Arc};
use tracing::debug;
@@ -66,6 +67,15 @@ impl FederatedAuthorization {
&self.provider_id
}
/// Issuer from the verified upstream identity.
pub fn issuer(&self) -> Option<&str> {
self.claims
.raw
.get("iss")
.and_then(Value::as_str)
.filter(|issuer| !issuer.is_empty())
}
pub fn subject(&self) -> &str {
&self.claims.sub
}
+149 -7
View File
@@ -415,6 +415,16 @@ where
self.api.save_iam_config(IAMFormat::new_version_1(), path).await
}
/// Current built-in IAM group membership, independent of credential snapshots.
pub fn user_group_memberships(&self, name: &str) -> Vec<String> {
self.cache
.snapshot()
.user_group_memberships
.get(name)
.map(|groups| groups.iter().cloned().collect())
.unwrap_or_default()
}
pub async fn get_user(&self, access_key: &str) -> Option<UserIdentity> {
let cache = self.cache.snapshot();
cache
@@ -1005,6 +1015,17 @@ where
}
pub async fn sts_policy_db_get(&self, name: &str, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
self.sts_policy_db_get_with_identity_mapping(name, groups)
.await
.map(|(policies, _)| policies)
}
/// Resolve STS policies and whether that same lookup found a nonempty identity mapping.
pub async fn sts_policy_db_get_with_identity_mapping(
&self,
name: &str,
groups: &Option<Vec<String>>,
) -> Result<(Vec<String>, bool)> {
if name.is_empty() {
return Err(Error::InvalidArgument);
}
@@ -1031,7 +1052,10 @@ where
}
};
self.policy_db_get_with_groups(mapped_policy.to_slice(), groups).await
let direct_policies = mapped_policy.to_slice();
let has_identity_mapping = !direct_policies.is_empty();
let policies = self.policy_db_get_with_groups(direct_policies, groups).await?;
Ok((policies, has_identity_mapping))
}
async fn policy_db_get_with_groups(&self, mut policies: Vec<String>, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
@@ -1158,7 +1182,7 @@ where
if let Some(groups) = u.credentials.groups.as_ref() {
for group in groups.iter() {
if groups_cache.get(group).filter(|v| v.status == STATUS_DISABLED).is_some() {
return Ok((Vec::new(), OffsetDateTime::now_utc()));
continue;
}
let mp = match group_policies.get(group) {
@@ -1189,7 +1213,7 @@ where
for group in user_group_memberships.get(name).cloned().unwrap_or_default().iter() {
if groups_cache.get(group).filter(|v| v.status == STATUS_DISABLED).is_some() {
return Ok((Vec::new(), OffsetDateTime::now_utc()));
continue;
}
let mp = match group_policies.get(group) {
@@ -1486,7 +1510,27 @@ where
args: &AddOrUpdateUserReq,
updated_at: OffsetDateTime,
) -> Result<OffsetDateTime> {
self.add_user_at_with_secret_change(access_key, args, updated_at)
.await
.map(|(updated_at, _, _guard)| updated_at)
}
/// Return whether this write replaced a different regular-user secret.
pub(crate) async fn add_user_at_with_secret_change(
&self,
access_key: &str,
args: &AddOrUpdateUserReq,
updated_at: OffsetDateTime,
) -> Result<(OffsetDateTime, bool, tokio::sync::OwnedMutexGuard<crate::cache::MutationLockState>)> {
let parent_lock = self.cache.parent_mutation_lock(access_key);
let parent_guard = parent_lock.lock_owned().await;
let _service_guard = self.cache.service_account_mutation_lock().lock().await;
let sts_lock = self.cache.sts_account_mutation_lock(access_key);
let _sts_guard = sts_lock.lock().await;
let cache = self.cache.snapshot();
if cache.sts_accounts.contains_key(access_key) {
return Err(Error::AccessKeyAlreadyExists);
}
let users = Arc::clone(&cache.users);
if let Some(x) = users.get(access_key) {
debug!(
@@ -1496,10 +1540,13 @@ where
state = "existing",
"Updating existing IAM user"
);
if x.credentials.is_temp() {
if x.credentials.is_temp() || x.credentials.is_service_account() {
return Err(Error::AccessKeyAlreadyExists);
}
}
let secret_changed = users
.get(access_key)
.is_some_and(|user| user.credentials.secret_key != args.secret_key);
drop(cache);
drop(users);
@@ -1523,7 +1570,7 @@ where
self.update_user_with_claims(access_key, user_entry)?;
Ok(updated_at)
Ok((updated_at, secret_changed, parent_guard))
}
pub async fn delete_user(&self, access_key: &str, utype: UserType) -> Result<()> {
@@ -1540,12 +1587,17 @@ where
return Err(Error::InvalidArgument);
}
let parent_lock = (utype == UserType::Reg).then(|| self.cache.parent_mutation_lock(access_key));
let _parent_guard = match &parent_lock {
Some(lock) => Some(lock.lock().await),
None => None,
};
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 {
let _service_account_guard = if matches!(utype, UserType::Svc | UserType::Reg) {
Some(self.cache.service_account_mutation_lock().lock().await)
} else {
None
@@ -1640,17 +1692,46 @@ where
}
pub async fn update_user_secret_key(&self, access_key: &str, secret_key: &str) -> Result<(OffsetDateTime, AccountStatus)> {
self.update_user_secret_key_with_guard(access_key, secret_key, None)
.await
.map(|(updated_at, status, _guard)| (updated_at, status))
}
pub(crate) async fn update_user_secret_key_with_guard(
&self,
access_key: &str,
secret_key: &str,
expected_secret: Option<&str>,
) -> Result<(
OffsetDateTime,
AccountStatus,
tokio::sync::OwnedMutexGuard<crate::cache::MutationLockState>,
)> {
if access_key.is_empty() || secret_key.is_empty() {
return Err(Error::InvalidArgument);
}
let parent_lock = self.cache.parent_mutation_lock(access_key);
let parent_guard = parent_lock.lock_owned().await;
let cache = self.cache.snapshot();
let users = Arc::clone(&cache.users);
let u = match users.get(access_key) {
Some(u) => u,
None if expected_secret.is_some() => return Err(Error::IAMActionNotAllowed),
None => return Err(Error::NoSuchUser(access_key.to_string())),
};
if let Some(expected) = expected_secret {
use subtle::ConstantTimeEq as _;
if !u.credentials.is_valid()
|| u.credentials.is_temp()
|| u.credentials.is_service_account()
|| !bool::from(u.credentials.secret_key.as_bytes().ct_eq(expected.as_bytes()))
{
return Err(Error::IAMActionNotAllowed);
}
}
let mut cred = u.credentials.clone();
cred.secret_key = secret_key.to_string();
@@ -1672,7 +1753,7 @@ where
.await?;
self.update_user_with_claims(access_key, u)?;
Ok((updated_at, status))
Ok((updated_at, status, parent_guard))
}
/// Add SSH public key for a user (for SFTP authentication)
@@ -1681,6 +1762,8 @@ where
return Err(Error::InvalidArgument);
}
let parent_lock = self.cache.parent_mutation_lock(access_key);
let _parent_guard = parent_lock.lock().await;
let cache = self.cache.snapshot();
let users = Arc::clone(&cache.users);
let u = match users.get(access_key) {
@@ -1720,6 +1803,8 @@ where
return Err(Error::InvalidArgument);
}
let parent_lock = self.cache.parent_mutation_lock(access_key);
let _parent_guard = parent_lock.lock().await;
let cache = self.cache.snapshot();
let users = Arc::clone(&cache.users);
let u = match users.get(access_key) {
@@ -2238,6 +2323,11 @@ where
Ok(())
}
pub async fn user_notification_handler(&self, name: &str, user_type: UserType) -> Result<()> {
let parent_lock = (user_type == UserType::Reg).then(|| self.cache.parent_mutation_lock(name));
let _parent_guard = match &parent_lock {
Some(lock) => Some(lock.lock().await),
None => None,
};
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),
@@ -2985,6 +3075,58 @@ mod tests {
}
}
#[tokio::test]
async fn disabled_group_preserves_direct_and_other_group_policies() {
for credential_groups in [false, true] {
let iam = build_test_iam_cache(FailingInitialLoadStore);
let groups = vec!["disabled".to_string(), "enabled".to_string()];
let user = UserIdentity::new(Credentials {
access_key: "alice".to_string(),
secret_key: "alice-secret".to_string(),
status: STATUS_ENABLED.to_string(),
groups: credential_groups.then(|| groups.clone()),
..Default::default()
});
let now = OffsetDateTime::now_utc();
iam.cache.add_or_update_user("alice", &user, now);
iam.cache
.add_or_update_user_policy("alice", &MappedPolicy::new("direct-deny"), now);
if !credential_groups {
iam.cache
.add_or_update_user_group_membership("alice", &groups.iter().cloned().collect(), now);
}
for (group, status, policy) in [
("disabled", STATUS_DISABLED, "disabled-grant"),
("enabled", STATUS_ENABLED, "other-group-deny"),
] {
iam.cache.add_or_update_group(
group,
&GroupInfo {
status: status.to_string(),
members: vec!["alice".to_string()],
..Default::default()
},
now,
);
iam.cache.add_or_update_group_policy(group, &MappedPolicy::new(policy), now);
}
let policies: HashSet<_> = iam
.policy_db_get("alice", &None)
.await
.expect("user policy lookup")
.into_iter()
.collect();
assert_eq!(policies, HashSet::from(["direct-deny".to_string(), "other-group-deny".to_string()]));
assert!(
iam.policy_db_get_internal("disabled", true, false)
.await
.expect("disabled group lookup")
.0
.is_empty()
);
}
}
#[tokio::test]
async fn sts_policy_lookup_ignores_colliding_regular_user_mapping() {
let iam = build_test_iam_cache(FailingInitialLoadStore);
+59 -14
View File
@@ -18,18 +18,19 @@ use super::{
use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
use openidconnect::core::{CoreIdToken, CoreIdTokenClaims, CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
use openidconnect::{
AsyncHttpClient, ClaimsVerificationError, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, NonceVerifier,
AsyncHttpClient, ClaimsVerificationError, ClientId, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, NonceVerifier,
ProviderMetadataWithLogout, SignatureVerificationError, TokenUrl,
};
use serde::Deserialize;
use std::collections::HashMap;
use std::sync::RwLock;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration as StdDuration, Instant};
use tokio::time::sleep;
use tracing::{debug, warn};
use url::Url;
pub(super) const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60);
const OIDC_JWKS_REFRESH_COOLDOWN: StdDuration = StdDuration::from_secs(60);
pub(super) const OIDC_DISCOVERY_TRANSPORT_RETRIES: usize = 3;
pub(super) const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration::from_millis(50);
pub(super) const OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY: &str = "JWKS request blocked by outbound policy";
@@ -97,9 +98,14 @@ pub(super) struct ProviderState {
pub(super) discovered_at: Instant,
}
type RefreshAttempt = Option<(Instant, Result<ProviderState, String>)>;
type ProviderRefreshLock = Arc<tokio::sync::Mutex<RefreshAttempt>>;
#[derive(Clone)]
pub(super) struct ProviderRuntime {
states: RwLock<HashMap<String, ProviderState>>,
http_client: ReqwestHttpClient,
states: Arc<RwLock<HashMap<String, ProviderState>>>,
http_client: Arc<ReqwestHttpClient>,
refreshes: Arc<Mutex<HashMap<String, ProviderRefreshLock>>>,
}
// Workload issuers do not implement the browser authorization flow. Keep their
@@ -200,16 +206,25 @@ impl DiscoveredProviderMetadata {
pub(super) fn verifier(&self, config: &OidcProviderConfig) -> CoreIdTokenVerifier<'static> {
let client_id = ClientId::new(config.client_id.clone());
let secret = config.client_secret.as_ref().map(|secret| ClientSecret::new(secret.clone()));
let (issuer, jwks, algorithms) = match self {
Self::Console(metadata) => (metadata.issuer(), metadata.jwks(), metadata.id_token_signing_alg_values_supported()),
Self::Workload(metadata) => (&metadata.issuer, &metadata.jwks, &metadata.id_token_signing_alg_values_supported),
};
let verifier = match secret {
Some(secret) => CoreIdTokenVerifier::new_confidential_client(client_id, secret, issuer.clone(), jwks.clone()),
None => CoreIdTokenVerifier::new_public_client(client_id, issuer.clone(), jwks.clone()),
};
verifier.set_allowed_algs(algorithms.clone())
// Caller-supplied tokens must prove possession of the issuer's signing
// key. A client secret is shared with clients and cannot provide that proof.
let algorithms = algorithms
.iter()
.filter(|algorithm| {
!matches!(
algorithm,
CoreJwsSigningAlgorithm::HmacSha256
| CoreJwsSigningAlgorithm::HmacSha384
| CoreJwsSigningAlgorithm::HmacSha512
)
})
.cloned()
.collect::<Vec<_>>();
CoreIdTokenVerifier::new_public_client(client_id, issuer.clone(), jwks.clone()).set_allowed_algs(algorithms)
}
}
@@ -273,8 +288,9 @@ impl ProviderState {
impl ProviderRuntime {
pub(super) fn new(http_client: ReqwestHttpClient, states: HashMap<String, ProviderState>) -> Self {
Self {
states: RwLock::new(states),
http_client,
states: Arc::new(RwLock::new(states)),
http_client: Arc::new(http_client),
refreshes: Arc::new(Mutex::new(HashMap::new())),
}
}
@@ -326,9 +342,38 @@ impl ProviderRuntime {
provider_id: &str,
config: &OidcProviderConfig,
) -> Result<ProviderState, String> {
self.recover_provider_state(provider_id, config)
let refresh = self
.refreshes
.lock()
.map_err(|_| "provider refresh registry is unavailable".to_string())?
.entry(provider_id.to_string())
.or_default()
.clone();
let mut attempt = refresh.lock_owned().await;
if let Some((completed, result)) = attempt.as_ref()
&& completed.elapsed() < OIDC_JWKS_REFRESH_COOLDOWN
{
return result.clone();
}
// The admitted fetch owns the per-provider lock. A disconnected request
// cannot abandon discovery while consuming the provider's cooldown.
// Waiting callers share its completed result, including genuine failures.
let runtime = tokio::runtime::Handle::try_current().map_err(|err| err.to_string())?;
let provider_runtime = self.clone();
let provider_id = provider_id.to_string();
let config = config.clone();
runtime
.spawn(async move {
let result = provider_runtime
.recover_provider_state(&provider_id, &config)
.await
.map_err(|err| err.to_string());
*attempt = Some((Instant::now(), result.clone()));
result
})
.await
.map_err(|err| err.to_string())
.map_err(|err| err.to_string())?
}
pub(super) async fn recover_provider_state(
+162 -17
View File
@@ -180,7 +180,7 @@ fn workload_metadata_rejects_duplicate_fields() {
}
}
#[test]
fn workload_verifier_preserves_algorithm_secret_and_audience_policy() {
fn caller_supplied_tokens_cannot_be_signed_with_the_client_secret() {
let secret = Nonce::new_random().secret().to_string();
let mut config = build_mocked_oidc_provider_config("workload", "https://issuer.example.com");
config.client_secret = Some(secret.clone());
@@ -195,7 +195,8 @@ fn workload_verifier_preserves_algorithm_secret_and_audience_policy() {
let token: CoreIdToken = signed.parse().unwrap();
for workload in [false, true] {
for (algorithms, accepted) in [
(serde_json::json!(["HS256", "unsupported-future-algorithm"]), true),
(serde_json::json!(["RS256", "HS256"]), false),
(serde_json::json!(["HS256", "unsupported-future-algorithm"]), false),
(serde_json::json!(["ES256"]), false),
(serde_json::json!(["unsupported-future-algorithm"]), false),
] {
@@ -216,21 +217,6 @@ fn workload_verifier_preserves_algorithm_secret_and_audience_policy() {
accepted,
"workload={workload}, algorithms={algorithms}"
);
if accepted {
assert!(
token.claims(&metadata.verifier(&config), |_: Option<&Nonce>| Ok(())).is_err(),
"additional audiences require explicit trust"
);
let mut wrong_secret = config.clone();
wrong_secret.client_secret = Some(Nonce::new_random().secret().to_string());
let verifier = metadata
.verifier(&wrong_secret)
.set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
assert!(
token.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_err(),
"incorrect client secret must fail"
);
}
}
}
}
@@ -274,6 +260,19 @@ fn configured_verifier_rejects_malformed_hmac_signature_lengths() {
let metadata = DiscoveredProviderMetadata::parse(&serde_json::to_vec(&document).expect("encode metadata"), true)
.expect("parse configured provider metadata");
let verifier = metadata.verifier(&config);
assert!(
verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_err(),
"caller-supplied HMAC tokens must be rejected"
);
// The authorization-code flow obtains tokens directly from the
// provider and still supports confidential-client verification.
let verifier = openidconnect::core::CoreIdTokenVerifier::new_confidential_client(
openidconnect::ClientId::new(config.client_id.clone()),
openidconnect::ClientSecret::new(secret.to_owned()),
metadata.issuer().clone(),
openidconnect::core::CoreJsonWebKeySet::new(vec![]),
)
.set_allowed_algs(vec![serde_json::from_value(serde_json::json!(name)).expect("parse signing algorithm")]);
assert!(
verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
"valid {name} token must pass for workload={workload}"
@@ -780,3 +779,149 @@ async fn test_validate_oidc_provider_config_returns_detailed_errors() {
assert!(err.contains(&format!("issuer '{base}/application/o/rustfs/'")));
assert!(handle.join().is_ok());
}
#[tokio::test]
async fn jwks_refresh_is_single_flight_and_survives_request_cancellation() {
let (initial_key, initial_jwk) = oidc_es256_key_and_jwk("initial");
let (rotated_key, rotated_jwk) = oidc_es256_key_and_jwk("rotated");
let (wrong_key, _) = oidc_es256_key_and_jwk("wrong");
let requests = Arc::new(AtomicUsize::new(0));
let started = Arc::new(tokio::sync::Notify::new());
let (release_tx, release_rx) = std::sync::mpsc::channel();
let server_requests = Arc::clone(&requests);
let server_started = Arc::clone(&started);
let Some((base, server)) = start_mock_oidc_discovery_server_with_jwks(
|base| (base.to_string(), format!("{base}/jwks"), "/jwks".to_string()),
6,
"ES256",
false,
move |fetch| {
let request = server_requests.fetch_add(1, Ordering::SeqCst);
if request == 2 {
server_started.notify_one();
release_rx
.recv_timeout(StdDuration::from_secs(5))
.expect("release refresh response");
}
serde_json::json!({"keys": [if fetch == 0 { &initial_jwk } else { &rotated_jwk }]}).to_string()
},
) else {
return;
};
let mut config = build_mocked_oidc_provider_config("bounded-refresh", &base);
config.issuer = Some(base.clone());
let http = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).expect("allow test origin"));
let state = discover_provider(&config, &http).await.expect("initial discovery");
let sys = Arc::new(OidcSys {
configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]),
provider_runtime: ProviderRuntime::new(http, HashMap::from([(config.id.clone(), state)])),
state_store: OidcStateStore::new(),
pending_discovery: Default::default(),
});
let now = time::OffsetDateTime::now_utc().unix_timestamp();
let payload = serde_json::json!({"iss": base, "sub": "user", "aud": config.client_id, "iat": now, "exp": now + 300});
let sign = |kid: &str, claims: &serde_json::Value, key: &EncodingKey| {
let mut header = Header::new(Algorithm::ES256);
header.kid = Some(kid.to_string());
jsonwebtoken::encode(&header, claims, key).expect("sign test token")
};
for (claim, value) in [
("aud", serde_json::json!("wrong-audience")),
("exp", serde_json::json!(now - 300)),
] {
let mut invalid = payload.clone();
invalid[claim] = value;
assert!(
sys.verify_web_identity_token(&sign("initial", &invalid, &initial_key))
.await
.is_err()
);
}
assert_eq!(requests.load(Ordering::SeqCst), 2, "claim failures cannot refresh signing keys");
let invalid = sign("rotated", &payload, &wrong_key);
let attack = {
let sys = Arc::clone(&sys);
let invalid = invalid.clone();
tokio::spawn(async move { sys.verify_web_identity_token(&invalid).await })
};
tokio::time::timeout(StdDuration::from_secs(5), started.notified())
.await
.expect("refresh started");
attack.abort();
assert!(attack.await.expect_err("request canceled").is_cancelled());
let mut retries = Vec::new();
for _ in 0..16 {
let sys = Arc::clone(&sys);
let invalid = invalid.clone();
retries.push(tokio::spawn(async move { sys.verify_web_identity_token(&invalid).await }));
}
release_tx.send(()).expect("release provider");
for retry in retries {
assert!(
tokio::time::timeout(StdDuration::from_secs(5), retry)
.await
.expect("retry timeout")
.expect("retry join")
.is_err()
);
}
let rotated = sign("rotated", &payload, &rotated_key);
sys.verify_web_identity_token(&rotated)
.await
.expect("completed refresh must accept rotated keys after the original caller left");
assert_eq!(requests.load(Ordering::SeqCst), 4, "one discovery/JWKS pair for all rejected tokens");
sys.provider_runtime
.refresh_provider_state("independent-provider", &config)
.await
.expect("provider cooldowns must be independent");
assert_eq!(requests.load(Ordering::SeqCst), 6);
server.join().expect("mock server");
}
#[tokio::test]
async fn jwks_refresh_cools_down_failures_and_recovers_after_the_interval() {
let (_, initial_jwk) = oidc_es256_key_and_jwk("initial");
let (_, recovered_jwk) = oidc_es256_key_and_jwk("recovered");
let requests = Arc::new(AtomicUsize::new(0));
let server_requests = Arc::clone(&requests);
let Some((base, server)) = start_mock_oidc_discovery_server_with_jwks(
|base| (base.to_string(), format!("{base}/jwks"), "/jwks".to_string()),
6,
"ES256",
false,
move |fetch| {
server_requests.fetch_add(1, Ordering::SeqCst);
match fetch {
0 => serde_json::json!({"keys": [initial_jwk]}).to_string(),
1 => "invalid JWKS".to_string(),
_ => serde_json::json!({"keys": [recovered_jwk]}).to_string(),
}
},
) else {
return;
};
let mut config = build_mocked_oidc_provider_config("recovering-refresh", &base);
config.issuer = Some(base.clone());
let http = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).expect("allow test origin"));
let initial = discover_provider(&config, &http).await.expect("initial discovery");
let runtime = ProviderRuntime::new(http, HashMap::from([(config.id.clone(), initial.clone())]));
assert!(runtime.refresh_provider_state(&config.id, &config).await.is_err());
assert_eq!(requests.load(Ordering::SeqCst), 4);
for _ in 0..8 {
assert!(runtime.refresh_provider_state(&config.id, &config).await.is_err());
}
assert_eq!(requests.load(Ordering::SeqCst), 4, "completed failures also consume a bounded cooldown");
assert_eq!(
runtime.get_provider_state(&config.id).expect("cached keys").discovered_at,
initial.discovered_at
);
let refresh = runtime.refreshes.lock().expect("registry")[&config.id].clone();
refresh.lock().await.as_mut().expect("completed attempt").0 = Instant::now() - OIDC_JWKS_REFRESH_COOLDOWN;
runtime
.refresh_provider_state(&config.id, &config)
.await
.expect("retry after cooldown");
assert_eq!(requests.load(Ordering::SeqCst), 6);
assert!(runtime.get_provider_state(&config.id).expect("recovered keys").discovered_at > initial.discovered_at);
server.join().expect("mock server");
}
+8 -2
View File
@@ -33,8 +33,8 @@ use crate::{
};
use openidconnect::core::{CoreAuthenticationFlow, CoreClient, CoreIdToken};
use openidconnect::{
Audience, AuthType, AuthorizationCode, ClientId, ClientSecret, CsrfToken, LogoutRequest, Nonce, PkceCodeChallenge,
PkceCodeVerifier, PostLogoutRedirectUrl, RedirectUrl, RequestTokenError, Scope,
Audience, AuthType, AuthorizationCode, ClaimsVerificationError, ClientId, ClientSecret, CsrfToken, LogoutRequest, Nonce,
PkceCodeChallenge, PkceCodeVerifier, PostLogoutRedirectUrl, RedirectUrl, RequestTokenError, Scope,
};
use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
use rustfs_utils::logging::LogThrottle;
@@ -537,6 +537,9 @@ impl OidcSys {
.set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
let verified = verify_id_token_claims(id_token, &verifier, &Nonce::new(session.nonce.clone()));
if let Err(e) = verified {
if !matches!(e, ClaimsVerificationError::SignatureVerification(_)) {
return Err(format!("ID token verification failed: {e}"));
}
let verification_error = e.to_string();
warn!(
event = EVENT_OIDC_DIAGNOSTICS,
@@ -767,6 +770,9 @@ impl OidcSys {
.verifier(&config)
.set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
if let Err(e) = verify_id_token_claims(&id_token, &verifier, |_: Option<&Nonce>| Ok(())) {
if !matches!(e, ClaimsVerificationError::SignatureVerification(_)) {
return Err(format!("ID token verification failed: {e}"));
}
state = self
.provider_runtime
.refresh_provider_state(&provider_id, &config)
+14
View File
@@ -14,6 +14,20 @@
use std::sync::Arc;
pub(crate) fn is_unreadable_bucket_tags(error: &crate::error::Error) -> bool {
let crate::error::Error::Io(error) = error else {
return false;
};
let Some(storage_error) = error.get_ref().and_then(|inner| inner.downcast_ref::<IamStorageError>()) else {
return false;
};
rustfs_ecstore::api::bucket::metadata::unreadable_config_refusal(storage_error)
.is_some_and(|failure| failure.config_file == rustfs_ecstore::api::bucket::metadata::BUCKET_TAGGING_CONFIG)
}
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::bucket::metadata::UnreadableBucketConfig;
pub(crate) use rustfs_ecstore::api::config::RUSTFS_CONFIG_PREFIX as IAM_CONFIG_ROOT_PREFIX;
use rustfs_ecstore::api::config::com::{
delete_config as ecstore_delete_config, read_config_no_lock as ecstore_read_config_no_lock,
+146 -37
View File
@@ -1154,7 +1154,7 @@ impl Store for ObjectStore {
// node-counted lock quorum (rustfs#4304; startup contract rustfs#4056).
const LOAD_ALL_MODE: LoadMode = LoadMode::BootstrapNoLock;
let cache_snapshot = cache.snapshot();
let reload = cache.begin_reload().await;
let listed_config_items = self.list_all_iamconfig_items().await?;
let mut policy_docs_cache = CacheEntity::new(get_default_policies());
@@ -1201,26 +1201,20 @@ impl Store for ObjectStore {
}
// groups
let mut groups_cache = None;
let mut groups_cache = CacheEntity::default();
if let Some(item_name_list) = listed_config_items.get(GROUPS_LIST_KEY) {
let mut items_cache = CacheEntity::default();
for item in item_name_list.iter() {
let name = rustfs_utils::path::dir(item);
debug!(group = %name, "IAM group loaded");
if let Err(err) = self.load_group_with(&name, &mut items_cache, LOAD_ALL_MODE).await {
if let Err(err) = self.load_group_with(&name, &mut groups_cache, LOAD_ALL_MODE).await {
return Err(Error::other(format!("load group failed: {err}")));
};
}
groups_cache = Some(items_cache);
}
// user policies
let mut user_policies_cache = None;
let mut user_policies_cache = CacheEntity::default();
if let Some(item_name_list) = listed_config_items.get(POLICY_DB_USERS_LIST_KEY) {
let mut items_cache = CacheEntity::default();
// Load in fixed-size chunks so each mapping is fetched exactly once
// (backlog#806 — the split_off loop was O(n^2)).
for chunk in item_name_list.chunks(32) {
@@ -1235,32 +1229,26 @@ impl Store for ObjectStore {
let name = chunk[idx].trim_end_matches(".json").to_owned();
debug!(user = %name, "IAM user policy loaded");
items_cache.insert(name, p);
user_policies_cache.insert(name, p);
}
}
user_policies_cache = Some(items_cache);
}
// group policy
let mut group_policies_cache = None;
let mut group_policies_cache = CacheEntity::default();
if let Some(item_name_list) = listed_config_items.get(POLICY_DB_GROUPS_LIST_KEY) {
let mut items_cache = CacheEntity::default();
for item in item_name_list.iter() {
let name = item.trim_end_matches(".json");
debug!(group = %name, "IAM group policy loaded");
if let Err(err) = self
.load_mapped_policy_with(name, UserType::Reg, true, &mut items_cache, LOAD_ALL_MODE)
.load_mapped_policy_with(name, UserType::Reg, true, &mut group_policies_cache, LOAD_ALL_MODE)
.await
&& !is_err_no_such_policy(&err)
{
return Err(Error::other(format!("load group policy failed: {err}")));
};
}
group_policies_cache = Some(items_cache);
}
let mut sts_policies_cache = CacheEntity::default();
@@ -1330,25 +1318,48 @@ impl Store for ObjectStore {
}
}
#[cfg(test)]
if let Ok(barrier) = LOAD_ALL_COMMIT_BARRIER.try_with(std::sync::Arc::clone) {
barrier.0.notify_one();
barrier.1.notified().await;
}
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!("IAM full reload cache commit skipped due to concurrent cache changes");
}
reload.merge_partition(
crate::cache::CachePartition::PolicyDocs,
&mut policy_docs_cache,
&cache.state().policy_docs,
);
reload.merge_partition(crate::cache::CachePartition::Groups, &mut groups_cache, &cache.state().groups);
reload.merge_partition(
crate::cache::CachePartition::UserPolicies,
&mut user_policies_cache,
&cache.state().user_policies,
);
reload.merge_partition(
crate::cache::CachePartition::GroupPolicies,
&mut group_policies_cache,
&cache.state().group_policies,
);
reload.merge_partition(crate::cache::CachePartition::Users, &mut user_items_cache, &cache.state().users);
reload.merge_partition(
crate::cache::CachePartition::StsAccounts,
&mut sts_items_cache,
&cache.state().sts_accounts,
);
reload.merge_partition(
crate::cache::CachePartition::StsPolicies,
&mut sts_policies_cache,
&cache.state().sts_policies,
);
cache.replace_policy_docs(policy_docs_cache);
cache.replace_groups(groups_cache);
cache.replace_user_policies(user_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();
});
Ok(())
@@ -1363,6 +1374,11 @@ fn map_delete_user_identity_error(name: &str, err: Error) -> Error {
}
}
#[cfg(test)]
tokio::task_local! {
static LOAD_ALL_COMMIT_BARRIER: std::sync::Arc<(tokio::sync::Notify, tokio::sync::Notify)>;
}
#[cfg(test)]
mod tests {
use super::{DecryptSource, LoadMode, ObjectStore, map_delete_user_identity_error};
@@ -1372,6 +1388,99 @@ mod tests {
use serial_test::serial;
use temp_env::with_vars;
#[test]
#[serial]
fn test_load_all_clears_partitions_with_empty_prefix() {
use crate::cache::Cache;
use crate::store::{GroupInfo, MappedPolicy, Store, UserType};
std::thread::Builder::new()
.stack_size(8 * 1024 * 1024)
.spawn(|| {
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("build IAM reload test runtime");
runtime.block_on(async {
let _credentials = test_cred();
let temp = tempfile::tempdir().expect("create IAM reload store directory");
let env = Box::pin(
rustfs_test_utils::TestECStoreEnv::builder()
.base_dir(temp.path())
.init_bucket_metadata(false)
.build(),
)
.await;
let store = ObjectStore::new(env.ecstore);
store
.save_group_info("reload-group", GroupInfo::new(vec!["reload-user".to_string()]))
.await
.expect("seed group membership");
store
.save_mapped_policy("reload-user", UserType::Reg, false, MappedPolicy::new("readwrite"), None)
.await
.expect("seed user policy");
store
.save_mapped_policy("reload-group", UserType::Reg, true, MappedPolicy::new("readonly"), None)
.await
.expect("seed group policy");
let cache = Cache::default();
store.load_all(&cache).await.expect("load nonempty IAM partitions");
{
let snapshot = cache.snapshot();
assert!(snapshot.groups.contains_key("reload-group"));
assert_eq!(snapshot.user_policies["reload-user"].policies, "readwrite");
assert_eq!(snapshot.group_policies["reload-group"].policies, "readonly");
assert!(snapshot.user_group_memberships["reload-user"].contains("reload-group"));
}
// Mutate durable state without notifying this cache, as on
// a peer that missed the last mapping/group deletion.
store
.delete_mapped_policy("reload-user", UserType::Reg, false)
.await
.expect("remove last user mapping");
store
.delete_mapped_policy("reload-group", UserType::Reg, true)
.await
.expect("remove last group mapping");
store.delete_group_info("reload-group").await.expect("remove last group");
assert!(cache.snapshot().groups.contains_key("reload-group"));
// Simulate a missed peer revocation while STS publication
// and a separate user update overlap this full disk reload.
let stale = rustfs_policy::auth::UserIdentity {
version: 1,
..Default::default()
};
cache.add_or_update_user("revoked-user", &stale, time::OffsetDateTime::now_utc());
let barrier = std::sync::Arc::new((tokio::sync::Notify::new(), tokio::sync::Notify::new()));
let loading = super::LOAD_ALL_COMMIT_BARRIER.scope(std::sync::Arc::clone(&barrier), store.load_all(&cache));
let concurrent = async {
barrier.0.notified().await;
cache.add_or_update_sts_account("new-session", &stale, time::OffsetDateTime::now_utc());
cache.add_or_update_user("new-service", &stale, time::OffsetDateTime::now_utc());
barrier.1.notify_one();
};
let (loaded, ()) = tokio::join!(loading, concurrent);
loaded.expect("reconcile revoked identities despite STS traffic");
let snapshot = cache.snapshot();
assert!(!snapshot.users.contains_key("revoked-user"));
assert!(snapshot.users.contains_key("new-service"));
assert!(snapshot.sts_accounts.contains_key("new-session"));
assert!(snapshot.groups.is_empty());
assert!(snapshot.user_policies.is_empty());
assert!(snapshot.group_policies.is_empty());
assert!(snapshot.user_group_memberships.is_empty());
assert!(!snapshot.policy_docs.is_empty(), "built-in policies must remain available");
});
})
.expect("spawn IAM reload test thread")
.join()
.expect("IAM reload test should complete");
}
/// rustfs#4304 / startup contract rustfs#4056: the bulk snapshot load
/// (`load_all`) must never depend on the node-counted namespace lock
/// quorum, so its reads have to carry `no_lock = true` down to the
+1013 -18
View File
File diff suppressed because it is too large Load Diff
+170 -1
View File
@@ -293,6 +293,18 @@ pub async fn authenticate_request(
result
}
/// The policy client address comes only from verified middleware context.
/// Keep the original socket peer in request extensions for auditing.
pub(crate) fn policy_source_addr(extensions: &http::Extensions) -> Option<std::net::SocketAddr> {
let peer = extensions
.get::<Option<RemoteAddr>>()
.and_then(|peer| peer.map(|addr| addr.0));
extensions
.get::<rustfs_trusted_proxies::ClientInfo>()
.map(|client| std::net::SocketAddr::new(client.real_ip, peer.map_or(0, |addr| addr.port())))
.or(peer)
}
/// Full admin gate over an `S3Request`: extract the request credentials,
/// authenticate them ([`authenticate_request`]), then authorize the caller for
/// `actions` ([`validate_admin_request`], allowing on the first permitted
@@ -306,7 +318,7 @@ pub async fn authorize_admin_request(req: &S3Request<Body>, actions: Vec<Action>
let (cred, owner) = authenticate_request(&req.headers, &req.uri, input_cred).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = policy_source_addr(&req.extensions);
validate_admin_request(&req.headers, &cred, owner, false, actions, remote_addr).await?;
Ok(cred)
@@ -547,6 +559,163 @@ mod tests {
);
}
#[test]
#[serial_test::serial]
fn admin_and_sts_handlers_use_verified_client_ip_and_bucket_scope() {
crate::app::gating_test_env::run_large_stack_test("admin-client-conditions", || async {
use crate::admin::router::Operation;
use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface, publish_test_app_context};
use rustfs_iam::store::object::ObjectStore;
use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
use std::sync::Arc;
struct ReadyIam(Arc<rustfs_iam::sys::IamSys<ObjectStore>>);
impl IamInterface for ReadyIam {
fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<ObjectStore>> {
self.0.clone()
}
fn is_ready(&self) -> bool {
true
}
}
struct TestKms;
impl KmsInterface for TestKms {
fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
Arc::new(rustfs_kms::KmsServiceManager::new())
}
}
let previous = crate::runtime_sources::current_app_context();
let (_directory, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
ObjectStore::new(store.clone())
.save_iam_config(
serde_json::json!({"version": 1}),
format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
)
.await
.unwrap();
let iam = rustfs_iam::build_iam_sys(store.clone()).await.unwrap();
for (key, document) in [
(
"proxy-admin",
serde_json::json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":"admin:*"},
{"Effect":"Allow","Action":"sts:AssumeRole"},
{"Effect":"Deny","Action":"admin:*","Condition":{"NotIpAddress":{"aws:SourceIp":"10.0.0.0/8"}}},
{"Effect":"Deny","Action":"sts:AssumeRole","Condition":{"NotIpAddress":{"aws:SourceIp":"10.0.0.0/8"}}}
]}),
),
(
"scoped-usage",
serde_json::json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":"s3:ListBucket","Resource":"arn:aws:s3:::usage-allowed"},
{"Effect":"Deny","Action":"s3:ListBucket","NotResource":"arn:aws:s3:::usage-allowed"}
]}),
),
] {
iam.create_user(
key,
&AddOrUpdateUserReq {
secret_key: "test-user-secret".to_owned(),
policy: None,
status: AccountStatus::Enabled,
},
)
.await
.unwrap();
iam.set_policy(key, Policy::parse_config(&serde_json::to_vec(&document).unwrap()).unwrap())
.await
.unwrap();
iam.policy_db_set(key, UserType::Reg, false, key).await.unwrap();
}
let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(iam)), Arc::new(TestKms)));
assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
access_key: "proxy-test-root".to_owned(),
secret_key: "proxy-root-secret".to_owned(),
status: "on".to_owned(),
..Default::default()
}));
publish_test_app_context(context);
let request = |key: &str, client_ip: Option<&str>| {
let mut extensions = http::Extensions::new();
extensions.insert(Some(RemoteAddr("10.1.2.3:4567".parse().unwrap())));
if let Some(ip) = client_ip {
extensions.insert(rustfs_trusted_proxies::ClientInfo::direct(format!("{ip}:7890").parse().unwrap()));
}
let mut headers = HeaderMap::new();
headers.insert("x-forwarded-for", "10.1.2.3".parse().unwrap());
headers.insert("x-real-ip", "10.1.2.3".parse().unwrap());
S3Request {
input: Body::from(String::new()),
method: http::Method::GET,
uri: "/".parse().unwrap(),
headers,
extensions,
credentials: Some(s3s::auth::Credentials {
access_key: key.to_owned(),
secret_key: s3s::auth::SecretKey::from("test-user-secret".to_owned()),
}),
region: None,
service: None,
trailing_headers: None,
}
};
for (client, allowed) in [(Some("203.0.113.17"), false), (Some("10.8.7.6"), true), (None, true)] {
let req = request("proxy-admin", client);
assert_eq!(policy_source_addr(&req.extensions).unwrap().port(), 4567);
let result = authorize_admin_request(&req, vec![admin_action()]).await;
if allowed {
result.expect("verified private client is allowed");
} else {
assert_eq!(result.unwrap_err().code(), &s3s::S3ErrorCode::AccessDenied);
}
let mut req = request("proxy-admin", client);
req.method = http::Method::POST;
req.input = Body::from("Action=AssumeRole&Version=invalid-version".to_owned());
let error = crate::admin::handlers::sts::AssumeRoleHandle {}
.call(req, matchit::Params::new())
.await
.unwrap_err();
if allowed {
assert_eq!(error.code(), &s3s::S3ErrorCode::InvalidArgument);
assert_eq!(
error.message(),
Some("not support version"),
"only an authorized caller reaches version validation"
);
} else {
assert_eq!(error.code(), &s3s::S3ErrorCode::AccessDenied);
}
let response = crate::admin::handlers::is_admin::IsAdminHandler {}
.call(request("proxy-admin", client), matchit::Params::new())
.await
.unwrap();
let mut body = response.output.1;
let json: serde_json::Value = serde_json::from_slice(&body.store_all_limited(4096).await.unwrap()).unwrap();
assert_eq!(json["is_admin"], allowed);
}
let mut routes = matchit::Router::new();
routes.insert("/usage/{bucket}", ()).unwrap();
for (bucket, allowed) in [("usage-allowed", true), ("usage-secret", false)] {
let path = format!("/usage/{bucket}");
let matched = routes.at(&path).unwrap();
let mut req = request("scoped-usage", None);
req.uri = format!("{path}?unexpected=1").parse().unwrap();
let error = crate::admin::handlers::usage_prefix::BucketPrefixUsageHandler {}
.call(req, matched.params)
.await
.unwrap_err();
if allowed {
assert_eq!(error.code(), &s3s::S3ErrorCode::InvalidArgument);
assert_eq!(error.message(), Some("unknown query parameter: unexpected"));
} else {
assert_eq!(error.code(), &s3s::S3ErrorCode::AccessDenied);
}
}
if let Some(previous) = previous {
publish_test_app_context(previous);
}
});
}
/// Owner (root/admin) credential is authorized for an admin action.
#[tokio::test]
async fn admin_owner_credential_is_allowed() {
+34 -51
View File
@@ -42,7 +42,6 @@ use crate::admin::service::caller_identity::{CallerIdentity, oidc_profile_fields
use crate::admin::storage_api::s3::{self, Body, S3ErrorCode, S3Request, S3Response, S3Result};
use crate::admin::utils::read_compatible_admin_body;
use crate::auth::constant_time_eq;
use crate::server::RemoteAddr;
use http::StatusCode;
use hyper::Method;
use matchit::Params;
@@ -162,6 +161,13 @@ pub struct ChangeOwnPasswordHandler {}
#[async_trait::async_trait]
impl Operation for ChangeOwnPasswordHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let signing_secret = req
.credentials
.as_ref()
.ok_or_else(|| s3::error(S3ErrorCode::AccessDenied, "authentication required"))?
.secret_key
.expose()
.to_string();
let caller = CallerIdentity::resolve(&req).await?;
let audit = AccountAuditContext::from_request(&req);
@@ -203,7 +209,9 @@ impl Operation for ChangeOwnPasswordHandler {
return Err(s3::error(S3ErrorCode::InvalidRequest, "the calling identity no longer exists"));
};
if !constant_time_eq(&request.current_secret_key, &stored.credentials.secret_key) {
if !constant_time_eq(&request.current_secret_key, &stored.credentials.secret_key)
|| !constant_time_eq(&request.current_secret_key, &signing_secret)
{
warn!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_ACCOUNT,
@@ -256,31 +264,16 @@ impl Operation for ChangeOwnPasswordHandler {
let iam_store =
current_ready_iam_handle().map_err(|_| s3::error(S3ErrorCode::InternalError, "iam is not initialized"))?;
let (updated_at, status) = iam_store
.set_user_secret_key(&access_key, &new_secret_key)
let (updated_at, status, revoked) = iam_store
.change_own_secret_key(&access_key, &signing_secret, &new_secret_key)
.await
.map_err(iam_error_to_s3_error)?;
// Sessions minted under the old secret must not outlive it. A
// failure here is reported but does not undo the rotation: the new
// secret is already authoritative, and re-running the revocation is
// safe, whereas rolling the secret back would resurrect it.
let revoked = match iam_store.revoke_sts_sessions_for_parent(&access_key).await {
Ok(revoked) => revoked,
Err(err) => {
warn!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_ACCOUNT,
event = EVENT_ADMIN_ACCOUNT_STATE,
action = "change_own_password",
access_key = %MaskedAccessKey(&access_key),
result = "session_revocation_incomplete",
error = ?err,
"admin account state"
);
0
}
};
.map_err(|error| {
if matches!(error, rustfs_iam::error::Error::IAMActionNotAllowed) {
s3::error(S3ErrorCode::InvalidRequest, "the current secret key is incorrect")
} else {
iam_error_to_s3_error(error)
}
})?;
emit_audit(
&audit_for_task,
@@ -378,7 +371,7 @@ impl Operation for SetUserSecretKeyHandler {
caller.is_owner,
false,
vec![Action::AdminAction(AdminAction::CreateUserAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await
.inspect_err(|_| {
@@ -424,28 +417,11 @@ impl Operation for SetUserSecretKeyHandler {
));
}
let (updated_at, status) = iam_store
.set_user_secret_key(&target, &request.secret_key)
let (updated_at, status, revoked) = iam_store
.rotate_user_secret_key(&target, &request.secret_key)
.await
.map_err(iam_error_to_s3_error)?;
let revoked = match iam_store.revoke_sts_sessions_for_parent(&target).await {
Ok(revoked) => revoked,
Err(err) => {
warn!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_ACCOUNT,
event = EVENT_ADMIN_ACCOUNT_STATE,
action = "set_user_secret_key",
access_key = %MaskedAccessKey(&target),
result = "session_revocation_incomplete",
error = ?err,
"admin account state"
);
0
}
};
emit_audit(
&audit_for_task,
AccountAuditRecord::success(AccountAuditOperation::ResetUserPassword, &target, identity_type)
@@ -648,16 +624,23 @@ mod tests {
#[test]
fn both_rotation_handlers_broadcast_after_revoking_sessions() {
let src = include_str!("account.rs");
for marker in [
"impl Operation for ChangeOwnPasswordHandler",
"impl Operation for SetUserSecretKeyHandler",
// Session revocation now runs inside the IAM rotation call, under the
// per-key mutation lock, so the handler must reach that call before it
// broadcasts the new secret to peer sites.
for (marker, rotation) in [
("impl Operation for ChangeOwnPasswordHandler", "change_own_secret_key("),
("impl Operation for SetUserSecretKeyHandler", "rotate_user_secret_key("),
] {
let start = src.find(marker).expect("handler should exist");
let block = &src[start..];
let block = &block[..block.find("\n}\n").expect("handler block end")];
let revoke = block
.find("revoke_sts_sessions_for_parent")
.expect("handler must revoke sessions");
.find(rotation)
.expect("handler must rotate through the IAM call that revokes sessions");
assert!(
!block.contains("revoke_sts_sessions_for_parent"),
"{marker}: revocation belongs to the locked IAM rotation, not a separate handler step"
);
let broadcast = block
.find("broadcast_secret_key_rotation(")
.expect("handler must broadcast the rotation to peer sites");
+2 -2
View File
@@ -22,7 +22,7 @@ use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOption
use crate::admin::storage_api::data_usage::{apply_bucket_usage_memory_overlay, load_admin_data_usage_from_backend_cached};
use crate::admin::storage_api::metadata_sys;
use crate::auth::get_condition_values;
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use http::{HeaderMap, HeaderValue};
use hyper::{Method, StatusCode};
use matchit::Params;
@@ -146,7 +146,7 @@ impl Operation for AccountInfoHandler {
let claims = cred.claims.as_ref().unwrap_or(&default_claims);
let cred_clone = cred.clone();
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
let conditions = get_condition_values(&req.headers, &cred_clone, None, None, remote_addr);
let cred_clone = Arc::new(cred_clone);
let conditions = Arc::new(conditions);
+1 -2
View File
@@ -28,7 +28,6 @@ use crate::admin::storage_api::bucket::metadata::BUCKET_DURABILITY_CONFIG;
use crate::admin::storage_api::bucket::metadata_sys;
use crate::auth::{check_key_valid, get_session_token};
use crate::server::ADMIN_PREFIX;
use crate::server::RemoteAddr;
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_policy::policy::action::{Action, AdminAction};
@@ -127,7 +126,7 @@ async fn authenticate_admin(req: &S3Request<Body>) -> S3Result<()> {
let (cred, owner) = check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
validate_admin_request(
&req.headers,
&cred,
+154 -4
View File
@@ -434,7 +434,7 @@ impl Operation for SetGroupStatus {
pub struct UpdateGroupMembers {}
#[async_trait::async_trait]
impl Operation for UpdateGroupMembers {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!(
event = EVENT_ADMIN_GROUP_STATE,
component = LOG_COMPONENT_ADMIN,
@@ -448,9 +448,16 @@ impl Operation for UpdateGroupMembers {
return Err(s3_error!(InvalidRequest, "authentication required"));
}
authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::AddUserToGroupAdminAction)]).await?;
authorize_admin_request(
&req,
vec![
Action::AdminAction(AdminAction::AddUserToGroupAdminAction),
Action::AdminAction(AdminAction::RemoveUserFromGroupAdminAction),
],
)
.await?;
let mut input = req.input;
let mut input = std::mem::take(&mut req.input);
let body = match input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await {
Ok(b) => b,
Err(e) => {
@@ -470,6 +477,13 @@ impl Operation for UpdateGroupMembers {
let args: GroupAddRemove = serde_json::from_slice(&body)
.map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid JSON: {e}")))?;
let action = if args.is_remove {
AdminAction::RemoveUserFromGroupAdminAction
} else {
AdminAction::AddUserToGroupAdminAction
};
authorize_admin_request(&req, vec![Action::AdminAction(action)]).await?;
warn!(
event = EVENT_ADMIN_GROUP_STATE,
component = LOG_COMPONENT_ADMIN,
@@ -719,6 +733,143 @@ mod tests {
&block[..end]
}
#[tokio::test]
#[serial_test::serial]
async fn group_members_handler_enforces_add_and_remove_permissions() {
use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface, publish_test_app_context};
use rustfs_iam::store::{Store as _, UserType, object::ObjectStore};
use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
use rustfs_policy::policy::Policy;
use std::sync::Arc;
struct ReadyIam(Arc<rustfs_iam::sys::IamSys<ObjectStore>>);
impl IamInterface for ReadyIam {
fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<ObjectStore>> {
Arc::clone(&self.0)
}
fn is_ready(&self) -> bool {
true
}
}
struct TestKms;
impl KmsInterface for TestKms {
fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
Arc::new(rustfs_kms::KmsServiceManager::new())
}
}
let (_directory, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
ObjectStore::new(store.clone())
.save_iam_config(
serde_json::json!({"version": 1}),
format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
)
.await
.expect("seed IAM format");
let iam = rustfs_iam::build_iam_sys(store.clone()).await.expect("initialize IAM");
for key in ["group-adder", "group-remover", "group-member"] {
iam.create_user(
key,
&AddOrUpdateUserReq {
secret_key: "group-user-secret".to_owned(),
policy: None,
status: AccountStatus::Enabled,
},
)
.await
.expect("create group test user");
}
for (key, allow, deny) in [
("group-adder", "admin:AddUserToGroup", "admin:RemoveUserFromGroup"),
("group-remover", "admin:RemoveUserFromGroup", "admin:AddUserToGroup"),
] {
let document = serde_json::json!({"Version": "2012-10-17", "Statement": [
{"Effect": "Allow", "Action": [allow]}, {"Effect": "Deny", "Action": [deny]}
]});
iam.set_policy(
key,
Policy::parse_config(&serde_json::to_vec(&document).expect("policy JSON")).expect("parse policy"),
)
.await
.expect("store delegated policy");
iam.policy_db_set(key, UserType::Reg, false, key)
.await
.expect("attach delegated policy");
}
iam.add_users_to_group("guardrail", vec!["group-member".to_owned()])
.await
.expect("seed protected group");
let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(Arc::clone(&iam))), Arc::new(TestKms)));
assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
access_key: "group-test-root".to_owned(),
secret_key: "group-root-secret".to_owned(),
status: "on".to_owned(),
..Default::default()
}));
publish_test_app_context(context);
for (key, remove, allowed) in [
("group-adder", true, false),
("group-remover", false, false),
("group-remover", true, true),
("group-adder", false, true),
] {
let before = iam
.get_group_description("guardrail")
.await
.expect("read original membership")
.members;
let mut req = credential_less_request(Method::PUT, "/rustfs/admin/v3/update-group-members");
req.credentials = Some(s3s::auth::Credentials {
access_key: key.to_owned(),
secret_key: s3s::auth::SecretKey::from("group-user-secret".to_owned()),
});
req.input = Body::from(
serde_json::to_vec(&serde_json::json!({
"group": "guardrail", "members": ["group-member"], "groupStatus": "enabled", "isRemove": remove
}))
.expect("group update JSON"),
);
let result = UpdateGroupMembers {}.call(req, Params::new()).await;
let after = iam
.get_group_description("guardrail")
.await
.expect("read resulting membership")
.members;
if allowed {
result.expect("matching delegated action permits the mutation");
assert_eq!(after.contains(&"group-member".to_owned()), !remove);
} else {
let error = result.expect_err("the opposite action must not authorize the mutation");
assert_eq!(error.code(), &S3ErrorCode::AccessDenied);
assert_eq!(after, before, "a rejected mutation must preserve the group");
}
}
}
#[test]
fn update_group_members_authorizes_the_decoded_mutation() {
let source = include_str!("group.rs")
.split("\n#[cfg(test)]\n")
.next()
.expect("production source");
let handler = source_block(source, "impl Operation for UpdateGroupMembers");
assert_eq!(handler.matches("authorize_admin_request(").count(), 2);
let decoded = handler
.find("let args: GroupAddRemove")
.expect("decode request before choosing action");
let action = handler
.find("let action = if args.is_remove")
.expect("body controls requested action");
let gate = handler.rfind("authorize_admin_request(").expect("mutation authorization");
assert!(decoded < action && action < gate);
assert!(handler[action..gate].contains("AdminAction::RemoveUserFromGroupAdminAction"));
assert!(handler[action..gate].contains("AdminAction::AddUserToGroupAdminAction"));
for mutation in [".remove_users_from_group(", ".add_users_to_group("] {
assert!(gate < handler.find(mutation).expect("group mutation"));
}
}
#[test]
fn group_handlers_use_the_shared_admin_gate_with_their_actions() {
let production = include_str!("group.rs")
@@ -731,7 +882,6 @@ mod tests {
("GetGroup", "GetGroupAdminAction"),
("DeleteGroup", "RemoveUserFromGroupAdminAction"),
("SetGroupStatus", "EnableGroupAdminAction"),
("UpdateGroupMembers", "AddUserToGroupAdminAction"),
] {
let block = source_block(production, &format!("impl Operation for {handler}"));
assert_eq!(
+1 -1
View File
@@ -1474,7 +1474,7 @@ async fn authenticate_heal_admin_request(req: &S3Request<Body>) -> S3Result<Stri
owner,
false,
vec![Action::AdminAction(AdminAction::HealAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await?;
Ok(cred.access_key)
+2 -2
View File
@@ -48,7 +48,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::{current_app_context, current_ready_iam_handle, current_server_config_for_context};
use crate::admin::utils::is_compat_admin_request;
use crate::auth::{check_key_valid, get_session_token};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use http::StatusCode;
use hyper::Method;
use matchit::Params;
@@ -981,7 +981,7 @@ async fn authorize_action(req: &S3Request<Body>, cred: &StoredCredentials, owner
owner,
false,
vec![Action::AdminAction(action)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await
}
+2 -1
View File
@@ -66,7 +66,8 @@ impl Operation for IsAdminHandler {
let empty_claims = HashMap::new();
let iam_store = crate::admin::runtime_sources::current_ready_iam_handle()
.map_err(|_| s3_error!(InternalError, "iam not init"))?;
let conditions = get_condition_values(&req.headers, &cred, None, None, None);
let conditions =
get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
iam_store
.is_allowed(&Args {
account: &cred.access_key,
+2 -2
View File
@@ -46,7 +46,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::{current_deployment_id, current_kms_runtime_service_manager};
use crate::admin::utils::json_response;
use crate::auth::{check_key_valid, get_session_token};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use base64_simd::STANDARD as BASE64;
use hyper::{Method, StatusCode};
use matchit::Params;
@@ -923,7 +923,7 @@ async fn authorize(
owner,
false,
actions,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
operation,
+7 -7
View File
@@ -28,7 +28,7 @@ use crate::admin::storage_api::error::StorageError;
use crate::admin::storage_api::runtime::ECStore;
use crate::admin::storage_api::s3::{S3ErrorCode, error as admin_s3_error};
use crate::auth::{check_key_valid, get_session_token};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -859,7 +859,7 @@ impl Operation for ConfigureKmsHandler {
owner,
false,
kms_configure_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAdminOperation::Configure,
@@ -1012,7 +1012,7 @@ impl Operation for StartKmsHandler {
owner,
false,
kms_service_control_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAdminOperation::Start,
@@ -1169,7 +1169,7 @@ impl Operation for StopKmsHandler {
owner,
false,
kms_service_control_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAdminOperation::Stop,
@@ -1271,7 +1271,7 @@ impl Operation for ReloadKmsHandler {
owner,
false,
kms_service_control_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAdminOperation::Reload,
@@ -1365,7 +1365,7 @@ impl Operation for GetKmsStatusHandler {
owner,
false,
kms_service_control_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await?;
@@ -1459,7 +1459,7 @@ impl Operation for ReconfigureKmsHandler {
owner,
false,
kms_configure_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAdminOperation::Reconfigure,
@@ -21,7 +21,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::current_kms_runtime_service_manager;
use crate::admin::utils::{extract_query_params, json_response};
use crate::auth::{check_key_valid, get_session_token};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -270,7 +270,7 @@ async fn handle_lifecycle_request(
owner,
false,
actions,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
scoped.as_deref().unwrap_or_default(),
)
.await,
@@ -25,7 +25,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::current_kms_runtime_service_manager;
use crate::admin::utils::json_response;
use crate::auth::{check_key_valid, get_session_token};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -327,7 +327,7 @@ async fn handle_metadata_request(
owner,
false,
actions,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
scoped.as_deref().unwrap_or_default(),
)
.await,
+10 -10
View File
@@ -22,7 +22,7 @@ use crate::admin::storage_api::s3;
use crate::admin::utils::extract_query_params;
use crate::auth::{check_key_valid, get_session_token};
use crate::kms_deletion_gate::current_key_impact;
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use hyper::{HeaderMap, Method, StatusCode};
use matchit::Params;
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -325,7 +325,7 @@ impl Operation for CreateKeyHandler {
owner,
false,
kms_create_key_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAuditOperation::CreateKey,
@@ -418,7 +418,7 @@ impl Operation for DescribeKeyHandler {
owner,
false,
kms_describe_key_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
requested_key_id.as_deref().unwrap_or_default(),
)
.await,
@@ -1488,7 +1488,7 @@ impl Operation for ListKeysHandler {
owner,
false,
kms_list_keys_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAuditOperation::ListKeys,
@@ -1583,7 +1583,7 @@ impl Operation for GenerateDataKeyHandler {
owner,
false,
kms_generate_data_key_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
parsed.as_ref().map(|request| request.key_id.as_str()).unwrap_or_default(),
)
.await,
@@ -1664,7 +1664,7 @@ impl Operation for CreateKmsKeyHandler {
owner,
false,
kms_create_key_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAuditOperation::CreateKey,
@@ -1943,7 +1943,7 @@ impl Operation for DeleteKmsKeyHandler {
owner,
false,
kms_delete_key_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
scoped.as_deref().unwrap_or_default(),
)
.await,
@@ -2111,7 +2111,7 @@ impl Operation for CancelKmsKeyDeletionHandler {
owner,
false,
kms_delete_key_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
scoped.as_deref().unwrap_or_default(),
)
.await,
@@ -2263,7 +2263,7 @@ impl Operation for ListKmsKeysHandler {
owner,
false,
kms_list_keys_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await,
KmsAuditOperation::ListKeys,
@@ -2459,7 +2459,7 @@ impl Operation for DescribeKmsKeyHandler {
owner,
false,
kms_describe_key_actions(),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
params.get("key_id").unwrap_or_default(),
)
.await,
+2 -3
View File
@@ -51,7 +51,6 @@ use crate::admin::storage_api::runtime::ECStore;
use crate::admin::storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
use crate::admin::utils::read_compatible_admin_body;
use crate::auth::constant_time_eq;
use crate::server::RemoteAddr;
use http::StatusCode;
use hyper::Method;
use matchit::Params;
@@ -484,7 +483,7 @@ impl Operation for UserMfaStatusHandler {
caller.is_owner,
false,
vec![Action::AdminAction(AdminAction::GetUserAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await?;
@@ -521,7 +520,7 @@ impl Operation for UserMfaResetHandler {
caller.is_owner,
false,
vec![Action::AdminAction(AdminAction::EnableUserAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await
.inspect_err(|_| {
@@ -11,7 +11,7 @@ use crate::admin::{
};
use crate::{
auth::{check_key_valid_with_context, get_session_token},
server::{ADMIN_PREFIX, RemoteAddr},
server::ADMIN_PREFIX,
};
use http::{HeaderMap, HeaderValue, Method, StatusCode};
use matchit::Params;
@@ -97,7 +97,7 @@ pub(super) async fn owner(
&credentials,
false,
Action::AdminAction(action),
req.extensions.get::<Option<RemoteAddr>>().and_then(|a| a.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
AdminResourceScope::bucket_object("", ""),
)
.await?;
+6 -7
View File
@@ -30,7 +30,6 @@ use crate::admin::storage_api::contract::bucket::{BucketOperations as _, BucketO
use crate::auth::{check_key_valid, get_session_token};
use crate::error::ApiError;
use crate::server::ADMIN_PREFIX;
use crate::server::RemoteAddr;
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
@@ -397,7 +396,7 @@ impl Operation for SetBucketQuotaHandler {
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
validate_admin_request(
&req.headers,
&cred,
@@ -538,7 +537,7 @@ impl Operation for GetBucketQuotaHandler {
if bucket.is_empty() {
return Err(s3_error!(InvalidRequest, "bucket name is required"));
}
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
validate_admin_request_with_bucket(
&req.headers,
&cred,
@@ -607,7 +606,7 @@ impl Operation for ClearBucketQuotaHandler {
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
validate_admin_request(
&req.headers,
&cred,
@@ -726,7 +725,7 @@ impl Operation for GetBucketQuotaStatsHandler {
return Err(s3_error!(InvalidRequest, "bucket name is required"));
}
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
validate_admin_request_with_bucket(
&req.headers,
&cred,
@@ -795,7 +794,7 @@ impl Operation for CheckBucketQuotaHandler {
return Err(s3_error!(InvalidRequest, "bucket name is required"));
}
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
validate_admin_request_with_bucket(
&req.headers,
&cred,
@@ -890,7 +889,7 @@ impl Operation for MigrateBucketQuotaProtocolHandler {
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
validate_admin_request(
&req.headers,
&cred,
+228 -56
View File
@@ -18,7 +18,7 @@ use crate::admin::handlers::site_replication::{encode_service_account_replicatio
use crate::admin::runtime_sources::current_action_credentials;
use crate::admin::utils::{encode_compatible_admin_payload, has_space_be, is_compat_admin_request, read_compatible_admin_body};
use crate::auth::{constant_time_eq, get_condition_values, get_session_token};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::server::ADMIN_PREFIX;
use crate::{
admin::router::{AdminOperation, Operation, S3Router},
auth::check_key_valid,
@@ -84,6 +84,25 @@ fn parse_new_service_account_policy(policy: Option<&serde_json::Value>) -> S3Res
Ok((!policy.id.is_empty() || !policy.version.is_empty() || !policy.statements.is_empty()).then_some(policy))
}
/// Service accounts cannot delegate beyond their own lifetime. STS sessions
/// retain the existing ability to create independently expiring service accounts.
fn bounded_service_account_expiration(caller: &StoredCredentials, requested: Option<OffsetDateTime>) -> Option<OffsetDateTime> {
match caller.expiration.filter(|_| caller.is_service_account()) {
Some(limit) => Some(requested.map_or(limit, |expiration| expiration.min(limit))),
None => requested,
}
}
fn set_service_account_duration_condition(
conditions: &mut HashMap<String, Vec<String>>,
expiration: Option<OffsetDateTime>,
now: OffsetDateTime,
) {
// An unlimited credential must not bypass a finite NumericGreaterThan Deny.
let seconds = expiration.map_or(i64::MAX, |expiration| (expiration - now).whole_seconds().max(0));
conditions.insert("DurationSeconds".to_string(), vec![seconds.to_string()]);
}
fn compat_time_sentinel() -> OffsetDateTime {
OffsetDateTime::UNIX_EPOCH
}
@@ -322,9 +341,13 @@ impl Operation for AddServiceAccount {
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &req_cred.access_key).await?;
if !constant_time_eq(&cred.secret_key, req_cred.secret_key.expose()) {
return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "access denied"));
}
let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
let create_req: AddServiceAccountReq =
let mut create_req: AddServiceAccountReq =
serde_json::from_slice(&body[..]).map_err(|e| s3_error!(InvalidRequest, "unmarshal body failed, e: {:?}", e))?;
// create_req.expiration = create_req.expiration.and_then(|expire| expire.replace_millisecond(0).ok());
@@ -370,6 +393,11 @@ impl Operation for AddServiceAccount {
return Err(s3_error!(InvalidRequest, "iam not init"));
};
create_req.expiration = bounded_service_account_expiration(&cred, create_req.expiration);
let mut conditions =
get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
set_service_account_duration_condition(&mut conditions, create_req.expiration, OffsetDateTime::now_utc());
// This family deliberately calls `is_allowed` directly instead of going
// through `validate_admin_request`, and must keep doing so
// (backlog#1886). The shared helper returns as soon as *any* candidate
@@ -383,13 +411,7 @@ impl Operation for AddServiceAccount {
groups: &cred.groups,
action: Action::AdminAction(AdminAction::CreateServiceAccountAdminAction),
bucket: "",
conditions: &get_condition_values(
&req.headers,
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
),
conditions: &conditions,
is_owner: owner,
object: "",
claims: cred.claims_or_empty(),
@@ -444,17 +466,9 @@ impl Operation for AddServiceAccount {
let replication_description = opts.description.clone().unwrap_or_default();
let replication_expiration = opts.expiration;
let create_result = if is_svc_acc {
iam_store
.new_service_account_from_caller(&target_user, target_groups, opts, &cred)
.await
} else {
let replication_claims = opts.claims.clone().unwrap_or_default();
iam_store
.new_service_account(&target_user, target_groups, opts)
.await
.map(|(credentials, updated_at)| (credentials, updated_at, replication_claims))
};
let create_result = iam_store
.new_service_account_from_caller(&target_user, target_groups, opts, &cred)
.await;
let (new_cred, updated_at, replication_claims) = create_result.map_err(|e| {
debug!(
component = LOG_COMPONENT_ADMIN,
@@ -466,7 +480,8 @@ impl Operation for AddServiceAccount {
"admin service account state"
);
match e {
rustfs_iam::error::Error::InvalidAccessKeyLength
rustfs_iam::error::Error::InvalidArgument
| rustfs_iam::error::Error::InvalidAccessKeyLength
| rustfs_iam::error::Error::InvalidSecretKeyLength
| rustfs_iam::error::Error::AccessKeyAlreadyExists => iam_error_to_s3_error(e),
rustfs_iam::error::Error::IAMActionNotAllowed => s3_error!(AccessDenied, "access denied"),
@@ -599,6 +614,9 @@ impl Operation for UpdateServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
return Err(s3_error!(InvalidRequest, "iam not init"));
};
@@ -608,17 +626,29 @@ impl Operation for UpdateServiceAccount {
// s3_error!(InternalError, "get service account failed")
// })?;
let body =
read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), input_cred.secret_key.expose())
.await?;
let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
let update_req: UpdateServiceAccountReq =
let mut update_req: UpdateServiceAccountReq =
serde_json::from_slice(&body[..]).map_err(|e| s3_error!(InvalidRequest, "unmarshal body failed, e: {:?}", e))?;
update_req.validate().map_err(|e| S3Error::with_message(InvalidRequest, e))?;
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
// Keep lookup errors behind authorization even though lifetime conditions
// need the existing expiration when the request omits it.
let existing_account = iam_store.get_service_account(&access_key).await;
if update_req.new_expiration.is_some() {
update_req.new_expiration = bounded_service_account_expiration(&cred, update_req.new_expiration);
}
let mut conditions =
get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
set_service_account_duration_condition(
&mut conditions,
update_req
.new_expiration
.or_else(|| existing_account.as_ref().ok().and_then(|(account, _)| account.expiration)),
OffsetDateTime::now_utc(),
);
if !iam_store
.is_allowed(&Args {
@@ -626,13 +656,7 @@ impl Operation for UpdateServiceAccount {
groups: &cred.groups,
action: Action::AdminAction(AdminAction::UpdateServiceAccountAdminAction),
bucket: "",
conditions: &get_condition_values(
&req.headers,
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
),
conditions: &conditions,
is_owner: owner,
object: "",
claims: cred.claims_or_empty(),
@@ -643,10 +667,7 @@ impl Operation for UpdateServiceAccount {
return Err(s3_error!(AccessDenied, "access denied"));
}
let (svc_account, _) = iam_store
.get_service_account(&access_key)
.await
.map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
let (svc_account, _) = existing_account.map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
// The admin action permits updates within a caller's scope; only an
// owner may cross service-account parent boundaries.
@@ -723,11 +744,6 @@ impl Operation for InfoServiceAccount {
return Err(s3_error!(InvalidRequest, "iam not init"));
};
let (svc_account, session_policy) = iam_store
.get_service_account(&access_key)
.await
.map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
let Some(input_cred) = req.credentials else {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
@@ -735,6 +751,11 @@ impl Operation for InfoServiceAccount {
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let (svc_account, session_policy) = iam_store
.get_service_account(&access_key)
.await
.map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
@@ -746,7 +767,7 @@ impl Operation for InfoServiceAccount {
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
),
is_owner: owner,
object: "",
@@ -811,7 +832,7 @@ impl Operation for TemporaryAccountInfo {
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
),
is_owner: owner,
object: "",
@@ -872,13 +893,8 @@ impl Operation for InfoAccessKey {
let target_cred = iam_store.get_user(&access_key).await.map(|identity| identity.credentials);
let conditions = get_condition_values(
&req.headers,
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
);
let conditions =
get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
let claims = cred.claims_or_empty();
if !iam_store
@@ -1007,7 +1023,7 @@ impl Operation for ListServiceAccount {
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
),
is_owner: owner,
object: "",
@@ -1151,7 +1167,7 @@ impl Operation for ListAccessKeysBulk {
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
),
is_owner: owner,
object: "",
@@ -1174,7 +1190,7 @@ impl Operation for ListAccessKeysBulk {
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
),
is_owner: owner,
object: "",
@@ -1355,7 +1371,7 @@ impl Operation for DeleteServiceAccount {
&cred,
None,
None,
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
),
is_owner: owner,
object: "",
@@ -1444,12 +1460,168 @@ mod tests {
use serde_json::json;
use serde_urlencoded::from_bytes;
#[test]
#[serial_test::serial]
fn service_account_handlers_authenticate_before_decryption_or_lookup() {
crate::app::gating_test_env::run_large_stack_test("service-account-auth-first", || async {
use crate::admin::runtime_sources::IamInterface;
use crate::admin::runtime_sources::{AppContext, publish_test_app_context};
use rustfs_iam::store::object::ObjectStore;
use std::sync::Arc;
struct ReadyIam(Arc<rustfs_iam::sys::IamSys<ObjectStore>>);
impl IamInterface for ReadyIam {
fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<ObjectStore>> {
self.0.clone()
}
fn is_ready(&self) -> bool {
true
}
}
let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
let (_directory, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
ObjectStore::new(store.clone())
.save_iam_config(
serde_json::json!({"version": 1}),
format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
)
.await
.expect("initialize IAM storage");
let iam = rustfs_iam::build_iam_sys(store.clone()).await.expect("initialized IAM");
let disabled_key = format!("disabled-{}", uuid::Uuid::new_v4());
iam.create_user(
&disabled_key,
&rustfs_madmin::AddOrUpdateUserReq {
secret_key: "disabled-user-secret".to_owned(),
policy: None,
status: rustfs_madmin::AccountStatus::Disabled,
},
)
.await
.expect("create disabled IAM caller");
let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(iam)), ambient.kms()));
assert!(context.publish_action_credentials(StoredCredentials {
access_key: "auth-first-root".to_owned(),
secret_key: "auth-first-root-secret".to_owned(),
status: "on".to_owned(),
..Default::default()
}));
publish_test_app_context(context);
for caller in ["keystone:forged", disabled_key.as_str()] {
for operation in ["update-service-account", "info-service-account"] {
let req = S3Request {
input: Body::from(vec![0u8; 64]),
method: if operation == "update-service-account" {
http::Method::POST
} else {
http::Method::GET
},
uri: format!("/minio/admin/v3/{operation}?accessKey=missing-target")
.parse()
.unwrap(),
headers: HeaderMap::new(),
extensions: http::Extensions::new(),
credentials: Some(s3s::auth::Credentials {
access_key: caller.to_owned(),
secret_key: s3s::auth::SecretKey::from("disabled-user-secret"),
}),
region: None,
service: None,
trailing_headers: None,
};
let result = if operation == "update-service-account" {
UpdateServiceAccount {}.call(req, Params::new()).await
} else {
InfoServiceAccount {}.call(req, Params::new()).await
};
assert_eq!(
result
.expect_err("invalid identity precedes KDF and missing target lookup")
.code(),
&s3s::S3ErrorCode::InvalidAccessKeyId,
"{caller}: {operation}"
);
}
}
publish_test_app_context(ambient);
});
}
#[test]
fn access_key_query_supports_external_alias() {
let query: AccessKeyQuery = from_bytes(b"access-key=test-access-key").expect("parse query");
assert_eq!(query.access_key, "test-access-key");
}
#[test]
fn service_account_cannot_delegate_a_longer_lifetime() {
let deadline = OffsetDateTime::UNIX_EPOCH + time::Duration::days(10);
let earlier = deadline - time::Duration::hours(1);
let later = deadline + time::Duration::days(365);
let mut caller = StoredCredentials {
access_key: "expiring-service".to_string(),
parent_user: "parent".to_string(),
expiration: Some(deadline),
claims: Some(HashMap::from([("sa-policy".to_string(), json!("inherited-policy"))])),
..Default::default()
};
assert_eq!(bounded_service_account_expiration(&caller, None), Some(deadline));
assert_eq!(bounded_service_account_expiration(&caller, Some(later)), Some(deadline));
assert_eq!(bounded_service_account_expiration(&caller, Some(earlier)), Some(earlier));
caller.expiration = None;
assert_eq!(bounded_service_account_expiration(&caller, None), None);
caller.expiration = Some(deadline);
caller.claims = None;
caller.session_token = "signed-sts".to_string();
assert_eq!(
bounded_service_account_expiration(&caller, None),
None,
"STS retains independent service lifetime"
);
assert_eq!(bounded_service_account_expiration(&caller, Some(later)), Some(later));
}
#[tokio::test]
async fn service_account_duration_condition_enforces_finite_and_unlimited_lifetimes() {
let policy = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"],"Resource":"*"},
{"Effect":"Deny","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"],"Resource":"*",
"Condition":{"NumericGreaterThan":{"svc:DurationSeconds":"3600"}}}
]}"#,
)
.expect("service lifetime policy");
let now = OffsetDateTime::UNIX_EPOCH;
for (expiration, allowed) in [
(Some(now + time::Duration::seconds(3600)), true),
(Some(now + time::Duration::seconds(3601)), false),
(None, false),
] {
let mut conditions = HashMap::from([("DurationSeconds".to_string(), vec!["1".to_string()])]);
set_service_account_duration_condition(&mut conditions, expiration, now);
for action in [
AdminAction::CreateServiceAccountAdminAction,
AdminAction::UpdateServiceAccountAdminAction,
] {
assert_eq!(
policy
.is_allowed(&Args {
account: "user",
groups: &None,
action: Action::AdminAction(action),
conditions: &conditions,
is_owner: false,
claims: &HashMap::new(),
deny_only: false,
bucket: "",
object: "",
})
.await,
allowed
);
}
}
}
#[test]
fn non_admin_delete_is_fail_closed_on_unloadable_target() {
// Regression (backlog#806): a non-admin caller may delete ONLY a service
+53 -8
View File
@@ -27,7 +27,6 @@ use crate::{
},
auth::{check_key_valid, get_session_token},
server::ADMIN_PREFIX,
server::RemoteAddr,
};
use http::StatusCode;
use http::header::HeaderValue;
@@ -54,6 +53,7 @@ use s3s::{
use serde::Deserialize;
use serde_json::Value;
use serde_urlencoded::from_bytes;
use std::collections::HashMap;
use time::{Duration, OffsetDateTime};
use tracing::{debug, error, warn};
@@ -82,6 +82,12 @@ fn clamp_assume_role_duration(duration_seconds: usize) -> usize {
}
}
fn set_assume_role_duration_condition(conditions: &mut HashMap<String, Vec<String>>, requested: usize) -> usize {
let duration = clamp_assume_role_duration(requested);
conditions.insert("DurationSeconds".to_string(), vec![duration.to_string()]);
duration
}
/// Record that AssumeRole assembled its session claims without echoing any claim values.
///
/// Claims carry caller-supplied identity material (parent user, session policy, arbitrary
@@ -186,7 +192,7 @@ impl Operation for AssumeRoleHandle {
match body.action.as_str() {
ASSUME_ROLE_ACTION => {
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
handle_assume_role(req.credentials, req.uri, req.headers, remote_addr, body, store, &audit).await
}
ASSUME_ROLE_WITH_WEB_IDENTITY_ACTION => handle_assume_role_with_web_identity(body).await,
@@ -216,6 +222,12 @@ async fn handle_assume_role(
let (cred, owner) = check_key_valid(get_session_token(&uri, &headers).unwrap_or_default(), &user.access_key).await?;
// The signature was verified before the request body was consumed. A fresh
// IAM lookup must not silently replace the secret that authenticated it.
if !crate::auth::constant_time_eq(&cred.secret_key, user.secret_key.expose()) {
return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied"));
}
if cred.is_temp() || cred.is_service_account() {
return Err(s3_error!(AccessDenied, "Access Denied"));
}
@@ -223,7 +235,8 @@ async fn handle_assume_role(
let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
return Err(s3_error!(InvalidRequest, "iam not init"));
};
let conditions = crate::auth::get_condition_values(&headers, &cred, None, None, remote_addr);
let mut conditions = crate::auth::get_condition_values(&headers, &cred, None, None, remote_addr);
let exp = set_assume_role_duration_condition(&mut conditions, body.duration_seconds);
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
@@ -260,7 +273,7 @@ async fn handle_assume_role(
// behaviour of every existing deployment is unchanged.
let mfa_verified = enforce_second_factor(&cred.access_key, &body, store, audit).await?;
let mut claims = cred.claims.unwrap_or_default();
let mut claims = cred.claims.clone().unwrap_or_default();
if mfa_verified {
// Recorded on the session so a later policy condition can require it,
@@ -271,8 +284,6 @@ async fn handle_assume_role(
populate_session_policy(&mut claims, &body.policy)?;
let exp = clamp_assume_role_duration(body.duration_seconds);
claims.insert(
"exp".to_string(),
Value::Number(serde_json::Number::from(
@@ -309,9 +320,9 @@ async fn handle_assume_role(
);
let updated_at = iam_store
.set_temp_user(&new_cred.access_key, &new_cred, None)
.set_temp_user_from_parent(&new_cred, &cred)
.await
.map_err(|_| s3_error!(InternalError, "set_temp_user failed"))?;
.map_err(super::iam_error::iam_error_to_s3_error)?;
let root_access_key = current_action_credentials().map(|cred| cred.access_key);
if root_access_key.as_deref() != Some(new_cred.parent_user.as_str())
@@ -604,6 +615,40 @@ mod tests {
assert_eq!(clamp(999999), 43200); // clamped to max
}
#[tokio::test]
async fn assume_role_duration_condition_enforces_policy_after_clamping() {
use rustfs_policy::policy::Policy;
let policy = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[{
"Effect":"Deny","Action":"sts:AssumeRole","Resource":"*",
"Condition":{"NumericGreaterThan":{"sts:DurationSeconds":"3600"}}
}]}"#,
)
.expect("duration policy");
for (requested, allowed) in [(0, true), (1, true), (3600, true), (3601, false), (usize::MAX, false)] {
let mut conditions = HashMap::from([("DurationSeconds".to_string(), vec!["1".to_string()])]);
let lifetime = set_assume_role_duration_condition(&mut conditions, requested);
assert_eq!(conditions["DurationSeconds"], [lifetime.to_string()]);
assert_eq!(
policy
.is_allowed(&Args {
account: "user",
groups: &None,
action: Action::StsAction(StsAction::AssumeRoleAction),
conditions: &conditions,
is_owner: false,
claims: &HashMap::new(),
deny_only: true,
bucket: "",
object: "",
})
.await,
allowed,
"requested={requested}"
);
}
}
#[test]
fn test_assume_role_duration_is_clamped_to_max() {
// Regression: the standard AssumeRole path previously used the raw client-supplied
@@ -24,7 +24,7 @@ use crate::admin::{
};
use crate::auth::{check_key_valid_with_context, get_session_token};
use crate::error::ApiError;
use crate::server::{RemoteAddr, TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX};
use crate::server::{TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX};
use crate::table_catalog::{DEFAULT_WAREHOUSE_ID, TableCatalogStore};
use bytes::Bytes;
use futures::{StreamExt, TryStreamExt, stream};
@@ -1141,7 +1141,7 @@ async fn authorize_table_catalog_request(req: &S3Request<Body>, action: AdminAct
&principal.credentials,
principal.owner,
Action::AdminAction(action),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
AdminResourceScope::bucket(""),
)
.await
@@ -1255,7 +1255,7 @@ async fn authorize_table_catalog_object_for_principal(
&principal.credentials,
principal.owner,
Action::AdminAction(action),
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
AdminResourceScope::bucket_object(bucket, object),
)
.await
+15 -6
View File
@@ -19,7 +19,7 @@
//! usage caches, with a one-level sub-prefix breakdown — the data console
//! buckets view MinIO serves from `loadPrefixUsageFromBackend`.
use crate::admin::auth::authorize_admin_request;
use crate::admin::auth::{authenticate_request, policy_source_addr, validate_admin_request_with_bucket};
use crate::admin::handlers::system::data_usage_info_gate_actions;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::server::ADMIN_PREFIX;
@@ -69,15 +69,24 @@ fn parse_usage_prefix_query(query: Option<&str>) -> S3Result<(String, usize)> {
#[async_trait::async_trait]
impl Operation for BucketPrefixUsageHandler {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
// The shared gate reports the same `InvalidRequest` "get cred failed" this
// handler has always returned for a credential-less request, so it needs no
// message-preserving pre-check.
authorize_admin_request(&req, data_usage_info_gate_actions()).await?;
let Some(input_cred) = req.credentials.as_ref() else {
return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "get cred failed"));
};
let (cred, owner) = authenticate_request(&req.headers, &req.uri, input_cred).await?;
let bucket = params.get("bucket").unwrap_or_default().to_string();
if bucket.is_empty() {
return Err(s3_error!(InvalidRequest, "bucket path parameter is required"));
}
validate_admin_request_with_bucket(
&req.headers,
&cred,
owner,
false,
data_usage_info_gate_actions(),
policy_source_addr(&req.extensions),
&bucket,
)
.await?;
let (prefix, max_entries) = parse_usage_prefix_query(req.uri.query())?;
// Authorization is bucket-scoped by the same any-of gate as the
+106 -12
View File
@@ -23,7 +23,6 @@ use crate::{
utils::{encode_compatible_admin_payload, has_space_be, read_compatible_admin_body},
},
auth::{check_key_valid, constant_time_eq, get_session_token},
server::RemoteAddr,
};
use http::{HeaderMap, StatusCode};
use matchit::Params;
@@ -31,7 +30,7 @@ use rustfs_config::{MAX_ADMIN_REQUEST_BODY_SIZE, MAX_IAM_IMPORT_EXPANDED_SIZE, M
use rustfs_credentials::Credentials;
use rustfs_iam::{
store::{GroupInfo, MappedPolicy, UserType},
sys::{NewServiceAccountOpts, UpdateServiceAccountOpts},
sys::{NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts},
};
use rustfs_madmin::{
AccountStatus, AddOrUpdateUserReq, IAMEntities, IAMErrEntities, IAMErrEntity, IAMErrPolicyEntity, SITE_REPL_API_VERSION,
@@ -196,6 +195,19 @@ fn imported_service_account_parent_scope_failure(
})
}
fn retain_exportable_service_accounts(accounts: &mut HashMap<String, rustfs_policy::auth::UserIdentity>) {
accounts.retain(|key, account| {
key != SITE_REPLICATOR_SERVICE_ACCOUNT && account.credentials.access_key != SITE_REPLICATOR_SERVICE_ACCOUNT
});
}
fn imported_service_account_reserved_failure(access_key: &str) -> Option<IAMErrEntity> {
(access_key == SITE_REPLICATOR_SERVICE_ACCOUNT).then(|| IAMErrEntity {
name: access_key.to_owned(),
error: "service account name is reserved for site replication".to_owned(),
})
}
fn imported_service_account_access_key_failure(entry_access_key: &str, payload_access_key: &str) -> Option<IAMErrEntity> {
(entry_access_key != payload_access_key).then(|| IAMErrEntity {
name: entry_access_key.to_string(),
@@ -208,6 +220,7 @@ pub struct AddUser {}
fn map_add_user_create_error(err: rustfs_iam::error::Error) -> S3Error {
match err {
rustfs_iam::error::Error::AccessKeyAlreadyExists => iam_error_to_s3_error(err),
rustfs_iam::error::Error::IAMActionNotAllowed => S3Error::new(S3ErrorCode::AccessDenied),
err => S3Error::with_message(S3ErrorCode::InternalError, format!("failed to create user: {err}")),
}
}
@@ -243,7 +256,7 @@ impl Operation for AddUser {
// let body_bytes = decrypt_data(input_cred.secret_key.expose().as_bytes(), &body)
// .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, format!("decrypt_data err {}", e)))?;
let args: AddOrUpdateUserReq = serde_json::from_slice(&body)
let mut args: AddOrUpdateUserReq = serde_json::from_slice(&body)
.map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid JSON: {e}")))?;
if args.secret_key.is_empty() {
@@ -261,7 +274,7 @@ impl Operation for AddUser {
};
if let Some(user) = iam_store.get_user(ak).await {
if (user.credentials.is_temp() || user.credentials.is_service_account()) && cred.parent_user == ak {
if user.credentials.is_temp() || user.credentials.is_service_account() {
return Err(s3_error!(InvalidArgument, "cannot create a user with a service account access key"));
}
} else if has_space_be(ak) {
@@ -306,11 +319,22 @@ impl Operation for AddUser {
owner,
check_deny_only,
vec![Action::AdminAction(AdminAction::CreateUserAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await?;
let updated_at = iam_store.create_user(ak, &args).await.map_err(map_add_user_create_error)?;
let updated_at = if check_deny_only {
let (updated_at, status, _) = iam_store
.change_own_secret_key(ak, input_cred.secret_key.expose(), &args.secret_key)
.await
.map_err(map_add_user_create_error)?;
args.status = status;
// A self-service secret rotation does not change policy attachment.
args.policy = None;
updated_at
} else {
iam_store.create_user(ak, &args).await.map_err(map_add_user_create_error)?
};
if let Err(err) = site_replication_iam_change_hook(SRIAMItem {
r#type: "iam-user".to_string(),
@@ -619,7 +643,7 @@ impl Operation for GetUserInfo {
owner,
check_deny_only,
vec![Action::AdminAction(AdminAction::GetUserAdminAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await?;
@@ -749,12 +773,9 @@ impl Operation for ExportIam {
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
retain_exportable_service_accounts(&mut service_accounts);
let mut svc_accts: HashMap<String, SRSvcAccCreate> = HashMap::new();
for (k, acc) in service_accounts {
if k == "siteReplicatorSvcAcc" {
continue;
}
let claims = iam_store
.get_claims_for_svc_acc(&acc.credentials.access_key)
.await
@@ -904,7 +925,7 @@ impl Operation for ImportIam {
owner,
false,
vec![Action::AdminAction(AdminAction::ImportIAMAction)],
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
crate::admin::auth::policy_source_addr(&req.extensions),
)
.await?;
@@ -1089,6 +1110,11 @@ impl Operation for ImportIam {
return Err(s3_error!(InvalidArgument, "has space be {ak}"));
}
if let Some(err) = imported_service_account_reserved_failure(&ak) {
failed.service_accounts.push(err);
continue;
}
if let Some(err) = imported_service_account_access_key_failure(&ak, &req.access_key) {
failed.service_accounts.push(err);
continue;
@@ -1463,6 +1489,74 @@ mod tests {
}
}
#[test]
fn iam_export_excludes_site_replicator_credentials() {
let mut accounts = HashMap::from([
(
"site-replicator-0".to_owned(),
rustfs_policy::auth::UserIdentity::new(Credentials {
access_key: "site-replicator-0".to_owned(),
secret_key: "replicator-secret".to_owned(),
..Default::default()
}),
),
(
"ordinary-service".to_owned(),
rustfs_policy::auth::UserIdentity::new(Credentials {
access_key: "ordinary-service".to_owned(),
secret_key: "ordinary-secret".to_owned(),
..Default::default()
}),
),
(
"mismatched-entry".to_owned(),
rustfs_policy::auth::UserIdentity::new(Credentials {
access_key: "site-replicator-0".to_owned(),
secret_key: "aliased-replicator-secret".to_owned(),
..Default::default()
}),
),
]);
super::retain_exportable_service_accounts(&mut accounts);
assert_eq!(accounts.len(), 1);
assert_eq!(accounts["ordinary-service"].credentials.secret_key, "ordinary-secret");
let exported = serde_json::to_string(&accounts).expect("serialize selected export records");
assert!(!exported.contains("replicator"));
}
#[test]
fn iam_archives_preserve_the_reserved_site_replicator_identity() {
let failure = super::imported_service_account_reserved_failure(rustfs_iam::sys::SITE_REPLICATOR_SERVICE_ACCOUNT)
.expect("the site replicator must never be replaced by an IAM import");
assert_eq!(failure.name, "site-replicator-0");
assert_eq!(failure.error, "service account name is reserved for site replication");
assert!(super::imported_service_account_reserved_failure("ordinary-service-account").is_none());
let source = include_str!("user.rs")
.split("\n#[cfg(test)]\n")
.next()
.expect("production source");
let export = source_block(source, "impl Operation for ExportIam");
let guard = export
.find("retain_exportable_service_accounts(&mut service_accounts)")
.expect("export uses the canonical reserved identity");
assert!(
guard
< export
.find("let svc_acc_create_req = SRSvcAccCreate")
.expect("exported credentials")
);
let import = source_block(source, "impl Operation for ImportIam");
let guard = import
.find("imported_service_account_reserved_failure(&ak)")
.expect("import rejects the reserved identity");
assert!(
guard
< import
.find(".delete_service_account(&ak")
.expect("replacement deletes the original account")
);
}
#[test]
fn import_iam_enqueues_a_site_replication_snapshot() {
let body = source_block(include_str!("user.rs"), "impl Operation for ImportIam");
+5 -7
View File
@@ -18,7 +18,6 @@ use rustfs_security_governance::{
};
const ALL_ADMIN: AdminActionRef = AdminActionRef::new("AllAdminActions");
const ADD_USER_TO_GROUP: AdminActionRef = AdminActionRef::new("AddUserToGroupAdminAction");
const ATTACH_POLICY: AdminActionRef = AdminActionRef::new("AttachPolicyAdminAction");
const CONFIG_UPDATE: AdminActionRef = AdminActionRef::new("ConfigUpdateAdminAction");
const TLS_CERTIFICATE_PUBLISH: AdminActionRef = AdminActionRef::new("TLSCertificatePublishAdminAction");
@@ -187,12 +186,6 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
RouteRiskLevel::High,
),
admin(HttpMethod::Put, "/rustfs/admin/v3/set-group-status", ENABLE_GROUP, RouteRiskLevel::High),
admin(
HttpMethod::Put,
"/rustfs/admin/v3/update-group-members",
ADD_USER_TO_GROUP,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/rustfs/admin/v3/temporary-account-info",
@@ -1806,6 +1799,11 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
];
pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[
deferred(
HttpMethod::Put,
"/rustfs/admin/v3/update-group-members",
DeferredRoutePolicyReason::MultipleActions,
),
deferred(HttpMethod::Get, "/rustfs/admin/v3/accountinfo", DeferredRoutePolicyReason::S3Action),
// The self-service account routes act on the caller, never on a target
// named in the request, so they gate on possession of the credential (plus,
+178 -8
View File
@@ -20,7 +20,7 @@ use crate::admin::{
use rustfs_iam::{
federation::{FederatedSessionBinding, FederatedSessionBindingError, FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM},
store::{MappedPolicy, object::ObjectStore},
sys::{IamSys, is_safe_claim_policy_name},
sys::{IamSys, OIDC_ISSUER_CLAIM, is_safe_claim_policy_name},
};
use rustfs_madmin::{SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC, SRIAMItem, SRSTSCredential};
use rustfs_policy::auth::get_new_credentials_with_metadata;
@@ -111,6 +111,9 @@ fn build_oidc_token_claims(transaction: &FederatedSessionTransaction) -> HashMap
let mut token_claims = HashMap::new();
token_claims.insert("sub".to_string(), Value::String(authorization.subject().to_string()));
token_claims.insert("iss".to_string(), Value::String("rustfs-oidc".to_string()));
if let Some(issuer) = authorization.issuer() {
token_claims.insert(OIDC_ISSUER_CLAIM.to_string(), Value::String(issuer.to_string()));
}
token_claims.insert("oidc_provider".to_string(), Value::String(authorization.provider_id().to_string()));
if !authorization.email().is_empty() {
@@ -329,14 +332,31 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding {
let parent_user = authorization.oidc_virtual_parent().ok_or_else(|| {
FederatedSessionBindingError::InvalidRequest("verified OIDC identity is missing issuer or subject".to_string())
})?;
let (selected_policy_names, ignorable_policy_names) = match iam_store
.sts_policy_db_get(&parent_user, &Some(authorization.mapped_iam_groups().to_vec()))
let (mapped_policy_names, has_identity_mapping) = iam_store
.sts_policy_db_get_with_identity_mapping(&parent_user, &Some(authorization.mapped_iam_groups().to_vec()))
.await
.map_err(|_| FederatedSessionBindingError::Internal("failed to resolve OIDC policy mapping".to_string()))?
{
mapped_policy_names if !mapped_policy_names.is_empty() => (mapped_policy_names, &[][..]),
_ => (authorization.policies().to_vec(), authorization.group_claim_policies()),
.map_err(|_| FederatedSessionBindingError::Internal("failed to resolve OIDC policy mapping".to_string()))?;
// An explicit identity mapping overrides token policies. Group mappings
// are additive and must not discard claim-derived Deny policies.
let ignorable_policy_names: Vec<String> = if has_identity_mapping {
Vec::new()
} else {
authorization
.group_claim_policies()
.iter()
.filter(|name| !mapped_policy_names.contains(name))
.cloned()
.collect()
};
let mut selected_policy_names = mapped_policy_names;
if !has_identity_mapping {
for name in authorization.policies() {
if !selected_policy_names.contains(name) {
selected_policy_names.push(name.clone());
}
}
}
// A name may be skipped only after storage confirms the policy is absent. A failed lookup
// must not be mistaken for absence, or an unreadable Deny policy would be dropped from the
// signed session and could not be restored at request time.
@@ -368,7 +388,7 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding {
let resolved_policy_names = MappedPolicy::new(&resolved_policy_mapping).to_slice();
let selection = select_oidc_policies(
&selected_policy_names,
ignorable_policy_names,
&ignorable_policy_names,
&resolved_policy_names,
&missing_policy_names,
)
@@ -543,6 +563,7 @@ mod tests {
HashMap::from([
("sub".to_string(), serde_json::json!("subject")),
("iss".to_string(), serde_json::json!("rustfs-oidc")),
(OIDC_ISSUER_CLAIM.to_string(), serde_json::json!("https://idp.example.test")),
("oidc_provider".to_string(), serde_json::json!("default")),
("email".to_string(), serde_json::json!("user@example.com")),
("preferred_username".to_string(), serde_json::json!("user")),
@@ -566,6 +587,7 @@ mod tests {
HashMap::from([
("sub".to_string(), serde_json::json!("subject")),
("iss".to_string(), serde_json::json!("rustfs-oidc")),
(OIDC_ISSUER_CLAIM.to_string(), serde_json::json!("https://idp.example.test")),
("oidc_provider".to_string(), serde_json::json!("default")),
])
);
@@ -641,6 +663,7 @@ mod tests {
HashMap::from([
("sub".to_string(), serde_json::json!("subject")),
("iss".to_string(), serde_json::json!("rustfs-oidc")),
(OIDC_ISSUER_CLAIM.to_string(), serde_json::json!("https://idp.example.test")),
("oidc_provider".to_string(), serde_json::json!("default")),
("email".to_string(), serde_json::json!("user@example.com")),
("preferred_username".to_string(), serde_json::json!("user")),
@@ -684,6 +707,153 @@ mod tests {
);
}
#[tokio::test]
#[serial]
async fn binding_group_policies_preserve_claim_denies() {
let iam = ready_test_iam().await;
let deny = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[{
"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::payroll/*"
}]}"#,
)
.expect("parse payroll Deny");
iam.set_policy("claim-deny-payroll", deny).await.expect("store claim Deny");
iam.add_users_to_group("claim-policy-group", Vec::new())
.await
.expect("create mapped group");
iam.policy_db_set("claim-policy-group", UserType::Reg, true, "readwrite")
.await
.expect("attach group Allow");
let authorization = || {
let mut fixture = authorization_fixture();
fixture.sub = "claim-deny-subject".to_string();
fixture.policies = vec!["readwrite".to_string(), "claim-deny-payroll".to_string()];
fixture.mapped_iam_groups = vec!["claim-policy-group".to_string()];
fixture
};
let credentials = DefaultFederatedSessionBinding
.bind(&transaction_from_fixture(authorization()))
.await
.expect("bind group plus claim policies");
let claims = get_claims_for_test(&credentials);
let selected = MappedPolicy::new(claims.get("policy").and_then(Value::as_str).expect("policy claim")).to_slice();
assert!(selected.contains(&"claim-deny-payroll".to_string()));
let conditions = HashMap::new();
let args = Args {
account: &credentials.access_key,
groups: &credentials.groups,
action: Action::S3Action(S3Action::GetObjectAction),
bucket: "payroll",
object: "salary.csv",
conditions: &conditions,
claims: &claims,
is_owner: false,
deny_only: false,
};
assert!(!iam.is_allowed(&args).await, "group Allow must not discard claim Deny");
let mut authorization = authorization();
authorization.policies.push("missing-required-policy".to_string());
assert!(
DefaultFederatedSessionBinding
.bind(&transaction_from_fixture(authorization))
.await
.is_err(),
"group mappings must not hide unresolved required claim policies"
);
}
fn get_claims_for_test(credentials: &rustfs_credentials::Credentials) -> HashMap<String, Value> {
let signing_key = current_token_signing_key().expect("test signing key");
rustfs_iam::sys::get_claims_from_token_with_secret(&credentials.session_token, &signing_key)
.expect("verify issued session token")
}
#[tokio::test]
#[serial]
async fn group_conditions_follow_live_parent_membership_for_all_credentials() {
let iam = ready_test_iam().await;
let user = "live-group-condition-user";
let group = "live-restricted-group";
iam.create_user(
user,
&AddOrUpdateUserReq {
secret_key: "live-group-user-secret".to_string(),
policy: None,
status: AccountStatus::Enabled,
},
)
.await
.expect("create group condition user");
let policy = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*"},
{"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*",
"Condition":{"StringEquals":{"aws:groups":"live-restricted-group"}}}
]}"#,
)
.expect("parse group condition policy");
for member in [true, false, true] {
if member {
iam.add_users_to_group(group, vec![user.to_string()])
.await
.expect("add membership");
} else {
iam.remove_users_from_group(group, vec![user.to_string()])
.await
.expect("remove membership");
}
for kind in ["regular", "sts", "service"] {
let mut credentials = rustfs_credentials::Credentials {
access_key: if kind == "regular" {
user
} else {
"derived-group-condition-key"
}
.to_string(),
secret_key: "valid-derived-secret".to_string(),
status: "on".to_string(),
groups: Some(vec!["credential-group".to_string()]),
..Default::default()
};
if kind != "regular" {
credentials.parent_user = user.to_string();
credentials.session_token = "signed-token".to_string();
}
credentials.claims = Some(HashMap::from([("groups".to_string(), serde_json::json!(["token-group"]))]));
if kind == "service" {
credentials
.claims
.as_mut()
.expect("claims")
.insert("sa-policy".to_string(), serde_json::json!("inherited-policy"));
}
let conditions = crate::auth::get_condition_values(&http::HeaderMap::new(), &credentials, None, None, None);
let groups = conditions.get("groups").expect("credential and token groups");
assert!(groups.contains(&"token-group".to_string()));
assert!(groups.contains(&"credential-group".to_string()));
assert_eq!(groups.contains(&group.to_string()), member, "{kind}");
let claims = HashMap::new();
assert_eq!(
policy
.is_allowed(&Args {
account: &credentials.access_key,
groups: &credentials.groups,
action: Action::S3Action(S3Action::GetObjectAction),
bucket: "bucket",
object: "object",
conditions: &conditions,
claims: &claims,
is_owner: false,
deny_only: false,
})
.await,
!member,
"{kind}"
);
}
}
}
#[tokio::test]
#[serial]
async fn binding_uses_sts_policy_when_regular_mapping_collides() {
+76 -5
View File
@@ -19,7 +19,8 @@ use http::header::{IF_MATCH, IF_NONE_MATCH};
use rustfs_credentials::Credentials;
use rustfs_iam::error::Error as IamError;
use rustfs_iam::sys::{
SESSION_POLICY_NAME, get_claims_from_token_with_secret, get_claims_from_token_with_secret_allow_missing_exp,
OIDC_ISSUER_CLAIM, SESSION_POLICY_NAME, get_claims_from_token_with_secret,
get_claims_from_token_with_secret_allow_missing_exp, is_rustfs_oidc_claims,
};
use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive, is_server_derived_condition_key};
use rustfs_trusted_proxies::ClientInfo;
@@ -801,6 +802,17 @@ pub fn get_condition_values_with_query_and_client_info(
}
}
if is_rustfs_oidc_claims(claims) {
args.remove("iss");
if let Some(issuer) = claims
.get(OIDC_ISSUER_CLAIM)
.and_then(Value::as_str)
.filter(|issuer| !issuer.is_empty())
{
args.insert("iss".to_string(), vec![issuer.to_string()]);
}
}
let grps = extract_string_list_claim(claims, "groups");
if !grps.is_empty() {
args.insert("groups".to_string(), grps);
@@ -812,10 +824,19 @@ pub fn get_condition_values_with_query_and_client_info(
}
}
if let Some(groups) = &cred.groups
&& !args.contains_key("groups")
{
args.insert("groups".to_string(), groups.clone());
let effective_user = if (cred.is_service_account() || cred.is_temp()) && !cred.parent_user.is_empty() {
&cred.parent_user
} else {
&cred.access_key
};
let live_groups = current_ready_iam_handle()
.map(|iam| iam.user_group_memberships(effective_user))
.unwrap_or_default();
for group in cred.groups.iter().flatten().chain(live_groups.iter()) {
let groups = args.entry("groups".to_string()).or_default();
if !groups.contains(group) {
groups.push(group.clone());
}
}
// Every remaining header is attacker-controlled. A header must never contribute
@@ -2305,6 +2326,56 @@ mod tests {
// (we can't directly test this without changing the function signature)
}
#[tokio::test]
async fn oidc_issuer_condition_uses_verified_provider_not_session_marker_or_header() {
use rustfs_policy::policy::{
Args, Policy,
action::{Action, S3Action},
};
let policy = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*"},
{"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*",
"Condition":{"StringEquals":{"jwt:iss":"https://partner.example"}}}
]}"#,
)
.expect("parse issuer condition");
let mut credentials = create_test_credentials();
let mut headers = HeaderMap::new();
headers.insert("iss", HeaderValue::from_static("https://spoofed.invalid"));
headers.insert(OIDC_ISSUER_CLAIM, HeaderValue::from_static("https://spoofed.invalid"));
for issuer in [Some("https://partner.example"), Some("https://corporate.example"), None] {
let mut claims = HashMap::from([
("iss".to_string(), json!("rustfs-oidc")),
("oidc_provider".to_string(), json!("provider")),
("sub".to_string(), json!("subject")),
]);
if let Some(issuer) = issuer {
claims.insert(OIDC_ISSUER_CLAIM.to_string(), json!(issuer));
}
credentials.claims = Some(claims);
let conditions = get_condition_values(&headers, &credentials, None, None, None);
assert_eq!(conditions.get("iss"), issuer.map(|value| vec![value.to_string()]).as_ref());
assert_eq!(
policy
.is_allowed(&Args {
account: &credentials.access_key,
groups: &credentials.groups,
action: Action::S3Action(S3Action::GetObjectAction),
bucket: "bucket",
object: "object",
conditions: &conditions,
claims: credentials.claims_or_empty(),
is_owner: false,
deny_only: false,
})
.await,
issuer != Some("https://partner.example"),
);
}
}
#[test]
fn test_get_condition_values_with_claims() {
let mut cred = create_service_account_credentials();