From 1c88aa43c1c1a7a9377ae9e294b99aa5ade3a248 Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Fri, 24 Jul 2026 17:41:52 +0800 Subject: [PATCH] fix(iam): virtualize OIDC service account parents (#5152) * fix(iam): preserve OIDC service account policy boundary * fix(iam): virtualize OIDC service account parents * fix(iam): reject malformed OIDC policy boundaries * test(iam): isolate federated policy regression * fix(iam): keep OIDC replication envelope off claims --- Cargo.lock | 1 + crates/iam/Cargo.toml | 2 +- crates/iam/src/cache.rs | 7 + crates/iam/src/federation/mod.rs | 2 +- crates/iam/src/federation/model.rs | 87 +- crates/iam/src/lib.rs | 12 - crates/iam/src/manager.rs | 237 ++++- crates/iam/src/sys.rs | 862 +++++++++++++++++- crates/madmin/src/site_replication.rs | 7 + rustfs/Cargo.toml | 1 + rustfs/src/admin/handlers/service_account.rs | 310 ++++--- rustfs/src/admin/handlers/site_replication.rs | 439 ++++++++- rustfs/src/admin/handlers/user.rs | 5 +- .../src/admin/service/federated_identity.rs | 127 ++- rustfs/src/auth.rs | 39 +- rustfs/src/storage/rpc/node_service.rs | 74 +- 16 files changed, 1973 insertions(+), 239 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4f1dbc2c1..a73b2b96f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8928,6 +8928,7 @@ dependencies = [ "rustfs-signer", "rustfs-storage-api", "rustfs-targets", + "rustfs-test-utils", "rustfs-tls-runtime", "rustfs-trusted-proxies", "rustfs-utils", diff --git a/crates/iam/Cargo.toml b/crates/iam/Cargo.toml index e2b7191c4..8c4a3c2aa 100644 --- a/crates/iam/Cargo.toml +++ b/crates/iam/Cargo.toml @@ -47,7 +47,7 @@ base64-simd = { workspace = true } jsonwebtoken = { workspace = true, features = ["aws_lc_rs"] } tracing.workspace = true rustfs-madmin.workspace = true -rustfs-utils = { workspace = true, features = ["path", "egress"] } +rustfs-utils = { workspace = true, features = ["egress", "hash", "path"] } rustfs-io-metrics.workspace = true tokio-util = { workspace = true, features = ["io", "compat"] } pollster.workspace = true diff --git a/crates/iam/src/cache.rs b/crates/iam/src/cache.rs index 564eb0126..f0ff24d30 100644 --- a/crates/iam/src/cache.rs +++ b/crates/iam/src/cache.rs @@ -25,6 +25,7 @@ use rustfs_policy::{ policy::{Args, PolicyDoc}, }; use time::OffsetDateTime; +use tokio::sync::Mutex as AsyncMutex; use tracing::warn; use crate::store::{GroupInfo, MappedPolicy}; @@ -63,6 +64,7 @@ impl Default for CacheState { pub struct Cache { state: ArcSwap, write_lock: Mutex<()>, + service_account_mutation_lock: AsyncMutex<()>, } impl Default for Cache { @@ -70,6 +72,7 @@ impl Default for Cache { Self { state: ArcSwap::new(Arc::new(CacheState::default())), write_lock: Mutex::new(()), + service_account_mutation_lock: AsyncMutex::new(()), } } } @@ -77,6 +80,10 @@ impl Default for Cache { pub(crate) type CacheSnapshot = Guard>; impl Cache { + pub(crate) fn service_account_mutation_lock(&self) -> &AsyncMutex<()> { + &self.service_account_mutation_lock + } + pub(crate) fn snapshot(&self) -> CacheSnapshot { self.state.load() } diff --git a/crates/iam/src/federation/mod.rs b/crates/iam/src/federation/mod.rs index aea1c90f3..8611bbcf9 100644 --- a/crates/iam/src/federation/mod.rs +++ b/crates/iam/src/federation/mod.rs @@ -24,7 +24,7 @@ pub use binding::FederatedSessionBinding; pub use error::{FederatedSessionBindingError, FederationError, Result}; pub use model::{ FederatedAuthorization, FederatedClaims, FederatedCodeExchange, FederatedLoginSession, FederatedSession, - FederatedSessionTransaction, + FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM, }; pub use provider::FederatedIdentityProvider; pub use registry::FederatedIdentityRegistry; diff --git a/crates/iam/src/federation/model.rs b/crates/iam/src/federation/model.rs index 1645f8581..e0ce0222d 100644 --- a/crates/iam/src/federation/model.rs +++ b/crates/iam/src/federation/model.rs @@ -13,9 +13,13 @@ // limitations under the License. use rustfs_credentials::Credentials; +use rustfs_utils::HashAlgorithm; use serde_json::Value; use std::collections::HashMap; +pub const OIDC_VIRTUAL_PARENT_CLAIM: &str = "x-rustfs-internal-oidc-parent"; +const OIDC_VIRTUAL_PARENT_PREFIX: &str = "openid="; + #[derive(Debug, Clone)] pub struct FederatedClaims { pub sub: String, @@ -53,6 +57,27 @@ impl FederatedAuthorization { pub fn has_authorization_context(&self) -> bool { !self.policies.is_empty() || !self.groups.is_empty() } + + pub fn oidc_virtual_parent(&self) -> Option { + let issuer = self.claims.raw.get("iss")?.as_str()?; + let subject = self.claims.sub.as_str(); + if issuer.is_empty() || subject.is_empty() { + return None; + } + + let subject_len = u64::try_from(subject.len()).ok()?; + let issuer_len = u64::try_from(issuer.len()).ok()?; + let mut source = Vec::with_capacity(16 + subject.len() + issuer.len()); + source.extend_from_slice(&subject_len.to_be_bytes()); + source.extend_from_slice(subject.as_bytes()); + source.extend_from_slice(&issuer_len.to_be_bytes()); + source.extend_from_slice(issuer.as_bytes()); + let digest = HashAlgorithm::SHA256.hash_encode(&source); + Some(format!( + "{OIDC_VIRTUAL_PARENT_PREFIX}{}", + base64_simd::URL_SAFE_NO_PAD.encode_to_string(digest.as_ref()) + )) + } } #[derive(Debug)] @@ -99,7 +124,10 @@ mod tests { fn authorization(policies: Vec, groups: Vec) -> FederatedAuthorization { FederatedAuthorization { provider_id: "standard_oidc".to_string(), - claims: claims("", "", "subject"), + claims: FederatedClaims { + raw: HashMap::from([("iss".to_string(), Value::String("https://idp.example.test".to_string()))]), + ..claims("", "", "subject") + }, policies, groups, roles_claim_key: None, @@ -121,4 +149,61 @@ mod tests { assert!(authorization(vec!["consoleAdmin".to_string()], Vec::new()).has_authorization_context()); assert!(authorization(Vec::new(), vec!["RustFS.ConsoleAdmin".to_string()]).has_authorization_context()); } + + #[test] + fn oidc_virtual_parent_is_stable_and_issuer_scoped() { + let first = authorization(Vec::new(), Vec::new()); + let mut second = first.clone(); + second + .claims + .raw + .insert("iss".to_string(), Value::String("https://other-idp.example.test".to_string())); + + assert_eq!( + first.oidc_virtual_parent().as_deref(), + Some("openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I") + ); + assert!(rustfs_policy::auth::contains_reserved_chars( + first.oidc_virtual_parent().as_deref().expect("virtual parent") + )); + assert_ne!(first.oidc_virtual_parent(), second.oidc_virtual_parent()); + } + + #[test] + fn oidc_virtual_parent_requires_verified_identity_parts() { + let mut missing_issuer = authorization(Vec::new(), Vec::new()); + missing_issuer.claims.raw.clear(); + let mut missing_subject = authorization(Vec::new(), Vec::new()); + missing_subject.claims.sub.clear(); + + assert!(missing_issuer.oidc_virtual_parent().is_none()); + assert!(missing_subject.oidc_virtual_parent().is_none()); + } + + #[test] + fn oidc_virtual_parent_preserves_exact_subject() { + let plain = authorization(Vec::new(), Vec::new()); + let mut padded = plain.clone(); + padded.claims.sub = format!(" {} ", plain.claims.sub); + + assert_ne!(plain.oidc_virtual_parent(), padded.oidc_virtual_parent()); + } + + #[test] + fn oidc_virtual_parent_encoding_is_unambiguous() { + let mut first = authorization(Vec::new(), Vec::new()); + first.claims.sub = "subject".to_string(); + first.claims.raw.insert( + "iss".to_string(), + Value::String("https://issuer.example/path:https://other.example".to_string()), + ); + let mut second = authorization(Vec::new(), Vec::new()); + second.claims.sub = "subject:https://issuer.example/path".to_string(); + second + .claims + .raw + .insert("iss".to_string(), Value::String("https://other.example".to_string())); + + assert_ne!(first.oidc_virtual_parent(), second.oidc_virtual_parent()); + } } diff --git a/crates/iam/src/lib.rs b/crates/iam/src/lib.rs index 3f54282cd..316e064c9 100644 --- a/crates/iam/src/lib.rs +++ b/crates/iam/src/lib.rs @@ -126,18 +126,6 @@ pub(crate) async fn notify_iam_load_service_account(access_key: &str) -> Vec Vec { - match runtime_sources::notification_sys() { - Some(notification_sys) => notification_sys - .delete_service_account(access_key) - .await - .into_iter() - .map(Into::into) - .collect(), - None => Vec::new(), - } -} - pub(crate) async fn notify_iam_load_group(group: &str) -> Vec { match runtime_sources::notification_sys() { Some(notification_sys) => notification_sys.load_group(group).await.into_iter().map(Into::into).collect(), diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index 861f00ae1..9a1afff18 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -782,6 +782,7 @@ where return Err(Error::InvalidArgument); } + let _mutation_guard = self.cache.service_account_mutation_lock().lock().await; let cache = self.cache.snapshot(); if cache.users.contains_key(&cred.access_key) || cache.sts_accounts.contains_key(&cred.access_key) { return Err(Error::AccessKeyAlreadyExists); @@ -800,6 +801,7 @@ where } pub async fn update_service_account(&self, name: &str, opts: UpdateServiceAccountOpts) -> Result { + let _mutation_guard = self.cache.service_account_mutation_lock().lock().await; let cache = self.cache.snapshot(); let Some(ui) = cache.users.get(name).cloned() else { return Err(Error::NoSuchServiceAccount(name.to_string())); @@ -891,12 +893,13 @@ where cr.session_token = jwt_sign(&m, &cr.secret_key)?; let u = UserIdentity::new(cr); + let updated_at = u.update_at.unwrap_or_else(OffsetDateTime::now_utc); self.api .save_user_identity(&u.credentials.access_key, UserType::Svc, u.clone(), None) .await?; self.update_user_with_claims(&u.credentials.access_key, u.clone())?; - Ok(OffsetDateTime::now_utc()) + Ok(updated_at) } pub async fn policy_db_get(&self, name: &str, groups: &Option>) -> Result> { @@ -1361,10 +1364,25 @@ where } pub async fn delete_user(&self, access_key: &str, utype: UserType) -> Result<()> { + self.delete_user_with_revision(access_key, utype).await?; + Ok(()) + } + + pub async fn delete_service_account_with_revision(&self, access_key: &str) -> Result { + self.delete_user_with_revision(access_key, UserType::Svc).await + } + + async fn delete_user_with_revision(&self, access_key: &str, utype: UserType) -> Result { if access_key.is_empty() { return Err(Error::InvalidArgument); } + let _service_account_guard = if utype == UserType::Svc { + Some(self.cache.service_account_mutation_lock().lock().await) + } else { + None + }; + if utype == UserType::Reg { let cache = self.cache.snapshot(); let member_of = cache.user_group_memberships.get(access_key).cloned(); @@ -1429,15 +1447,15 @@ where return Err(err); } + let deleted_at = OffsetDateTime::now_utc(); self.cache.with_write_lock(|cache| { - let now = OffsetDateTime::now_utc(); if utype == UserType::Sts { - cache.delete_sts_account(access_key, now); + cache.delete_sts_account(access_key, deleted_at); } - cache.delete_user(access_key, now); + cache.delete_user(access_key, deleted_at); }); - Ok(()) + Ok(deleted_at) } pub async fn update_user_secret_key(&self, access_key: &str, secret_key: &str) -> Result<()> { @@ -1958,6 +1976,11 @@ where Ok(()) } pub async fn user_notification_handler(&self, name: &str, user_type: UserType) -> Result<()> { + let _service_account_guard = if user_type == UserType::Svc { + Some(self.cache.service_account_mutation_lock().lock().await) + } else { + None + }; let mut m = HashMap::new(); if let Err(err) = self.api.load_user_no_lock(name, user_type, &mut m).await { if !is_err_no_such_user(&err) { @@ -2222,9 +2245,10 @@ mod tests { collections::HashMap, sync::{ Arc, Mutex, - atomic::{AtomicUsize, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }, }; + use tokio::sync::Notify; #[derive(Clone)] struct FailingInitialLoadStore; @@ -2357,6 +2381,12 @@ mod tests { saved_user: Arc>>, load_attempts: Arc, visible_after_attempt: usize, + block_service_save: Arc, + service_save_started: Arc, + release_service_save: Arc, + block_service_load: Arc, + service_load_started: Arc, + release_service_load: Arc, } impl DelayedTempUserVisibilityStore { @@ -2365,6 +2395,12 @@ mod tests { saved_user: Arc::new(Mutex::new(None)), load_attempts: Arc::new(AtomicUsize::new(0)), visible_after_attempt, + block_service_save: Arc::new(AtomicBool::new(false)), + service_save_started: Arc::new(Notify::new()), + release_service_save: Arc::new(Notify::new()), + block_service_load: Arc::new(AtomicBool::new(false)), + service_load_started: Arc::new(Notify::new()), + release_service_load: Arc::new(Notify::new()), } } } @@ -2390,16 +2426,21 @@ mod tests { async fn save_user_identity( &self, _name: &str, - _user_type: UserType, + user_type: UserType, item: UserIdentity, _ttl: Option, ) -> Result<()> { + if user_type == UserType::Svc && self.block_service_save.load(Ordering::SeqCst) { + self.service_save_started.notify_one(); + self.release_service_save.notified().await; + } *self.saved_user.lock().expect("saved_user mutex poisoned") = Some(item); Ok(()) } async fn delete_user_identity(&self, _name: &str, _user_type: UserType) -> Result<()> { - Err(Error::InvalidArgument) + *self.saved_user.lock().expect("saved_user mutex poisoned") = None; + Ok(()) } async fn load_user_identity(&self, name: &str, _user_type: UserType) -> Result { @@ -2415,8 +2456,19 @@ mod tests { .ok_or_else(|| Error::NoSuchUser(name.to_string())) } - async fn load_user(&self, _name: &str, _user_type: UserType, _m: &mut HashMap) -> Result<()> { - Err(Error::InvalidArgument) + async fn load_user(&self, name: &str, user_type: UserType, m: &mut HashMap) -> Result<()> { + let loaded = self + .saved_user + .lock() + .expect("saved_user mutex poisoned") + .clone() + .ok_or_else(|| Error::NoSuchUser(name.to_string()))?; + if user_type == UserType::Svc && self.block_service_load.load(Ordering::SeqCst) { + self.service_load_started.notify_one(); + self.release_service_load.notified().await; + } + m.insert(name.to_string(), loaded); + Ok(()) } async fn load_users(&self, _user_type: UserType, _m: &mut HashMap) -> Result<()> { @@ -2485,7 +2537,7 @@ mod tests { _is_group: bool, _m: &mut HashMap, ) -> Result<()> { - Err(Error::InvalidArgument) + Err(Error::NoSuchPolicy) } async fn load_mapped_policies( @@ -2576,6 +2628,169 @@ mod tests { ); } + #[tokio::test] + async fn service_account_notification_cannot_overwrite_concurrent_update() { + let _ = rustfs_credentials::init_global_action_credentials( + Some("TESTROOTACCESSKEY".to_string()), + Some("TESTROOTSECRET123".to_string()), + ); + let store = DelayedTempUserVisibilityStore::new(0); + let cache = Arc::new(build_test_iam_cache(store.clone())); + let access_key = "SERIALIZEDSERVICE01"; + let secret_key = "serializedServiceSecret123"; + let metadata = HashMap::from([ + ("parent".to_string(), Value::String("parent-user".to_string())), + (iam_policy_claim_name_sa(), Value::String(INHERITED_POLICY_TYPE.to_string())), + ]); + let credentials = Credentials { + access_key: access_key.to_string(), + secret_key: secret_key.to_string(), + session_token: jwt_sign(&metadata, secret_key).expect("sign service account token"), + status: auth::ACCOUNT_ON.to_string(), + parent_user: "parent-user".to_string(), + claims: Some(metadata), + description: Some("old".to_string()), + ..Default::default() + }; + cache.add_service_account(credentials).await.expect("seed service account"); + + store.block_service_load.store(true, Ordering::SeqCst); + let notification = { + let cache = Arc::clone(&cache); + tokio::spawn(async move { cache.user_notification_handler(access_key, UserType::Svc).await }) + }; + store.service_load_started.notified().await; + + let update = { + let cache = Arc::clone(&cache); + tokio::spawn(async move { + cache + .update_service_account( + access_key, + UpdateServiceAccountOpts { + session_policy: None, + secret_key: None, + name: None, + description: Some("new".to_string()), + expiration: None, + status: None, + allow_site_replicator_account: false, + }, + ) + .await + }) + }; + tokio::task::yield_now().await; + assert!(!update.is_finished(), "update must wait for the in-flight cache refresh"); + + store.release_service_load.notify_one(); + notification.await.expect("notification task").expect("notification refresh"); + update.await.expect("update task").expect("service account update"); + + let snapshot = cache.cache.snapshot(); + assert_eq!( + snapshot + .users + .get(access_key) + .and_then(|identity| identity.credentials.description.as_deref()), + Some("new") + ); + } + + #[tokio::test] + async fn concurrent_service_account_create_cannot_overwrite_first_writer() { + let store = DelayedTempUserVisibilityStore::new(0); + store.block_service_save.store(true, Ordering::SeqCst); + let cache = Arc::new(build_test_iam_cache(store.clone())); + let access_key = "SERIALIZEDSERVICE00"; + let credentials = |secret_key: &str| Credentials { + access_key: access_key.to_string(), + secret_key: secret_key.to_string(), + status: auth::ACCOUNT_ON.to_string(), + parent_user: "parent-user".to_string(), + ..Default::default() + }; + + let first = { + let cache = Arc::clone(&cache); + tokio::spawn(async move { cache.add_service_account(credentials("firstServiceSecret123")).await }) + }; + store.service_save_started.notified().await; + + let second = { + let cache = Arc::clone(&cache); + tokio::spawn(async move { cache.add_service_account(credentials("secondServiceSecret234")).await }) + }; + tokio::task::yield_now().await; + assert!(!second.is_finished(), "second create must wait for the first writer"); + + store.block_service_save.store(false, Ordering::SeqCst); + store.release_service_save.notify_waiters(); + first.await.expect("first create task").expect("first create"); + let err = second + .await + .expect("second create task") + .expect_err("duplicate create must fail"); + + assert_eq!(err, Error::AccessKeyAlreadyExists); + assert_eq!( + cache + .cache + .snapshot() + .users + .get(access_key) + .map(|identity| identity.credentials.secret_key.as_str()), + Some("firstServiceSecret123") + ); + } + + #[tokio::test] + async fn service_account_notification_cannot_restore_concurrent_delete() { + let _ = rustfs_credentials::init_global_action_credentials( + Some("TESTROOTACCESSKEY".to_string()), + Some("TESTROOTSECRET123".to_string()), + ); + let store = DelayedTempUserVisibilityStore::new(0); + let cache = Arc::new(build_test_iam_cache(store.clone())); + let access_key = "SERIALIZEDSERVICE02"; + let secret_key = "serializedServiceSecret234"; + let metadata = HashMap::from([ + ("parent".to_string(), Value::String("parent-user".to_string())), + (iam_policy_claim_name_sa(), Value::String(INHERITED_POLICY_TYPE.to_string())), + ]); + let credentials = Credentials { + access_key: access_key.to_string(), + secret_key: secret_key.to_string(), + session_token: jwt_sign(&metadata, secret_key).expect("sign service account token"), + status: auth::ACCOUNT_ON.to_string(), + parent_user: "parent-user".to_string(), + claims: Some(metadata), + ..Default::default() + }; + cache.add_service_account(credentials).await.expect("seed service account"); + + store.block_service_load.store(true, Ordering::SeqCst); + let notification = { + let cache = Arc::clone(&cache); + tokio::spawn(async move { cache.user_notification_handler(access_key, UserType::Svc).await }) + }; + store.service_load_started.notified().await; + + let delete = { + let cache = Arc::clone(&cache); + tokio::spawn(async move { cache.delete_service_account_with_revision(access_key).await }) + }; + tokio::task::yield_now().await; + assert!(!delete.is_finished(), "delete must wait for the in-flight cache refresh"); + + store.release_service_load.notify_one(); + notification.await.expect("notification task").expect("notification refresh"); + delete.await.expect("delete task").expect("service account delete"); + + assert!(!cache.cache.snapshot().users.contains_key(access_key)); + assert!(store.saved_user.lock().expect("saved_user mutex poisoned").is_none()); + } + #[tokio::test] async fn test_init_keeps_error_state_when_initial_load_fails() { let (sender, receiver) = mpsc::channel::(1); diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index a46a6312b..76e0dfde3 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -16,6 +16,7 @@ use crate::error::Error as IamError; use crate::error::is_err_no_such_account; use crate::error::is_err_no_such_temp_account; use crate::error::{Error, Result}; +use crate::federation::OIDC_VIRTUAL_PARENT_CLAIM; use crate::manager::extract_jwt_claims; use crate::manager::get_default_policyes; use crate::manager::{IamCache, IamSyncMetricsSnapshot}; @@ -25,8 +26,8 @@ use crate::store::Store; use crate::store::UserType; use crate::utils::{extract_claims, extract_claims_allow_missing_exp}; use crate::{ - notify_iam_delete_policy, notify_iam_delete_service_account, notify_iam_delete_user, notify_iam_load_group, - notify_iam_load_policy, notify_iam_load_policy_mapping, notify_iam_load_service_account, notify_iam_load_user, + notify_iam_delete_policy, notify_iam_delete_user, notify_iam_load_group, notify_iam_load_policy, + notify_iam_load_policy_mapping, notify_iam_load_service_account, notify_iam_load_user, }; use rustfs_credentials::{Credentials, EMBEDDED_POLICY_TYPE, INHERITED_POLICY_TYPE}; use rustfs_madmin::AddOrUpdateUserReq; @@ -56,6 +57,8 @@ pub const STATUS_DISABLED: &str = "disabled"; pub const POLICYNAME: &str = "policy"; pub const SESSION_POLICY_NAME: &str = "sessionPolicy"; pub const SESSION_POLICY_NAME_EXTRACTED: &str = "sessionPolicy-extracted"; +const VERIFIED_FEDERATED_POLICY_CLAIM: &str = "x-rustfs-internal-federated-policy"; +const FEDERATED_POLICY_BOUNDARY_CLAIM: &str = "x-rustfs-internal-federated-boundary"; pub(crate) const SITE_REPLICATOR_CLAIM: &str = "site-replicator"; static POLICY_PLUGIN_CLIENT: OnceLock>>> = OnceLock::new(); @@ -100,6 +103,7 @@ enum PreparedIamMode { bypass_parent_policy: bool, parent_user: String, combined_policy: Policy, + federated_boundary: Option, mode: PreparedServicePolicyMode, session_policy: PreparedSessionPolicy, }, @@ -130,11 +134,16 @@ impl PreparedIamAuth { } PreparedIamMode::ServiceAccount { combined_policy, + federated_boundary, mode, session_policy, .. } => { policy_needs_existing_object_tag_for_args(combined_policy, args).await + || match federated_boundary { + Some(policy) => policy_needs_existing_object_tag_for_args(policy, args).await, + None => false, + } || matches!(mode, PreparedServicePolicyMode::SessionBound) && prepared_session_policy_needs_existing_object_tag_for_args(session_policy, args).await } @@ -685,25 +694,30 @@ impl IamSys { } pub async fn delete_service_account(&self, access_key: &str, notify: bool) -> Result<()> { + self.delete_service_account_with_revision(access_key, notify).await?; + Ok(()) + } + + pub async fn delete_service_account_with_revision(&self, access_key: &str, notify: bool) -> Result> { let Some(u) = self.store.get_user(access_key).await else { - return Ok(()); + return Ok(None); }; if !u.credentials.is_service_account() { - return Ok(()); + return Ok(None); } - self.store.delete_user(access_key, UserType::Svc).await?; + let deleted_at = self.store.delete_service_account_with_revision(access_key).await?; if notify && !self.has_watcher() { - for r in notify_iam_delete_service_account(access_key).await { - if let Some(err) = r.err { - warn!("notify delete_service_account failed: {}", err); + for result in notify_iam_load_service_account(access_key).await { + if let Some(err) = result.err { + warn!("notify deleted service account failed: {}", err); } } } - Ok(()) + Ok(Some(deleted_at)) } async fn notify_for_group(&self, group: &str) { @@ -870,15 +884,7 @@ impl IamSys { /// - No characters outside the allowed set (helps mitigate path traversal /// and injection when names are used in storage keys or log output). fn is_safe_claim_policy_name(policy: &str) -> bool { - let mut has_alnum = false; - for c in policy.bytes() { - if c.is_ascii_alphanumeric() { - has_alnum = true; - } else if !matches!(c, b'_' | b'-' | b':' | b'.') { - return false; - } - } - has_alnum + is_safe_claim_policy_name(policy) } // JWT policy claims carry canned policy names only; policy documents are resolved by IAM store. @@ -907,6 +913,118 @@ impl IamSys { .collect() } + async fn verified_federated_service_account_claims( + &self, + claims: &HashMap, + parent_user: &str, + groups: &Option>, + ) -> Result>> { + let Some(policy_names) = self.verified_federated_policy_names(claims, parent_user, groups).await? else { + return Ok(None); + }; + + let (resolved, _) = self.store.merge_policies(&policy_names.join(",")).await; + let resolved_names = MappedPolicy::new(&resolved).to_slice(); + if policy_names.iter().any(|policy_name| !resolved_names.contains(policy_name)) { + return Err(IamError::InvalidArgument); + } + + let boundary = match claims.get(SESSION_POLICY_NAME) { + Some(Value::String(encoded)) => { + decode_session_policy(encoded)?; + Some(encoded.clone()) + } + Some(_) => return Err(IamError::InvalidArgument), + None => None, + }; + + let mut verified_claims = HashMap::from([ + (VERIFIED_FEDERATED_POLICY_CLAIM.to_string(), Value::Bool(true)), + (POLICYNAME.to_string(), Value::String(resolved)), + ]); + if claims.get(OIDC_VIRTUAL_PARENT_CLAIM).and_then(Value::as_str) == Some(parent_user) { + verified_claims.insert(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.to_string())); + } + if let Some(boundary) = boundary { + verified_claims.insert(FEDERATED_POLICY_BOUNDARY_CLAIM.to_string(), Value::String(boundary)); + } + Ok(Some(verified_claims)) + } + + pub async fn new_service_account_from_caller( + &self, + parent_user: &str, + groups: Option>, + mut opts: NewServiceAccountOpts, + caller: &Credentials, + ) -> Result<(Credentials, OffsetDateTime, HashMap)> { + if parent_user != caller.access_key && parent_user != caller.parent_user { + return Err(IamError::IAMActionNotAllowed); + } + + let verified_federated_policy = if caller.is_service_account() { + None + } else { + self.verified_federated_service_account_claims(caller.claims_or_empty(), &caller.parent_user, &caller.groups) + .await + .map_err(|_| IamError::IAMActionNotAllowed)? + }; + if let Some(source_claims) = caller.claims.as_ref() { + merge_derived_service_account_claims(opts.claims.get_or_insert_default(), source_claims, verified_federated_policy)?; + } + + let replication_claims = opts.claims.clone().unwrap_or_default(); + let (credentials, updated_at) = self.new_service_account(parent_user, groups, opts).await?; + Ok((credentials, updated_at, replication_claims)) + } + + async fn oidc_policy_names( + &self, + claims: &HashMap, + parent_user: &str, + groups: &Option>, + ) -> Result> { + let mapped_policy_names = self.policy_db_get(parent_user, groups).await?; + let policy_names = if mapped_policy_names.is_empty() { + Self::safe_claim_policy_names(claims, parent_user) + } else { + mapped_policy_names + }; + if policy_names.is_empty() { + return Err(IamError::InvalidArgument); + } + Ok(policy_names) + } + + fn signed_oidc_policy_names(claims: &HashMap, parent_user: &str) -> Result> { + let policy_names = Self::safe_claim_policy_names(claims, parent_user); + if policy_names.is_empty() { + return Err(IamError::InvalidArgument); + } + Ok(policy_names) + } + + async fn verified_federated_policy_names( + &self, + claims: &HashMap, + parent_user: &str, + groups: &Option>, + ) -> Result>> { + if !is_rustfs_oidc_claims(claims) { + return Ok(None); + } + if !is_verified_or_legacy_federated_session(claims, parent_user) { + return Err(IamError::InvalidArgument); + } + + let policy_names = if is_verified_federated_session(claims, parent_user) { + Self::signed_oidc_policy_names(claims, parent_user)? + } else { + self.oidc_policy_names(claims, parent_user, groups).await? + }; + Ok(Some(policy_names)) + } + /// Compatibility wrapper for service-account authorization entry points. /// The canonical evaluation path is `prepare_service_account_auth + eval_prepared`. pub async fn is_allowed_service_account(&self, args: &Args<'_>, parent_user: &str) -> bool { @@ -987,13 +1105,22 @@ impl IamSys { bypass_parent_policy, parent_user, combined_policy, + federated_boundary, mode, session_policy, } => { let mut parent_args = args.clone(); parent_args.account = parent_user; + let boundary_allowed = if let Some(policy) = federated_boundary { + let mut boundary_args = args.clone(); + boundary_args.is_owner = false; + policy.is_allowed(&boundary_args).await + } else { + true + }; - let parent_allowed = *bypass_parent_policy || *is_owner || combined_policy.is_allowed(&parent_args).await; + let parent_allowed = + (*bypass_parent_policy || *is_owner || combined_policy.is_allowed(&parent_args).await) && boundary_allowed; match mode { PreparedServicePolicyMode::Inherited => parent_allowed, PreparedServicePolicyMode::SessionBound => { @@ -1031,10 +1158,26 @@ impl IamSys { } pub(crate) async fn prepare_sts_auth(&self, args: &Args<'_>, parent_user: &str) -> PreparedIamAuth { - let is_owner = crate::is_root_access_key(parent_user); + let federated_policy_names = match self + .verified_federated_policy_names(args.claims, parent_user, args.groups) + .await + { + Ok(policy_names) => policy_names, + Err(_) => { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Deny, + }; + } + }; + let federated_session = federated_policy_names.is_some(); + let verified_federated_session = is_verified_federated_session(args.claims, parent_user); + let is_owner = !is_rustfs_oidc_claims(args.claims) && crate::is_root_access_key(parent_user); let role_arn = args.get_role_arn(); - let (effective_groups, groups_source, policies) = if is_owner { + let (effective_groups, groups_source, policies) = if federated_session { + (args.groups.clone(), "oidc_claims", Vec::new()) + } else if is_owner { (None, "owner", Vec::new()) } else if let Some(arn_str) = role_arn { let Ok(arn) = ARN::parse(arn_str) else { @@ -1068,8 +1211,12 @@ impl IamSys { (effective_groups, groups_source, p) }; - let mut policy_names = policies; - if !is_owner && policy_names.is_empty() { + let mut policy_names = if let Some(policy_names) = federated_policy_names { + policy_names + } else { + policies + }; + if !is_owner && policy_names.is_empty() && !is_rustfs_oidc_claims(args.claims) { policy_names = Self::safe_claim_policy_names(args.claims, parent_user); } @@ -1093,7 +1240,7 @@ impl IamSys { requested_policies = %requested_policies, "prepare_sts_auth: no STS policy names resolved" ); - if args.deny_only { + if args.deny_only && !verified_federated_session { combined_policy = Policy::default(); } else { return PreparedIamAuth { @@ -1107,6 +1254,12 @@ impl IamSys { .iter() .any(|policy_name| !resolved_policy_names.iter().any(|resolved| resolved == policy_name)); if has_unresolved_policy_names { + if verified_federated_session { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Deny, + }; + } tracing::debug!( parent_user = %parent_user, requested_policies = %requested_policies, @@ -1174,6 +1327,32 @@ impl IamSys { }; let session_policy = prepare_session_policy(args, true); + let has_verified_federated_policy = args.claims.contains_key(VERIFIED_FEDERATED_POLICY_CLAIM); + if is_rustfs_oidc_claims(args.claims) && !has_verified_federated_policy { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Deny, + }; + } + let federated_boundary = if has_verified_federated_policy { + if !is_valid_verified_federated_policy(args.claims, parent_user) { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Deny, + }; + } + match federated_policy_boundary(args.claims) { + Ok(boundary) => boundary, + Err(_) => { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Deny, + }; + } + } + } else { + None + }; let bypass_parent_policy = args.account == SITE_REPLICATOR_SERVICE_ACCOUNT && args .claims @@ -1183,11 +1362,13 @@ impl IamSys { && matches!(mode, PreparedServicePolicyMode::SessionBound) && matches!(session_policy, PreparedSessionPolicy::Policy(_)); - let is_owner = crate::is_root_access_key(parent_user); + let is_owner = !is_rustfs_oidc_claims(args.claims) && crate::is_root_access_key(parent_user); let role_arn = args.get_role_arn(); let svc_policies = if is_owner || bypass_parent_policy { Vec::new() + } else if has_verified_federated_policy { + Self::safe_claim_policy_names(args.claims, parent_user) } else if role_arn.is_some() { let Ok(arn) = ARN::parse(role_arn.unwrap_or_default()) else { tracing::warn!( @@ -1222,7 +1403,10 @@ impl IamSys { Policy::default() } else { let (a, c) = self.store.merge_policies(&svc_policies.join(",")).await; - if a.is_empty() { + let resolved = MappedPolicy::new(&a).to_slice(); + if a.is_empty() + || has_verified_federated_policy && svc_policies.iter().any(|policy_name| !resolved.contains(policy_name)) + { return PreparedIamAuth { needs_existing_object_tag: false, mode: PreparedIamMode::Deny, @@ -1231,6 +1415,10 @@ impl IamSys { c }; let needs_existing_object_tag = policy_needs_existing_object_tag_for_args(&combined_policy, args).await + || match &federated_boundary { + Some(policy) => policy_needs_existing_object_tag_for_args(policy, args).await, + None => false, + } || matches!(mode, PreparedServicePolicyMode::SessionBound) && prepared_session_policy_needs_existing_object_tag_for_args(&session_policy, args).await; @@ -1241,6 +1429,7 @@ impl IamSys { bypass_parent_policy, parent_user: parent_user.to_string(), combined_policy, + federated_boundary, mode, session_policy, }, @@ -1288,6 +1477,118 @@ fn prepare_session_policy(args: &Args<'_>, empty_is_none: bool) -> PreparedSessi PreparedSessionPolicy::Policy(sub_policy) } +fn decode_session_policy(encoded: &str) -> Result { + let bytes = base64_simd::URL_SAFE_NO_PAD.decode_to_vec(encoded.as_bytes())?; + if bytes.len() > MAX_SVCSESSION_POLICY_SIZE { + return Err(IamError::PolicyTooLarge); + } + let policy = Policy::parse_config(&bytes)?; + policy.validate()?; + if policy.version.is_empty() { + return Err(IamError::InvalidArgument); + } + Ok(policy) +} + +pub fn is_safe_claim_policy_name(policy: &str) -> bool { + let mut has_alnum = false; + for c in policy.bytes() { + if c.is_ascii_alphanumeric() { + has_alnum = true; + } else if !matches!(c, b'_' | b'-' | b':' | b'.') { + return false; + } + } + has_alnum +} + +pub fn is_rustfs_oidc_claims(claims: &HashMap) -> bool { + claims.get("iss").and_then(Value::as_str) == Some("rustfs-oidc") + && claims + .get("oidc_provider") + .and_then(Value::as_str) + .is_some_and(|provider| !provider.is_empty()) + && claims + .get("sub") + .and_then(Value::as_str) + .is_some_and(|subject| !subject.is_empty()) +} + +fn has_verified_federated_policy(claims: &HashMap) -> bool { + claims + .get(VERIFIED_FEDERATED_POLICY_CLAIM) + .and_then(Value::as_bool) + .unwrap_or(false) +} + +pub fn remove_verified_federated_policy(claims: &mut HashMap) -> bool { + let removed = claims.remove(VERIFIED_FEDERATED_POLICY_CLAIM).is_some(); + claims.remove(FEDERATED_POLICY_BOUNDARY_CLAIM); + claims.remove(OIDC_VIRTUAL_PARENT_CLAIM); + removed +} + +fn merge_derived_service_account_claims( + target_claims: &mut HashMap, + source_claims: &HashMap, + verified_federated_policy: Option>, +) -> Result<()> { + for (key, value) in source_claims { + if key != "exp" { + target_claims.insert(key.clone(), value.clone()); + } + } + let inherited_verified_policy = remove_verified_federated_policy(target_claims); + let Some(verified_policy) = verified_federated_policy else { + return if inherited_verified_policy { + Err(IamError::IAMActionNotAllowed) + } else { + Ok(()) + }; + }; + target_claims.remove(SESSION_POLICY_NAME); + target_claims.remove(SESSION_POLICY_NAME_EXTRACTED); + target_claims.extend(verified_policy); + Ok(()) +} + +fn is_verified_federated_session(claims: &HashMap, parent_user: &str) -> bool { + !parent_user.is_empty() + && is_rustfs_oidc_claims(claims) + && claims.get("parent").and_then(Value::as_str) == Some(parent_user) + && claims.get(OIDC_VIRTUAL_PARENT_CLAIM).and_then(Value::as_str) == Some(parent_user) +} + +fn is_verified_or_legacy_federated_session(claims: &HashMap, parent_user: &str) -> bool { + is_verified_federated_session(claims, parent_user) + || (!claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) + && !parent_user.is_empty() + && is_rustfs_oidc_claims(claims) + && claims.get("parent").and_then(Value::as_str) == Some(parent_user)) +} + +fn is_valid_verified_federated_policy(claims: &HashMap, parent_user: &str) -> bool { + has_verified_federated_policy(claims) + && is_verified_or_legacy_federated_session(claims, parent_user) + && claims.get(POLICYNAME).and_then(Value::as_str).is_some_and(|policies| { + policies + .split(',') + .map(str::trim) + .all(|policy| !policy.is_empty() && is_safe_claim_policy_name(policy)) + }) +} + +fn federated_policy_boundary(claims: &HashMap) -> Result> { + let Some(boundary) = claims.get(FEDERATED_POLICY_BOUNDARY_CLAIM) else { + return Ok(None); + }; + boundary + .as_str() + .ok_or(IamError::InvalidArgument) + .and_then(decode_session_policy) + .map(Some) +} + fn extract_session_policy_text(claims: &HashMap) -> Option { if let Some(policy_str) = claims.get(SESSION_POLICY_NAME_EXTRACTED).and_then(|v| v.as_str()) { return Some(policy_str.to_string()); @@ -1487,8 +1788,12 @@ mod tests { Ok(()) } - async fn delete_user_identity(&self, _name: &str, _user_type: UserType) -> Result<()> { - Err(Error::InvalidArgument) + async fn delete_user_identity(&self, name: &str, _user_type: UserType) -> Result<()> { + self.saved_sts_users + .lock() + .expect("saved_sts_users mutex poisoned") + .remove(name); + Ok(()) } async fn load_user_identity(&self, name: &str, _user_type: UserType) -> Result { @@ -1730,6 +2035,87 @@ mod tests { assert_eq!(err, Error::AccessKeyAlreadyExists); } + #[tokio::test] + async fn service_account_mutations_return_the_persisted_replication_state() { + ensure_test_global_credentials(); + + let iam_sys = test_iam_sys().await; + let access_key = "REPLICATEDSTATE123"; + iam_sys + .new_service_account("parent-user", None, service_account_opts(access_key, "replicatedStateSecret123")) + .await + .expect("create service account"); + + let session_policy = Policy::parse_config(CUSTOM_STS_CLAIM_POLICY_JSON.as_bytes()).expect("parse service-account policy"); + iam_sys + .update_service_account( + access_key, + UpdateServiceAccountOpts { + session_policy: Some(session_policy), + secret_key: None, + name: None, + description: None, + expiration: None, + status: None, + allow_site_replicator_account: false, + }, + ) + .await + .expect("update service account"); + + let (updated, session_policy) = iam_sys + .get_service_account(access_key) + .await + .expect("updated service account"); + assert!(session_policy.is_some_and(|policy| !policy.statements.is_empty())); + assert_eq!( + updated + .claims_or_empty() + .get(&iam_policy_claim_name_sa()) + .and_then(Value::as_str), + Some(EMBEDDED_POLICY_TYPE) + ); + + iam_sys + .update_service_account( + access_key, + UpdateServiceAccountOpts { + session_policy: Some(Policy::default()), + secret_key: None, + name: None, + description: None, + expiration: None, + status: None, + allow_site_replicator_account: false, + }, + ) + .await + .expect("clear service account session policy"); + let (cleared, session_policy) = iam_sys + .get_service_account(access_key) + .await + .expect("cleared service account"); + assert!(session_policy.is_none()); + assert_eq!( + cleared + .claims_or_empty() + .get(&iam_policy_claim_name_sa()) + .and_then(Value::as_str), + Some(INHERITED_POLICY_TYPE) + ); + + let deleted_at = iam_sys + .delete_service_account_with_revision(access_key, false) + .await + .expect("delete service account") + .expect("deleted revision"); + let (_, recreated_at) = iam_sys + .new_service_account("parent-user", None, service_account_opts(access_key, "recreatedStateSecret123")) + .await + .expect("recreate service account"); + assert!(deleted_at <= recreated_at); + } + #[tokio::test] async fn duplicate_service_account_returns_access_key_already_exists() { ensure_test_global_credentials(); @@ -2386,6 +2772,401 @@ mod tests { ); } + #[tokio::test] + async fn oidc_service_account_uses_verified_policy_and_persisted_boundary() { + ensure_test_global_credentials(); + let iam_sys = IamSys::new(IamCache::new(StsTestMockStore::new(true)).await.unwrap()); + let parent_user = "openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I"; + let mut oidc_claims = HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String(parent_user.to_string())), + (OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.to_string())), + (POLICYNAME.to_string(), Value::String("readwrite".to_string())), + ]); + oidc_claims.insert( + SESSION_POLICY_NAME.to_string(), + Value::String(base64_simd::URL_SAFE_NO_PAD.encode_to_string(CUSTOM_STS_CLAIM_POLICY_JSON.as_bytes())), + ); + oidc_claims.insert("exp".to_string(), Value::Number(123456.into())); + let caller = Credentials { + access_key: "OIDCTEMPORARYCALLER1".to_string(), + secret_key: "oidcTemporaryCallerSecret123".to_string(), + session_token: "signed-session-token".to_string(), + parent_user: parent_user.to_string(), + claims: Some(oidc_claims), + ..Default::default() + }; + + let (credential, _, replication_claims) = iam_sys + .new_service_account_from_caller( + parent_user, + None, + NewServiceAccountOpts { + access_key: "OIDCSERVICEACCOUNT01".to_string(), + secret_key: "oidcServiceAccountSecret123".to_string(), + ..Default::default() + }, + &caller, + ) + .await + .expect("OIDC service account should be created"); + assert_eq!(replication_claims.get(VERIFIED_FEDERATED_POLICY_CLAIM), Some(&Value::Bool(true))); + assert_eq!( + replication_claims.get(OIDC_VIRTUAL_PARENT_CLAIM), + Some(&Value::String(parent_user.to_string())) + ); + assert!(replication_claims.contains_key(FEDERATED_POLICY_BOUNDARY_CLAIM)); + assert!(!replication_claims.contains_key("exp")); + let claims = iam_sys + .get_claims_for_svc_acc(&credential.access_key) + .await + .expect("stored service-account claims should decode"); + let groups = None; + let conditions = HashMap::new(); + + for (action, allowed) in [(S3Action::GetObjectAction, true), (S3Action::PutObjectAction, false)] { + let args = Args { + account: &credential.access_key, + groups: &groups, + action: Action::S3Action(action), + bucket: CUSTOM_STS_CLAIM_BUCKET, + conditions: &conditions, + is_owner: false, + object: "allowed/object.txt", + claims: &claims, + deny_only: false, + }; + assert_eq!( + iam_sys.is_allowed(&args).await, + allowed, + "service-account permissions must be the OIDC policy intersected with its session boundary" + ); + } + + iam_sys.store.cache.delete_policy_doc("readwrite", OffsetDateTime::now_utc()); + let args = Args { + account: &credential.access_key, + groups: &groups, + action: Action::S3Action(S3Action::GetObjectAction), + bucket: CUSTOM_STS_CLAIM_BUCKET, + conditions: &conditions, + is_owner: false, + object: "allowed/object.txt", + claims: &claims, + deny_only: false, + }; + assert!( + !iam_sys.is_allowed(&args).await, + "OIDC service accounts must resolve current policy documents instead of retaining a policy snapshot" + ); + } + + #[tokio::test] + async fn oidc_group_policy_applies_to_sts_and_service_account() { + ensure_test_global_credentials(); + let iam_sys = IamSys::new(IamCache::new(StsTestMockStore::new(false)).await.unwrap()); + let parent_user = "openid=group-policy-parent"; + let groups = Some(vec!["testgroup".to_string()]); + let claims = HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String(parent_user.to_string())), + (OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.to_string())), + (POLICYNAME.to_string(), Value::String("readwrite".to_string())), + ]); + let conditions = HashMap::new(); + let sts_args = Args { + account: "OIDCGROUPCALLER01", + groups: &groups, + action: Action::S3Action(S3Action::GetObjectAction), + bucket: "bucket", + conditions: &conditions, + is_owner: false, + object: "object.txt", + claims: &claims, + deny_only: false, + }; + let prepared = iam_sys.prepare_sts_auth(&sts_args, parent_user).await; + assert!( + iam_sys.eval_prepared(&prepared, &sts_args).await, + "OIDC STS must include policies mapped from verified groups" + ); + + let caller = Credentials { + access_key: "OIDCGROUPCALLER01".to_string(), + parent_user: parent_user.to_string(), + groups: groups.clone(), + claims: Some(claims.clone()), + ..Default::default() + }; + let (service_account, _, replication_claims) = iam_sys + .new_service_account_from_caller( + parent_user, + groups.clone(), + NewServiceAccountOpts { + access_key: "OIDCGROUPSERVICE01".to_string(), + secret_key: "oidcGroupServiceSecret123".to_string(), + ..Default::default() + }, + &caller, + ) + .await + .expect("group-authorized OIDC caller should create a service account"); + assert_eq!(replication_claims.get(VERIFIED_FEDERATED_POLICY_CLAIM), Some(&Value::Bool(true))); + assert_eq!(replication_claims.get(POLICYNAME), Some(&Value::String("readwrite".to_string()))); + + let service_claims = iam_sys + .get_claims_for_svc_acc(&service_account.access_key) + .await + .expect("service-account claims"); + let service_args = Args { + account: &service_account.access_key, + groups: &groups, + claims: &service_claims, + ..sts_args + }; + assert!( + iam_sys.is_allowed(&service_args).await, + "OIDC service account must resolve the policy names selected at STS issuance" + ); + } + + #[tokio::test] + async fn verified_oidc_policy_names_do_not_drift_after_group_mapping_changes() { + let iam_sys = IamSys::new(IamCache::new(StsTestMockStore::new(false)).await.unwrap()); + let deny_delete = Policy::parse_config( + br#"{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":["s3:DeleteObject"],"Resource":["arn:aws:s3:::bucket/*"]}]}"#, + ) + .expect("parse deny policy"); + let now = OffsetDateTime::now_utc(); + iam_sys + .store + .cache + .add_or_update_policy_doc("deny-delete", &PolicyDoc::new(deny_delete), now); + iam_sys + .store + .cache + .add_or_update_group_policy("testgroup", &MappedPolicy::new("writeonly"), now); + + let parent_user = "openid=group-deny-parent"; + let claims = HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String(parent_user.to_string())), + (OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.to_string())), + (POLICYNAME.to_string(), Value::String("readonly,deny-delete".to_string())), + ]); + let groups = Some(vec!["testgroup".to_string()]); + let conditions = HashMap::new(); + + for (action, allowed) in [ + (S3Action::GetObjectAction, true), + (S3Action::PutObjectAction, false), + (S3Action::DeleteObjectAction, false), + ] { + let args = Args { + account: "OIDCGROUPCALLER02", + groups: &groups, + action: Action::S3Action(action), + bucket: "bucket", + conditions: &conditions, + is_owner: false, + object: "object.txt", + claims: &claims, + deny_only: false, + }; + let prepared = iam_sys.prepare_sts_auth(&args, parent_user).await; + assert_eq!( + iam_sys.eval_prepared(&prepared, &args).await, + allowed, + "verified OIDC sessions must retain the policy names selected at issuance" + ); + } + } + + #[tokio::test] + async fn active_legacy_oidc_session_can_create_service_account_after_upgrade() { + ensure_test_global_credentials(); + let iam_sys = IamSys::new(IamCache::new(StsTestMockStore::new(true)).await.unwrap()); + let parent_user = "legacy-oidc-user"; + let legacy_claims = HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String(parent_user.to_string())), + (POLICYNAME.to_string(), Value::String("readwrite".to_string())), + ]); + let caller = Credentials { + access_key: "LEGACYOIDCCALLER01".to_string(), + parent_user: parent_user.to_string(), + claims: Some(legacy_claims), + ..Default::default() + }; + + let (service_account, _, replication_claims) = iam_sys + .new_service_account_from_caller( + parent_user, + None, + NewServiceAccountOpts { + access_key: "LEGACYOIDCSERVICE01".to_string(), + secret_key: "legacyOidcServiceSecret123".to_string(), + ..Default::default() + }, + &caller, + ) + .await + .expect("active pre-upgrade OIDC session should remain usable"); + assert!(!replication_claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM)); + assert_eq!(replication_claims.get(POLICYNAME), Some(&Value::String("readwrite".to_string()))); + + let claims = iam_sys + .get_claims_for_svc_acc(&service_account.access_key) + .await + .expect("legacy-derived service-account claims"); + let groups = None; + let args = Args { + account: &service_account.access_key, + groups: &groups, + action: Action::S3Action(S3Action::GetObjectAction), + bucket: "bucket", + conditions: &HashMap::new(), + is_owner: false, + object: "object.txt", + claims: &claims, + deny_only: false, + }; + assert!(iam_sys.is_allowed(&args).await); + } + + #[tokio::test] + async fn non_owner_import_cannot_preserve_verified_federated_policy() { + ensure_test_global_credentials(); + + let iam_sys = test_iam_sys().await; + let parent_user = "sts-fallback-test-parent"; + iam_sys + .store + .cache + .add_or_update_user_policy(parent_user, &MappedPolicy::new("readwrite"), OffsetDateTime::now_utc()); + let mut claims = HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String(parent_user.to_string())), + (OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.to_string())), + (POLICYNAME.to_string(), Value::String("readwrite".to_string())), + (VERIFIED_FEDERATED_POLICY_CLAIM.to_string(), Value::Bool(true)), + ( + FEDERATED_POLICY_BOUNDARY_CLAIM.to_string(), + Value::String(base64_simd::URL_SAFE_NO_PAD.encode_to_string(CUSTOM_STS_CLAIM_POLICY_JSON.as_bytes())), + ), + ]); + + remove_verified_federated_policy(&mut claims); + let (credential, _) = iam_sys + .new_service_account( + parent_user, + None, + NewServiceAccountOpts { + access_key: "IMPORTEDOIDCSERVICE1".to_string(), + secret_key: "importedOidcServiceSecret123".to_string(), + claims: Some(claims), + ..Default::default() + }, + ) + .await + .expect("sanitized service account should be persisted"); + let stored_claims = iam_sys + .get_claims_for_svc_acc(&credential.access_key) + .await + .expect("stored service-account claims should decode"); + assert!(!stored_claims.contains_key(VERIFIED_FEDERATED_POLICY_CLAIM)); + assert!(!stored_claims.contains_key(FEDERATED_POLICY_BOUNDARY_CLAIM)); + assert!(!stored_claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM)); + + let args = Args { + account: &credential.access_key, + groups: &None, + action: Action::S3Action(S3Action::GetObjectAction), + bucket: CUSTOM_STS_CLAIM_BUCKET, + conditions: &HashMap::new(), + is_owner: false, + object: "allowed/object.txt", + claims: &stored_claims, + deny_only: false, + }; + assert!(!iam_sys.is_allowed(&args).await); + } + + #[tokio::test] + async fn oidc_service_account_rejects_mismatched_virtual_parent() { + ensure_test_global_credentials(); + let iam_sys = IamSys::new(IamCache::new(StsTestMockStore::new(true)).await.unwrap()); + let caller = Credentials { + access_key: "OIDCTEMPORARYCALLER2".to_string(), + parent_user: "openid=parent-b".to_string(), + claims: Some(HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String("openid=parent-b".to_string())), + (OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String("openid=parent-a".to_string())), + (POLICYNAME.to_string(), Value::String("readwrite".to_string())), + ])), + ..Default::default() + }; + + let result = iam_sys + .new_service_account_from_caller( + "openid=parent-b", + None, + NewServiceAccountOpts { + access_key: "OIDCSERVICEACCOUNT02".to_string(), + secret_key: "oidcServiceAccountSecret234".to_string(), + ..Default::default() + }, + &caller, + ) + .await; + + assert!(matches!(result, Err(IamError::IAMActionNotAllowed))); + } + + #[tokio::test] + async fn verified_oidc_sts_missing_policy_fails_closed_for_deny_only_checks() { + let iam_sys = IamSys::new(IamCache::new(StsTestMockStore::new(true)).await.unwrap()); + let parent_user = "openid=verified-parent"; + let claims = HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String(parent_user.to_string())), + (OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.to_string())), + (POLICYNAME.to_string(), Value::String("readwrite,missing-policy".to_string())), + ]); + let groups = None; + let args = Args { + account: "OIDCTEMPORARYCALLER3", + groups: &groups, + action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction), + bucket: "", + conditions: &HashMap::new(), + is_owner: false, + object: "", + claims: &claims, + deny_only: true, + }; + + let prepared = iam_sys.prepare_sts_auth(&args, parent_user).await; + + assert!(matches!(prepared.mode, PreparedIamMode::Deny)); + assert!(!iam_sys.eval_prepared(&prepared, &args).await); + } + #[tokio::test] async fn test_sts_claim_policy_ignores_unsafe_and_missing_policy_names() { let store = StsTestMockStore::new(true); @@ -3126,4 +3907,29 @@ mod tests { assert!(!IamSys::::is_safe_claim_policy_name("-_-")); assert!(!IamSys::::is_safe_claim_policy_name("-.:_")); } + + #[tokio::test] + async fn verified_federated_policy_rejects_malformed_session_boundary() { + let iam_sys = IamSys::new( + IamCache::new(StsTestMockStore::new(true)) + .await + .expect("IAM cache should initialize"), + ); + let parent_user = "virtual-parent"; + let claims = HashMap::from([ + ("iss".to_string(), Value::String("rustfs-oidc".to_string())), + ("oidc_provider".to_string(), Value::String("default".to_string())), + ("sub".to_string(), Value::String("subject-123".to_string())), + ("parent".to_string(), Value::String(parent_user.to_string())), + (OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.to_string())), + (POLICYNAME.to_string(), Value::String("readwrite".to_string())), + (SESSION_POLICY_NAME.to_string(), Value::Bool(true)), + ]); + + let result = iam_sys + .verified_federated_service_account_claims(&claims, parent_user, &None) + .await; + + assert!(matches!(result, Err(IamError::InvalidArgument))); + } } diff --git a/crates/madmin/src/site_replication.rs b/crates/madmin/src/site_replication.rs index 79f998325..e131d2ed2 100644 --- a/crates/madmin/src/site_replication.rs +++ b/crates/madmin/src/site_replication.rs @@ -298,6 +298,11 @@ pub struct SRSvcAccDelete { pub api_version: Option, } +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct SRSvcAccReplicationEnvelope { + pub version: u64, +} + #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct SRSvcAccChange { #[serde(rename = "crSvcAccCreate", skip_serializing_if = "Option::is_none")] @@ -306,6 +311,8 @@ pub struct SRSvcAccChange { pub update: Option, #[serde(rename = "crSvcAccDelete", skip_serializing_if = "Option::is_none")] pub delete: Option, + #[serde(rename = "oidcServiceAccountEnvelope", skip_serializing_if = "Option::is_none")] + pub oidc_service_account_envelope: Option, #[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")] pub api_version: Option, } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 8f2a50a99..3a5270698 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -212,6 +212,7 @@ metrics-util = { version = "0.20", features = ["debugging"] } opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] } rsa = { workspace = true } rcgen = { workspace = true } +rustfs-test-utils.workspace = true # diagnose_e2e fixtures (archives are generated in-test, never checked in) zip = { workspace = true } zstd = { workspace = true } diff --git a/rustfs/src/admin/handlers/service_account.rs b/rustfs/src/admin/handlers/service_account.rs index 31aebd08d..7919c5c66 100644 --- a/rustfs/src/admin/handlers/service_account.rs +++ b/rustfs/src/admin/handlers/service_account.rs @@ -14,7 +14,7 @@ use super::iam_error::iam_error_to_s3_error; use crate::admin::access_key_identity; -use crate::admin::handlers::site_replication::site_replication_iam_change_hook; +use crate::admin::handlers::site_replication::{encode_service_account_replication_policy, site_replication_iam_change_hook}; 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}; @@ -38,7 +38,7 @@ use rustfs_madmin::{ ServiceAccountInfo, TemporaryAccountInfoResp, UpdateServiceAccountReq, }; use rustfs_policy::policy::action::{Action, AdminAction}; -use rustfs_policy::policy::{Args, Policy}; +use rustfs_policy::policy::{Args, DEFAULT_VERSION, Policy}; use s3s::S3ErrorCode::InvalidRequest; use s3s::header::CONTENT_LENGTH; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; @@ -52,16 +52,38 @@ use url::form_urlencoded; const LOG_COMPONENT_ADMIN: &str = "admin"; const LOG_SUBSYSTEM_SERVICE_ACCOUNT: &str = "service_account"; const EVENT_ADMIN_SERVICE_ACCOUNT_STATE: &str = "admin_service_account_state"; +const EMPTY_EXPLICIT_SERVICE_ACCOUNT_POLICY: &str = + r#"{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::*"]}]}"#; -fn sr_session_policy_from_value(value: Option<&serde_json::Value>) -> S3Result { - let Some(value) = value else { +fn sr_session_policy_from_policy(policy: Option<&Policy>) -> S3Result { + let Some(policy) = policy else { return Ok(SRSessionPolicy::default()); }; - let raw = serde_json::to_string(value).map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?; + let raw = serde_json::to_string(policy).map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?; SRSessionPolicy::from_json(&raw).map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e)) } +fn normalize_service_account_policy(mut policy: Policy) -> S3Result { + if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty()) { + let id = policy.id; + policy = Policy::parse_config(EMPTY_EXPLICIT_SERVICE_ACCOUNT_POLICY.as_bytes()) + .map_err(|e| s3_error!(InternalError, "parse empty service account policy failed: {:?}", e))?; + policy.id = id; + } else if policy.version.is_empty() && !policy.statements.is_empty() { + policy.version = DEFAULT_VERSION.to_string(); + } + Ok(policy) +} + +fn parse_new_service_account_policy(policy: Option<&serde_json::Value>) -> S3Result> { + let Some(policy) = policy else { + return Ok(None); + }; + let policy = normalize_service_account_policy(parse_service_account_policy(policy)?)?; + Ok((!policy.id.is_empty() || !policy.version.is_empty() || !policy.statements.is_empty()).then_some(policy)) +} + fn compat_time_sentinel() -> OffsetDateTime { OffsetDateTime::UNIX_EPOCH } @@ -86,18 +108,6 @@ fn delete_service_account_success_status(path: &str) -> StatusCode { } } -fn merge_derived_service_account_claims( - target_claims: &mut HashMap, - source_claims: &HashMap, -) { - for (key, value) in source_claims { - if key == "exp" { - continue; - } - target_claims.insert(key.clone(), value.clone()); - } -} - fn is_service_account_owner_of(caller: &StoredCredentials, target_parent_user: &str) -> bool { let caller_parent = if caller.parent_user.is_empty() { caller.access_key.as_str() @@ -210,6 +220,27 @@ fn parse_update_service_account_policy(new_policy: Option) -> Ok(Some(parse_service_account_policy(&policy)?)) } +fn service_account_update_replication_change( + access_key: &str, + update: &UpdateServiceAccountReq, + session_policy: Option<&Policy>, +) -> S3Result { + Ok(SRSvcAccChange { + update: Some(SRSvcAccUpdate { + access_key: access_key.to_string(), + secret_key: update.new_secret_key.clone().unwrap_or_default(), + status: update.new_status.clone().unwrap_or_default(), + name: update.new_name.clone().unwrap_or_default(), + description: update.new_description.clone().unwrap_or_default(), + session_policy: sr_session_policy_from_policy(session_policy)?, + expiration: update.new_expiration, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }) +} + pub fn register_service_account_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( Method::POST, @@ -293,11 +324,12 @@ impl Operation for AddServiceAccount { create_req.validate().map_err(|e| S3Error::with_message(InvalidRequest, e))?; - let session_policy = if let Some(policy) = &create_req.policy { - Some(parse_service_account_policy(policy)?) - } else { - None - }; + let session_policy = parse_new_service_account_policy(create_req.policy.as_ref())?; + let replication_policy = session_policy + .as_ref() + .map(serde_json::to_string) + .transpose() + .map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?; let Some(sys_cred) = current_action_credentials() else { return Err(s3_error!(InvalidRequest, "get sys cred failed")); @@ -358,7 +390,6 @@ impl Operation for AddServiceAccount { } let is_svc_acc = target_user == req_user || target_user == req_parent_user; - // GHSA-5354: confine a non-owner caller to its own scope so it cannot mint // a root-parented (owner) service account. Evaluated on the original // `target_user` the caller submitted, before the derived-credential rewrite @@ -370,7 +401,7 @@ impl Operation for AddServiceAccount { } let mut target_groups = None; - let mut opts = NewServiceAccountOpts { + let opts = NewServiceAccountOpts { access_key: create_req.access_key, secret_key: create_req.secret_key, name: create_req.name, @@ -389,48 +420,44 @@ impl Operation for AddServiceAccount { } target_groups = req_groups; - - if let Some(claims) = cred.claims { - if opts.claims.is_none() { - opts.claims = Some(HashMap::new()); - } - - merge_derived_service_account_claims(opts.claims.as_mut().unwrap(), &claims); - } } - let replication_claims = opts.claims.clone().unwrap_or_default(); - let replication_policy = create_req - .policy - .as_ref() - .map(serde_json::to_string) - .transpose() - .map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?; let replication_groups = target_groups.clone().unwrap_or_default(); let replication_name = opts.name.clone().unwrap_or_default(); let replication_description = opts.description.clone().unwrap_or_default(); let replication_expiration = opts.expiration; - let (new_cred, _) = iam_store - .new_service_account(&target_user, target_groups, opts) - .await - .map_err(|e| { - debug!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, - event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE, - target_user = %target_user, - result = "create_failed", - error = ?e, - "admin service account state" - ); - match e { - rustfs_iam::error::Error::InvalidAccessKeyLength - | rustfs_iam::error::Error::InvalidSecretKeyLength - | rustfs_iam::error::Error::AccessKeyAlreadyExists => iam_error_to_s3_error(e), - err => s3_error!(InternalError, "create service account failed, e: {:?}", err), - } - })?; + 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 (new_cred, updated_at, replication_claims) = create_result.map_err(|e| { + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE, + target_user = %target_user, + result = "create_failed", + error = ?e, + "admin service account state" + ); + match e { + 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"), + err => s3_error!(InternalError, "create service account failed, e: {:?}", err), + } + })?; + let (replication_session_policy, oidc_service_account_envelope) = + encode_service_account_replication_policy(&replication_claims, replication_policy.as_deref())?; if let Err(err) = site_replication_iam_change_hook(SRIAMItem { r#type: "service-account".to_string(), @@ -441,22 +468,18 @@ impl Operation for AddServiceAccount { secret_key: new_cred.secret_key.clone(), groups: replication_groups, claims: replication_claims, - session_policy: replication_policy - .as_deref() - .map(SRSessionPolicy::from_json) - .transpose() - .map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))? - .unwrap_or_default(), + session_policy: replication_session_policy, status: String::new(), name: replication_name, description: replication_description, expiration: replication_expiration, api_version: Some(SITE_REPL_API_VERSION.to_string()), }), + oidc_service_account_envelope, api_version: Some(SITE_REPL_API_VERSION.to_string()), ..Default::default() }), - updated_at: Some(OffsetDateTime::now_utc()), + updated_at: Some(updated_at), api_version: Some(SITE_REPL_API_VERSION.to_string()), ..Default::default() }) @@ -620,6 +643,7 @@ impl Operation for UpdateServiceAccount { let new_policy = update_req.new_policy.clone(); let sp = parse_update_service_account_policy(new_policy.clone())?; + let svc_acc_change = service_account_update_replication_change(&access_key, &update_req, sp.as_ref())?; let opts = UpdateServiceAccountOpts { secret_key: new_secret_key.clone(), @@ -638,20 +662,7 @@ impl Operation for UpdateServiceAccount { if let Err(err) = site_replication_iam_change_hook(SRIAMItem { r#type: "service-account".to_string(), - svc_acc_change: Some(SRSvcAccChange { - update: Some(SRSvcAccUpdate { - access_key: access_key.clone(), - secret_key: new_secret_key.unwrap_or_default(), - status: new_status.unwrap_or_default(), - name: new_name.unwrap_or_default(), - description: new_description.unwrap_or_default(), - session_policy: sr_session_policy_from_value(new_policy.as_ref())?, - expiration: new_expiration, - api_version: Some(SITE_REPL_API_VERSION.to_string()), - }), - api_version: Some(SITE_REPL_API_VERSION.to_string()), - ..Default::default() - }), + svc_acc_change: Some(svc_acc_change), updated_at: Some(updated_at), api_version: Some(SITE_REPL_API_VERSION.to_string()), ..Default::default() @@ -1352,33 +1363,37 @@ impl Operation for DeleteServiceAccount { } } - iam_store.delete_service_account(&query.access_key, true).await.map_err(|e| { - debug!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, - event = "service_account_delete_failed", - access_key = %query.access_key, - error = ?e, - "Failed to delete service account" - ); - s3_error!(InternalError, "delete service account failed") - })?; + let deleted_at = iam_store + .delete_service_account_with_revision(&query.access_key, true) + .await + .map_err(|e| { + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = "service_account_delete_failed", + access_key = %query.access_key, + error = ?e, + "Failed to delete service account" + ); + s3_error!(InternalError, "delete service account failed") + })?; - if let Err(err) = site_replication_iam_change_hook(SRIAMItem { - r#type: "service-account".to_string(), - svc_acc_change: Some(SRSvcAccChange { - delete: Some(SRSvcAccDelete { - access_key: query.access_key.clone(), + if let Some(deleted_at) = deleted_at + && let Err(err) = site_replication_iam_change_hook(SRIAMItem { + r#type: "service-account".to_string(), + svc_acc_change: Some(SRSvcAccChange { + delete: Some(SRSvcAccDelete { + access_key: query.access_key.clone(), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() }), + updated_at: Some(deleted_at), api_version: Some(SITE_REPL_API_VERSION.to_string()), ..Default::default() - }), - updated_at: Some(OffsetDateTime::now_utc()), - api_version: Some(SITE_REPL_API_VERSION.to_string()), - ..Default::default() - }) - .await + }) + .await { warn!( component = LOG_COMPONENT_ADMIN, @@ -1713,6 +1728,81 @@ mod tests { assert!(policy.statements.is_empty()); } + #[test] + fn sparse_explicit_create_policy_is_normalized_before_persistence_and_replication() { + let id_only = parse_new_service_account_policy(Some(&json!({"ID": "deny-boundary", "Version": "", "Statement": []}))) + .expect("ID-only policy should normalize") + .expect("normalized explicit policy"); + assert_eq!(id_only.id.as_str(), "deny-boundary"); + assert_eq!(id_only.version, DEFAULT_VERSION); + assert!(!id_only.statements.is_empty()); + + let missing_version = parse_new_service_account_policy(Some(&json!({ + "Statement": [{ + "Effect": "Deny", + "Action": ["s3:GetObject"], + "Resource": ["arn:aws:s3:::bucket/*"] + }] + }))) + .expect("policy version should normalize") + .expect("normalized explicit policy"); + assert_eq!(missing_version.version, DEFAULT_VERSION); + } + + #[test] + fn explicit_empty_create_policy_is_canonicalized_as_inherited() { + let policy = parse_new_service_account_policy(Some(&json!({}))).expect("parse empty create policy"); + + assert!(policy.is_none()); + } + + #[test] + fn sparse_explicit_update_policy_preserves_existing_semantics() { + let id_only = parse_update_service_account_policy(Some(json!({"ID": "clear-boundary", "Version": "", "Statement": []}))) + .expect("parse ID-only update") + .expect("explicit update policy"); + assert_eq!(id_only.id.as_str(), "clear-boundary"); + assert!(id_only.version.is_empty()); + assert!(id_only.statements.is_empty()); + + let missing_version = parse_update_service_account_policy(Some(json!({ + "Statement": [{ + "Effect": "Deny", + "Action": ["s3:GetObject"], + "Resource": ["arn:aws:s3:::bucket/*"] + }] + }))) + .expect("parse versionless update") + .expect("explicit update policy"); + assert!(missing_version.version.is_empty()); + assert!(!missing_version.statements.is_empty()); + } + + #[test] + fn replication_update_stays_partial() { + let update = UpdateServiceAccountReq { + new_policy: Some(json!({})), + new_secret_key: None, + new_status: None, + new_name: None, + new_description: None, + new_expiration: None, + }; + + let session_policy = parse_update_service_account_policy(update.new_policy.clone()).expect("parse update policy"); + let change = service_account_update_replication_change("OIDCSERVICEACCOUNT01", &update, session_policy.as_ref()) + .expect("build replication update"); + + assert!(change.create.is_none()); + assert!(change.delete.is_none()); + let update = change.update.expect("partial update"); + let cleared: Policy = + serde_json::from_str(update.session_policy.as_str().expect("explicit policy clear")).expect("parse policy clear"); + assert!(cleared.id.is_empty()); + assert!(cleared.version.is_empty()); + assert!(cleared.statements.is_empty()); + } + #[test] fn parse_service_account_policy_reports_missing_resource() { let err = parse_service_account_policy(&json!({ @@ -1774,22 +1864,6 @@ mod tests { assert!(!is_service_account_owner_of(&foreign_user, "owner-user")); } - #[test] - fn merge_derived_service_account_claims_skips_only_expiration() { - let mut merged = HashMap::new(); - let source = HashMap::from([ - ("exp".to_string(), json!(123456)), - ("parent".to_string(), json!("owner-user")), - ("custom".to_string(), json!("value")), - ]); - - merge_derived_service_account_claims(&mut merged, &source); - - assert!(!merged.contains_key("exp")); - assert_eq!(merged.get("parent"), Some(&json!("owner-user"))); - assert_eq!(merged.get("custom"), Some(&json!("value"))); - } - #[test] fn fallback_access_key_info_allows_same_regular_user() { let requester = StoredCredentials { diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 3cd0d71a1..f5a00708a 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -57,6 +57,7 @@ use rustfs_config::{ MAX_ADMIN_REQUEST_BODY_SIZE, }; use rustfs_iam::error::is_err_no_such_service_account; +use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM; use rustfs_iam::store::{MappedPolicy, UserType}; use rustfs_iam::sys::{ NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret, @@ -67,8 +68,8 @@ use rustfs_madmin::{ ReplicateEditStatus, ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem, SRIAMPolicy, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation, SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, - SRResyncOpStatus, SRRetryStats, SRSiteSummary, SRStateEditReq, SRStateInfo, SRStatusInfo, SRUserStatsSummary, - SiteReplicationInfo, SyncStatus, WorkerStat, + SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRSiteSummary, SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccCreate, + SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat, }; use rustfs_policy::policy::{ Policy, @@ -106,7 +107,7 @@ use uuid::Uuid; const LOG_COMPONENT_ADMIN: &str = "admin"; const LOG_SUBSYSTEM_SITE_REPLICATION: &str = "site_replication"; const EVENT_ADMIN_SITE_REPLICATION_STATE: &str = "admin_site_replication_state"; - +const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2; const SITE_REPLICATION_STATE_PATH: &str = "config/site-replication/state.json"; const SITE_REPL_ADD_SUCCESS: &str = "Requested sites were configured for replication successfully."; const SITE_REPL_EDIT_SUCCESS: &str = "Requested site was updated successfully."; @@ -6273,6 +6274,103 @@ fn group_info_requires_upsert(update: &rustfs_madmin::GroupAddRemove) -> bool { !update.is_remove } +pub(crate) fn encode_service_account_replication_policy( + claims: &HashMap, + session_policy: Option<&str>, +) -> S3Result<(SRSessionPolicy, Option)> { + if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) { + return session_policy + .map(SRSessionPolicy::from_json) + .transpose() + .map(|policy| policy.unwrap_or_default()) + .map(|policy| (policy, None)) + .map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err)); + } + + let policy = match session_policy { + Some(policy) => serde_json::from_str::(policy) + .map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?, + None => Policy::default(), + }; + if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty()) + || policy.version.is_empty() && !policy.statements.is_empty() + { + return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized")); + } + let policy = serde_json::to_string(&policy) + .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?; + let policy = SRSessionPolicy::from_json(&policy) + .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?; + Ok(( + policy, + Some(rustfs_madmin::SRSvcAccReplicationEnvelope { + version: SERVICE_ACCOUNT_ENVELOPE_VERSION, + }), + )) +} + +#[derive(Debug)] +struct ReplicatedServiceAccountPolicy { + policy: Option, + is_envelope: bool, +} + +impl ReplicatedServiceAccountPolicy { + fn for_existing_account(self) -> Option { + if self.is_envelope { + Some(self.policy.unwrap_or_default()) + } else { + self.policy + } + } + + fn metadata_for_existing_account(&self, value: String) -> Option { + (self.is_envelope || !value.is_empty()).then_some(value) + } +} + +fn decode_service_account_replication_policy( + create: &SRSvcAccCreate, + envelope: Option<&rustfs_madmin::SRSvcAccReplicationEnvelope>, + incoming_updated_at: Option, + local_updated_at: Option, +) -> S3Result> { + if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) { + return Ok(None); + } + + let Some(envelope) = envelope else { + return Ok(Some(ReplicatedServiceAccountPolicy { + policy: create.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok()), + is_envelope: false, + })); + }; + if envelope.version != SERVICE_ACCOUNT_ENVELOPE_VERSION || !create.claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) { + return Err(s3_error!(InvalidRequest, "invalid service account replication envelope")); + } + + if incoming_updated_at.is_none() { + return Err(s3_error!(InvalidRequest, "service account replication envelope has no revision")); + } + let policy: Policy = serde_json::from_str( + create + .session_policy + .as_str() + .ok_or_else(|| s3_error!(InvalidRequest, "service account replication envelope has no session policy"))?, + ) + .map_err(|err| s3_error!(InvalidRequest, "invalid replicated service account session policy: {}", err))?; + if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty()) + || policy.version.is_empty() && !policy.statements.is_empty() + { + return Err(s3_error!(InvalidRequest, "replicated service account policy is not normalized")); + } + let policy = (!policy.id.is_empty() || !policy.version.is_empty() || !policy.statements.is_empty()).then_some(policy); + Ok(Some(ReplicatedServiceAccountPolicy { + policy, + is_envelope: true, + })) +} + async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { let Some(iam_sys) = current_iam_handle() else { return Err(s3_error!(InvalidRequest, "iam not init")); @@ -6339,6 +6437,8 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { .map(OffsetDateTime::from_unix_timestamp) .transpose() .map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?; + let groups = string_list_claim(&claims, "groups"); + let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping); let cred = rustfs_credentials::Credentials { access_key: sts_credential.access_key.clone(), secret_key: sts_credential.secret_key.clone(), @@ -6346,15 +6446,12 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { expiration, status: "on".to_string(), parent_user: sts_credential.parent_user.clone(), + groups, claims: Some(claims), ..Default::default() }; iam_sys - .set_temp_user( - &sts_credential.access_key, - &cred, - (!sts_credential.parent_policy_mapping.is_empty()).then_some(sts_credential.parent_policy_mapping.as_str()), - ) + .set_temp_user(&sts_credential.access_key, &cred, compatibility_policy) .await .map_err(ApiError::from)?; Ok(()) @@ -6393,16 +6490,31 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { let Some(change) = item.svc_acc_change else { return Err(s3_error!(InvalidRequest, "serviceAccountChange is required")); }; + let envelope = change.oidc_service_account_envelope; if let Some(create) = change.create { - if let Some(local) = iam_sys.get_user(&create.access_key).await - && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at) - { - return Ok(()); - } - let session_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT { - Some(site_replicator_service_account_policy()?) + let local_updated_at = iam_sys + .get_user(&create.access_key) + .await + .map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH)); + let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT { + if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) { + return Ok(()); + } + ReplicatedServiceAccountPolicy { + policy: Some(site_replicator_service_account_policy()?), + is_envelope: false, + } } else { - create.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok()) + let Some(replicated_policy) = decode_service_account_replication_policy( + &create, + envelope.as_ref(), + incoming_updated_at, + local_updated_at, + )? + else { + return Ok(()); + }; + replicated_policy }; match iam_sys.get_service_account(&create.access_key).await { Ok((existing, _)) => { @@ -6417,10 +6529,10 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { .update_service_account( &create.access_key, UpdateServiceAccountOpts { - session_policy, + name: replicated_policy.metadata_for_existing_account(create.name), + description: replicated_policy.metadata_for_existing_account(create.description), + session_policy: replicated_policy.for_existing_account(), secret_key: Some(create.secret_key), - name: (!create.name.is_empty()).then_some(create.name), - description: (!create.description.is_empty()).then_some(create.description), expiration: create.expiration, status: (!create.status.is_empty()).then_some(create.status), allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT, @@ -6435,7 +6547,7 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { &create.parent, Some(create.groups), NewServiceAccountOpts { - session_policy, + session_policy: replicated_policy.policy, access_key: create.access_key, secret_key: create.secret_key, name: (!create.name.is_empty()).then_some(create.name), @@ -6514,6 +6626,21 @@ fn claims_unix_timestamp(value: &Value) -> Option { } } +fn string_list_claim(claims: &HashMap, name: &str) -> Option> { + let values = claims.get(name)?.as_array()?; + let values: Vec = values + .iter() + .filter_map(Value::as_str) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned) + .collect(); + (!values.is_empty()).then_some(values) +} + +fn sts_replication_compatibility_policy<'a>(claims: &HashMap, parent_policy_mapping: &'a str) -> Option<&'a str> { + (!claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) && !parent_policy_mapping.is_empty()).then_some(parent_policy_mapping) +} + pub struct SiteReplicationAddHandler {} #[async_trait::async_trait] @@ -7923,6 +8050,278 @@ mod tests { use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; + #[test] + fn sts_replication_restores_groups_from_signed_claims() { + let claims = HashMap::from([("groups".to_string(), serde_json::json!(["devs", "auditors"]))]); + + assert_eq!( + string_list_claim(&claims, "groups"), + Some(vec!["devs".to_string(), "auditors".to_string()]) + ); + } + + #[test] + fn oidc_sts_replication_uses_signed_policy_instead_of_virtual_parent_mapping() { + let verified_claims = + HashMap::from([(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String("openid=parent".to_string()))]); + let legacy_claims = HashMap::new(); + + assert!(sts_replication_compatibility_policy(&verified_claims, "readonly").is_none()); + assert_eq!(sts_replication_compatibility_policy(&legacy_claims, "readonly"), Some("readonly")); + } + + #[test] + fn oidc_service_account_envelope_round_trips_actual_policy() { + let actual_policy = r#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::bucket/*"]}]}"#; + let updated_at = OffsetDateTime::UNIX_EPOCH; + let claims = + HashMap::from([(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String("openid=verified-parent".to_string()))]); + let (wire_policy, envelope) = + encode_service_account_replication_policy(&claims, Some(actual_policy)).expect("encode envelope"); + let create = SRSvcAccCreate { + parent: "openid=verified-parent".to_string(), + access_key: "OIDCREPLICATEDSERVICE".to_string(), + secret_key: "oidcReplicatedSecret123".to_string(), + groups: Vec::new(), + claims, + session_policy: wire_policy, + status: String::new(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }; + let old_receiver_policy: Policy = serde_json::from_str( + create + .session_policy + .as_str() + .expect("old receiver gets a standard session policy"), + ) + .expect("parse old receiver policy"); + assert_eq!( + serde_json::to_value(old_receiver_policy).expect("serialize old receiver policy"), + serde_json::from_str::(actual_policy).expect("parse expected policy") + ); + assert_eq!(envelope.as_ref().map(|envelope| envelope.version), Some(SERVICE_ACCOUNT_ENVELOPE_VERSION)); + assert_eq!(create.claims.len(), 1); + + let decoded = decode_service_account_replication_policy(&create, envelope.as_ref(), Some(updated_at), None) + .expect("decode envelope") + .expect("current envelope"); + assert!(decoded.is_envelope); + let restored = decoded.policy.expect("actual policy"); + + assert_eq!( + serde_json::to_value(restored).expect("serialize restored policy"), + serde_json::from_str::(actual_policy).expect("parse expected policy") + ); + } + + #[test] + fn oidc_service_account_envelope_clears_policy_on_existing_account() { + let updated_at = OffsetDateTime::UNIX_EPOCH; + let claims = + HashMap::from([(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String("openid=verified-parent".to_string()))]); + let (wire_policy, envelope) = + encode_service_account_replication_policy(&claims, None).expect("encode inherited envelope"); + let create = SRSvcAccCreate { + parent: "openid=verified-parent".to_string(), + access_key: "OIDCREPLICATEDSERVICE".to_string(), + secret_key: "oidcReplicatedSecret123".to_string(), + groups: Vec::new(), + claims, + session_policy: wire_policy, + status: String::new(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }; + let old_receiver_policy: Policy = serde_json::from_str( + create + .session_policy + .as_str() + .expect("old receiver gets an explicit empty policy"), + ) + .expect("parse old receiver policy"); + assert!(old_receiver_policy.version.is_empty()); + assert!(old_receiver_policy.statements.is_empty()); + + let decoded = decode_service_account_replication_policy(&create, envelope.as_ref(), Some(updated_at), None) + .expect("decode inherited envelope") + .expect("current envelope"); + + assert!(decoded.is_envelope); + assert!(decoded.policy.is_none()); + assert_eq!(decoded.metadata_for_existing_account(String::new()), Some(String::new())); + let update_policy = decoded + .for_existing_account() + .expect("existing account needs an explicit clear"); + assert!(update_policy.version.is_empty()); + assert!(update_policy.statements.is_empty()); + } + + #[test] + fn oidc_service_account_envelope_replays_normalized_empty_policy() { + let actual_policy = r#"{"ID":"deny-boundary","Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::*"]}]}"#; + let claims = + HashMap::from([(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String("openid=verified-parent".to_string()))]); + let (wire_policy, envelope) = + encode_service_account_replication_policy(&claims, Some(actual_policy)).expect("encode envelope"); + let create = SRSvcAccCreate { + parent: "openid=verified-parent".to_string(), + access_key: "OIDCREPLICATEDSERVICE".to_string(), + secret_key: "oidcReplicatedSecret123".to_string(), + groups: Vec::new(), + claims, + session_policy: wire_policy, + status: "on".to_string(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }; + + let decoded = + decode_service_account_replication_policy(&create, envelope.as_ref(), Some(OffsetDateTime::UNIX_EPOCH), None) + .expect("decode normalized empty policy") + .expect("current envelope"); + let restored = decoded.policy.as_ref().expect("normalized policy must remain explicit"); + assert_eq!( + serde_json::to_value(restored).expect("serialize restored policy"), + serde_json::from_str::(actual_policy).expect("parse expected policy") + ); + assert!(decoded.for_existing_account().is_some()); + } + + #[test] + fn oidc_service_account_envelope_rejects_missing_policy() { + let create = SRSvcAccCreate { + parent: "openid=verified-parent".to_string(), + access_key: "OIDCREPLICATEDSERVICE".to_string(), + secret_key: "oidcReplicatedSecret123".to_string(), + groups: Vec::new(), + claims: HashMap::from([(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String("openid=verified-parent".to_string()))]), + session_policy: SRSessionPolicy::default(), + status: String::new(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }; + + let envelope = rustfs_madmin::SRSvcAccReplicationEnvelope { + version: SERVICE_ACCOUNT_ENVELOPE_VERSION, + }; + let err = decode_service_account_replication_policy(&create, Some(&envelope), Some(OffsetDateTime::UNIX_EPOCH), None) + .expect_err("policy-less envelope must fail closed"); + + assert_eq!(*err.code(), S3ErrorCode::InvalidRequest); + } + + #[test] + fn stale_oidc_service_account_envelope_is_ignored_before_decoding() { + let create = SRSvcAccCreate { + parent: "openid=verified-parent".to_string(), + access_key: "OIDCREPLICATEDSERVICE".to_string(), + secret_key: "oidcReplicatedSecret123".to_string(), + groups: Vec::new(), + claims: HashMap::new(), + session_policy: SRSessionPolicy::default(), + status: String::new(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }; + + let envelope = rustfs_madmin::SRSvcAccReplicationEnvelope { + version: SERVICE_ACCOUNT_ENVELOPE_VERSION + 1, + }; + let decoded = decode_service_account_replication_policy( + &create, + Some(&envelope), + Some(OffsetDateTime::UNIX_EPOCH), + Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)), + ) + .expect("stale envelope must be ignored before validation"); + + assert!(decoded.is_none()); + } + + #[test] + fn oidc_service_account_envelope_does_not_survive_a_legacy_hop() { + #[derive(serde::Deserialize, serde::Serialize)] + struct LegacyServiceAccountChange { + #[serde(rename = "crSvcAccCreate", skip_serializing_if = "Option::is_none")] + create: Option, + #[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")] + api_version: Option, + } + + let claims = + HashMap::from([(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String("openid=verified-parent".to_string()))]); + let (session_policy, envelope) = + encode_service_account_replication_policy(&claims, None).expect("encode envelope for legacy hop"); + let change = rustfs_madmin::SRSvcAccChange { + create: Some(SRSvcAccCreate { + parent: "openid=verified-parent".to_string(), + access_key: "OIDCREPLICATEDSERVICE".to_string(), + secret_key: "oidcReplicatedSecret123".to_string(), + groups: Vec::new(), + claims, + session_policy, + status: String::new(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), + oidc_service_account_envelope: envelope, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }; + + let legacy: LegacyServiceAccountChange = + serde_json::from_value(serde_json::to_value(change).expect("serialize new replication payload")) + .expect("legacy node must ignore the unknown envelope field"); + let legacy_claims = legacy + .create + .as_ref() + .expect("legacy payload has a create operation") + .claims + .clone(); + assert_eq!(legacy_claims.len(), 1); + + let reemitted: rustfs_madmin::SRSvcAccChange = serde_json::from_value( + serde_json::to_value(LegacyServiceAccountChange { + create: Some(SRSvcAccCreate { + parent: "openid=verified-parent".to_string(), + access_key: "OIDCLEGACYCHILD001".to_string(), + secret_key: "oidcLegacyChildSecret123".to_string(), + groups: Vec::new(), + claims: legacy_claims, + session_policy: SRSessionPolicy::default(), + status: String::new(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }) + .expect("serialize legacy child replication payload"), + ) + .expect("new node accepts legacy child replication payload"); + + assert!(reemitted.oidc_service_account_envelope.is_none()); + let create = reemitted.create.expect("reemitted payload has a create operation"); + let decoded = decode_service_account_replication_policy(&create, None, Some(OffsetDateTime::UNIX_EPOCH), None) + .expect("legacy payload must not be parsed as an envelope") + .expect("legacy payload should be accepted"); + assert!(!decoded.is_envelope); + } + fn valid_test_ca_pem(name: &str) -> String { rcgen::generate_simple_self_signed(vec![name.to_string()]) .expect("generate test CA") diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs index ff88602d5..d6b86e08f 100644 --- a/rustfs/src/admin/handlers/user.rs +++ b/rustfs/src/admin/handlers/user.rs @@ -1078,7 +1078,7 @@ impl Operation for ImportIam { if let Some(file_content) = file_content { let svc_accts: HashMap = serde_json::from_slice(&file_content) .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?; - for (ak, req) in svc_accts { + for (ak, mut req) in svc_accts { if skipped.service_accounts.contains(&ak) { continue; } @@ -1123,6 +1123,9 @@ impl Operation for ImportIam { })?; } + if !owner { + let _ = rustfs_iam::sys::remove_verified_federated_policy(&mut req.claims); + } let opts = NewServiceAccountOpts { session_policy: sp, access_key: ak.clone(), diff --git a/rustfs/src/admin/service/federated_identity.rs b/rustfs/src/admin/service/federated_identity.rs index 21e2e08ae..36f059cf4 100644 --- a/rustfs/src/admin/service/federated_identity.rs +++ b/rustfs/src/admin/service/federated_identity.rs @@ -18,9 +18,9 @@ use crate::admin::{ runtime_sources::{current_action_credentials, current_ready_iam_handle, current_token_signing_key}, }; use rustfs_iam::{ - federation::{FederatedSessionBinding, FederatedSessionBindingError, FederatedSessionTransaction}, - store::object::ObjectStore, - sys::IamSys, + federation::{FederatedSessionBinding, FederatedSessionBindingError, FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM}, + store::{MappedPolicy, object::ObjectStore}, + sys::{IamSys, is_safe_claim_policy_name}, }; use rustfs_madmin::{SITE_REPL_API_VERSION, SRIAMItem, SRSTSCredential}; use rustfs_policy::auth::get_new_credentials_with_metadata; @@ -32,6 +32,17 @@ use tracing::{debug, warn}; pub(crate) struct DefaultFederatedSessionBinding; +// Legacy receivers reject an empty parsed mapping; current receivers ignore it when the virtual-parent marker is present. +const OIDC_STS_REQUIRES_VIRTUAL_PARENT_RECEIVER_POLICY: &str = " "; + +fn all_oidc_policies_resolved(selected_policy_names: &[String], resolved_policy_mapping: &str) -> bool { + let resolved_policy_names = MappedPolicy::new(resolved_policy_mapping).to_slice(); + !selected_policy_names.is_empty() + && selected_policy_names + .iter() + .all(|policy_name| is_safe_claim_policy_name(policy_name) && resolved_policy_names.contains(policy_name)) +} + fn build_oidc_token_claims(transaction: &FederatedSessionTransaction) -> HashMap { let authorization = &transaction.authorization; let claims = &authorization.claims; @@ -168,22 +179,23 @@ fn parent_user_is_reserved(parent_user: &str, root_access_key: Option<&str>) -> fn issue_credentials( transaction: &FederatedSessionTransaction, + selected_policy_names: &[String], secret: Option<&str>, ) -> Result { let authorization = &transaction.authorization; let claims = &authorization.claims; + let parent_user = authorization.oidc_virtual_parent().ok_or_else(|| { + FederatedSessionBindingError::InvalidRequest("verified OIDC identity is missing issuer or subject".to_string()) + })?; let mut token_claims = build_oidc_token_claims(transaction); let duration = i64::try_from(transaction.duration_seconds) .map_err(|_| FederatedSessionBindingError::InvalidRequest("invalid duration".to_string()))?; let exp = OffsetDateTime::now_utc().saturating_add(Duration::seconds(duration)); token_claims.insert("exp".to_string(), Value::Number(serde_json::Number::from(exp.unix_timestamp()))); - let parent_user = claims.session_identity(); // Fail closed if the derived federated parent collides with the root access key. At IAM - // request time `parent_user == root access key` is treated as owner (see auth.rs and - // rustfs_iam owner resolution), so issuing such a credential would silently grant a - // federated identity full owner access purely because its display name matched root. - // Deny before any credential generation, `set_temp_user`, or site replication. + // request time `parent_user == root access key` is treated as owner, so issuing such a + // credential would silently grant a federated identity full owner access. let root_access_key = current_action_credentials().map(|cred| cred.access_key); if parent_user_is_reserved(&parent_user, root_access_key.as_deref()) { return Err(FederatedSessionBindingError::InvalidRequest( @@ -205,9 +217,12 @@ fn issue_credentials( "OIDC STS credential claims prepared" ); token_claims.insert("parent".to_string(), Value::String(parent_user.clone())); + token_claims.insert(OIDC_VIRTUAL_PARENT_CLAIM.to_string(), Value::String(parent_user.clone())); - if !authorization.policies.is_empty() { - token_claims.insert("policy".to_string(), Value::String(authorization.policies.join(","))); + if !selected_policy_names.is_empty() { + token_claims.insert("policy".to_string(), Value::String(selected_policy_names.join(","))); + } else { + token_claims.remove("policy"); } if let Some(policy) = transaction.session_policy.as_deref() { populate_session_policy(&mut token_claims, policy).map_err(binding_error_from_s3)?; @@ -221,11 +236,7 @@ fn issue_credentials( Ok(credentials) } -fn site_replication_item( - credentials: &rustfs_credentials::Credentials, - transaction: &FederatedSessionTransaction, - updated_at: OffsetDateTime, -) -> SRIAMItem { +fn site_replication_item(credentials: &rustfs_credentials::Credentials, updated_at: OffsetDateTime) -> SRIAMItem { SRIAMItem { r#type: "sts-credential".to_string(), sts_credential: Some(SRSTSCredential { @@ -233,7 +244,7 @@ fn site_replication_item( secret_key: credentials.secret_key.clone(), session_token: credentials.session_token.clone(), parent_user: credentials.parent_user.clone(), - parent_policy_mapping: transaction.authorization.policies.join(","), + parent_policy_mapping: OIDC_STS_REQUIRES_VIRTUAL_PARENT_RECEIVER_POLICY.to_string(), api_version: Some(SITE_REPL_API_VERSION.to_string()), }), updated_at: Some(updated_at), @@ -249,17 +260,35 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding { transaction: &FederatedSessionTransaction, ) -> Result { let authorization = &transaction.authorization; - let secret = current_token_signing_key(); - let credentials = issue_credentials(transaction, secret.as_deref())?; - let iam_store = current_ready_iam_handle().map_err(|_| FederatedSessionBindingError::Internal("IAM not initialized".to_string()))?; + 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 = match iam_store + .policy_db_get(&parent_user, &Some(authorization.groups.clone())) + .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.clone(), + }; + let selected_policy_mapping = selected_policy_names.join(","); + let resolved_policy_mapping = iam_store.current_policies(&selected_policy_mapping).await; + if !all_oidc_policies_resolved(&selected_policy_names, &resolved_policy_mapping) { + return Err(FederatedSessionBindingError::InvalidRequest( + "OIDC policy mapping did not resolve to current policies".to_string(), + )); + } + + let secret = current_token_signing_key(); + let credentials = issue_credentials(transaction, &selected_policy_names, secret.as_deref())?; if tracing::enabled!(tracing::Level::DEBUG) { log_oidc_policy_diagnostics( &iam_store, &authorization.provider_id, &credentials.parent_user, - &authorization.policies, + &selected_policy_names, &authorization.groups, ) .await; @@ -270,7 +299,7 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding { .await .map_err(|_| FederatedSessionBindingError::Internal("failed to store temp user".to_string()))?; - if let Err(err) = site_replication_iam_change_hook(site_replication_item(&credentials, transaction, updated_at)).await { + if let Err(err) = site_replication_iam_change_hook(site_replication_item(&credentials, updated_at)).await { warn!("site replication OIDC STS hook failed, err: {err}"); } @@ -292,7 +321,7 @@ mod tests { email: "user@example.com".to_string(), username: "user".to_string(), groups: vec!["source-group".to_string()], - raw: HashMap::new(), + raw: HashMap::from([("iss".to_string(), serde_json::json!("https://idp.example.test"))]), }, policies: vec!["readwrite".to_string()], groups: vec!["devs".to_string()], @@ -322,23 +351,32 @@ mod tests { fn issued_credentials_and_replication_item_preserve_existing_shape() { let transaction = transaction(); let secret = "federated-session-test-signing-secret"; + let selected_policy_names = vec!["readonly".to_string()]; - let credentials = issue_credentials(&transaction, Some(secret)).expect("credential issuance should succeed"); - assert_eq!(credentials.parent_user, "user"); + let credentials = + issue_credentials(&transaction, &selected_policy_names, Some(secret)).expect("credential issuance should succeed"); + assert_eq!(credentials.parent_user, "openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I"); assert_eq!(credentials.groups, Some(vec!["devs".to_string()])); let claims = rustfs_iam::sys::get_claims_from_token_with_secret(&credentials.session_token, secret) .expect("issued session token should verify"); assert_eq!(claims.get("iss"), Some(&serde_json::json!("rustfs-oidc"))); assert_eq!(claims.get("oidc_provider"), Some(&serde_json::json!("default"))); - assert_eq!(claims.get("parent"), Some(&serde_json::json!("user"))); - assert_eq!(claims.get("policy"), Some(&serde_json::json!("readwrite"))); + assert_eq!( + claims.get("parent"), + Some(&serde_json::json!("openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I")) + ); + assert_eq!( + claims.get(OIDC_VIRTUAL_PARENT_CLAIM), + Some(&serde_json::json!("openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I")) + ); + assert_eq!(claims.get("policy"), Some(&serde_json::json!("readonly"))); assert_eq!(claims.get("groups"), Some(&serde_json::json!(["devs"]))); assert_eq!(claims.get("roles"), Some(&serde_json::json!(["admin", "reader"]))); assert!(!claims.contains_key("oidc_issuer")); let updated_at = OffsetDateTime::UNIX_EPOCH; - let item = site_replication_item(&credentials, &transaction, updated_at); + let item = site_replication_item(&credentials, updated_at); assert_eq!(item.r#type, "sts-credential"); assert_eq!(item.updated_at, Some(updated_at)); assert_eq!(item.api_version.as_deref(), Some(SITE_REPL_API_VERSION)); @@ -346,8 +384,10 @@ mod tests { assert_eq!(replicated.access_key, credentials.access_key); assert_eq!(replicated.secret_key, credentials.secret_key); assert_eq!(replicated.session_token, credentials.session_token); - assert_eq!(replicated.parent_user, "user"); - assert_eq!(replicated.parent_policy_mapping, "readwrite"); + assert_eq!(replicated.parent_user, "openid=pUmguI1petsjVfDFQppmmR9yqdmWnBAXGJhHV_s9W3I"); + assert_eq!(replicated.parent_policy_mapping, OIDC_STS_REQUIRES_VIRTUAL_PARENT_RECEIVER_POLICY); + assert!(replicated.parent_policy_mapping.trim().is_empty()); + assert!(MappedPolicy::new(&replicated.parent_policy_mapping).to_slice().is_empty()); assert_eq!(replicated.api_version.as_deref(), Some(SITE_REPL_API_VERSION)); } @@ -356,7 +396,8 @@ mod tests { let mut transaction = transaction(); transaction.session_policy = Some("not-json".to_string()); - let error = issue_credentials(&transaction, None).expect_err("invalid policy should fail first"); + let error = issue_credentials(&transaction, &transaction.authorization.policies, None) + .expect_err("invalid policy should fail first"); assert!(matches!(error, FederatedSessionBindingError::InvalidRequest(_))); } @@ -377,9 +418,29 @@ mod tests { // Reuse the collision decision the issuance path applies: a federated identity whose // derived parent_user equals the root access key must be denied at issuance. let transaction = transaction(); - let parent_user = transaction.authorization.claims.session_identity(); - assert_eq!(parent_user, "user"); - assert!(parent_user_is_reserved(&parent_user, Some("user"))); + let parent_user = transaction + .authorization + .oidc_virtual_parent() + .expect("fixture must contain issuer and subject"); + assert!(parent_user_is_reserved(&parent_user, Some(&parent_user))); assert!(!parent_user_is_reserved(&parent_user, Some("root"))); } + + #[test] + fn missing_oidc_issuer_fails_closed_before_credential_issuance() { + let mut transaction = transaction(); + transaction.authorization.claims.raw.clear(); + + let error = issue_credentials(&transaction, &transaction.authorization.policies, Some("signing-secret")) + .expect_err("credential issuance should reject a missing issuer"); + assert!(matches!(error, FederatedSessionBindingError::InvalidRequest(_))); + } + + #[test] + fn oidc_replication_requires_all_selected_policies() { + assert!(all_oidc_policies_resolved(&["readonly".to_string()], "readonly")); + assert!(!all_oidc_policies_resolved(&["readonly".to_string(), "missing".to_string()], "readonly")); + assert!(!all_oidc_policies_resolved(&[], "")); + assert!(!all_oidc_policies_resolved(&["team+readonly".to_string()], "team+readonly")); + } } diff --git a/rustfs/src/auth.rs b/rustfs/src/auth.rs index 81f328b46..ae4cd4a73 100644 --- a/rustfs/src/auth.rs +++ b/rustfs/src/auth.rs @@ -246,6 +246,14 @@ pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result< check_key_valid_with_context(session_token, access_key, None).await } +fn has_root_access(sys_cred: &Credentials, cred: &Credentials) -> bool { + (constant_time_eq(&sys_cred.access_key, &cred.access_key) || constant_time_eq(&cred.parent_user, &sys_cred.access_key)) + && !cred + .claims + .as_ref() + .is_some_and(|claims| claims.contains_key(SESSION_POLICY_NAME) || rustfs_iam::sys::is_rustfs_oidc_claims(claims)) +} + /// Validate an access key, resolving the root credentials and IAM system from /// an explicit application context when one is given (backlog#1052 S6). /// @@ -427,16 +435,7 @@ pub async fn check_key_valid_with_context( cred.claims = if !claims.is_empty() { Some(claims) } else { None }; - let mut owner = - constant_time_eq(&sys_cred.access_key, &cred.access_key) || constant_time_eq(&cred.parent_user, &sys_cred.access_key); - - // permitRootAccess - if let Some(claims) = &cred.claims - && claims.contains_key(SESSION_POLICY_NAME) - { - owner = false - } - + let owner = has_root_access(&sys_cred, &cred); Ok((cred, owner)) } @@ -1080,6 +1079,26 @@ mod tests { } } + #[test] + fn oidc_session_cannot_inherit_root_access_from_display_name() { + let sys_cred = Credentials { + access_key: "root-access-key".to_string(), + ..Default::default() + }; + let oidc_cred = Credentials { + access_key: "temporary-access-key".to_string(), + parent_user: sys_cred.access_key.clone(), + claims: Some(HashMap::from([ + ("iss".to_string(), json!("rustfs-oidc")), + ("oidc_provider".to_string(), json!("default")), + ("sub".to_string(), json!("subject-123")), + ])), + ..Default::default() + }; + + assert!(!has_root_access(&sys_cred, &oidc_cred)); + } + #[test] fn test_iam_auth_creation() { let access_key = "test-access-key"; diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 07a9c1473..b82abb52b 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -1380,13 +1380,20 @@ impl Node for NodeService { error_info: Some("access_key name is missing".to_string()), })); } - let Some(iam_sys) = runtime_sources::current_iam_handle() else { + let Some(iam_sys) = self + .context + .as_ref() + .map(|context| context.iam().handle()) + .or_else(runtime_sources::current_iam_handle) + else { return Ok(Response::new(DeleteServiceAccountResponse { success: false, error_info: Some("errServerNotInitialized".to_string()), })); }; - let resp = iam_sys.delete_service_account(&access_key, false).await; + // This legacy RPC is a cache notification. Reloading shared state keeps a + // delayed delete notification from removing a recreated service account. + let resp = iam_sys.load_service_account(&access_key).await; if let Err(err) = resp { return Ok(Response::new(DeleteServiceAccountResponse { success: false, @@ -1859,7 +1866,7 @@ mod tests { PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, STORAGE_CLASS_SUB_SYS, admit_heal_control_replay, background_rebalance_start_error_message, execute_heal_control_envelope_with_manager, initialize_heal_topology_fingerprint, make_heal_control_server, - make_heal_control_server_with_cache, make_server, make_tier_mutation_control_server_for_context, + make_heal_control_server_with_cache, make_server, make_server_for_context, make_tier_mutation_control_server_for_context, remove_heal_control_replay, scanner_activity_response, stop_rebalance_response, }; use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; @@ -1871,6 +1878,14 @@ mod tests { }; use bytes::Bytes; use rustfs_heal::heal::{manager::HealManager, storage::HealStorageAPI}; + use rustfs_iam::{ + store::{ + Store as _, + object::{IAM_CONFIG_PREFIX, ObjectStore}, + }, + sys::NewServiceAccountOpts, + }; + use rustfs_kms::KmsServiceManager; use rustfs_protos::models::PingBodyBuilder; use rustfs_protos::proto_gen::node_service::{ BackgroundHealStatusRequest, CheckPartsRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, @@ -4088,6 +4103,59 @@ mod tests { assert!(delete_response.error_info.unwrap().contains("access_key name is missing")); } + #[tokio::test] + async fn delete_service_account_rpc_reloads_instead_of_deleting_shared_state() { + let _ = rustfs_credentials::init_global_action_credentials( + Some("TESTROOTACCESSKEY".to_string()), + Some("TESTROOTSECRET123".to_string()), + ); + let temp_dir = tempfile::tempdir().expect("service-account RPC test directory"); + let env = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(temp_dir.path()) + .init_bucket_metadata(false) + .build() + .await; + ObjectStore::new(Arc::clone(&env.ecstore)) + .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)) + .await + .expect("seed IAM format"); + let iam = rustfs_iam::build_iam_sys(Arc::clone(&env.ecstore)) + .await + .expect("build isolated IAM"); + let context = Arc::new(crate::runtime_sources::AppContext::with_default_interfaces( + Arc::clone(&env.ecstore), + Arc::clone(&iam), + Arc::new(KmsServiceManager::new()), + )); + let service = make_server_for_context(Some(context)); + let access_key = "RPCRELOADSERVICE01"; + iam.new_service_account( + "parent-user", + None, + NewServiceAccountOpts { + access_key: access_key.to_string(), + secret_key: "rpcReloadServiceSecret123".to_string(), + ..Default::default() + }, + ) + .await + .expect("create service account"); + + let response = service + .delete_service_account(Request::new(DeleteServiceAccountRequest { + access_key: access_key.to_string(), + })) + .await + .expect("legacy notification RPC response") + .into_inner(); + + assert!(response.success, "cache reload notification must succeed"); + assert!( + iam.get_service_account(access_key).await.is_ok(), + "legacy delete notification must not delete durable service-account state" + ); + } + #[tokio::test] async fn test_load_user_empty_access_key() { let service = create_test_node_service();