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

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

- Admin and STS authorization used the socket peer as aws:SourceIp
  behind a trusted proxy; they now read the verified client address the
  S3 path already uses (backlog#3025), and the admin usage route passes
  the bucket named in the path to the s3:ListBucket check (backlog#3033).
- AssumeRoleWithWebIdentity accepted HS256 ID tokens MACed with the OIDC
  client secret; only provider-signed tokens verify (backlog#3028), and
  JWKS refresh is single-flight with a failure cool-down (backlog#3103).
- A disabled group emptied the member's whole policy list, dropping
  every Deny; direct and other-group policies are preserved (backlog#3042).
- Identity rewrites (secret change, add-user, status) ran without a lock
  and without re-reading the record, so a user's password change could
  undo an admin disable or reset (backlog#3048). add-user on an existing
  user and in-flight AssumeRole both left sessions minted under the old
  secret valid (backlog#3071, backlog#3079); rotation now revokes under
  the per-key mutation lock and derived-credential publication is ordered
  with it.
- aws:groups never carried a built-in user's IAM group membership and
  jwt:iss was always the local marker, so group- and issuer-scoped
  Denies never matched; both now come from live IAM state
  (backlog#3095, backlog#3096).
- OPA mode skipped the account-status gates for service accounts and STS
  sessions of a disabled user (backlog#3057).
- The periodic IAM reload published nothing when any cache write raced
  it and kept stale partitions when a prefix listed empty, so a missed
  revocation never applied (backlog#3064, backlog#3114).
- update-group-members never evaluated admin:RemoveUserFromGroup
  (backlog#3098); ExportIAM compared the site-replicator account against
  a name it never has (backlog#3045); add-user accepted the access key of
  a live STS session or service account (backlog#3134); a service account
  could create or extend a lifetime beyond its own (backlog#3112);
  sts:DurationSeconds and svc:DurationSeconds were never populated
  (backlog#3136); temporary-account-info returned service accounts
  (backlog#3137); an access key the object layer refuses no longer holds
  the service-account lock through the write retries (backlog#3133).

.github/workflows/nightly-gnu.yml

@@ -87,18 +87,206 @@ jobs:
           install-build-packaging-tools: 'false'
           install-test-tools: 'false'
 
+      - name: Download static console assets
+        shell: bash
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          # Packaging tools are off on this lane, so setup does not install unzip.
+          if ! command -v unzip >/dev/null 2>&1; then
+            sudo apt-get update -qq
+            sudo apt-get install -y -qq unzip
+          fi
+          if ! command -v unzip >/dev/null 2>&1; then
+            echo "unzip is required to extract console assets" >&2
+            exit 1
+          fi
+
+          mkdir -p ./rustfs/static
+
+          verify_sha256() {
+            local expected="$1"
+            local file="$2"
+            local actual=""
+            local actual_line=""
+
+            if command -v sha256sum >/dev/null 2>&1; then
+              actual_line=$(sha256sum -- "$file") || return 1
+            elif command -v shasum >/dev/null 2>&1; then
+              actual_line=$(shasum -a 256 -- "$file") || return 1
+            else
+              echo "No SHA-256 verification tool found" >&2
+              return 1
+            fi
+            actual="${actual_line%%[[:space:]]*}"
+
+            if [ "$actual" = "$expected" ]; then
+              echo "SHA256 verified OK: $actual"
+              return 0
+            else
+              echo "SHA256 mismatch: expected=$expected actual=$actual" >&2
+              return 1
+            fi
+          }
+
+          download_console_assets() {
+            local curl_bin="curl"
+            local python_bin="python3"
+            local console_api="https://api.github.com/repos/rustfs/console/releases/latest"
+            local console_json="console-release.json"
+            local console_url
+            local console_sha256
+            local curl_auth_args=()
+            console_tag=""
+
+            chmod +w ./rustfs/static/LICENSE || true
+
+            if ! command -v "$python_bin" >/dev/null 2>&1; then
+              python_bin="python"
+            fi
+            if ! command -v "$python_bin" >/dev/null 2>&1; then
+              echo "No Python interpreter found for release metadata parsing" >&2
+              return 1
+            fi
+
+            if [[ -n "${GITHUB_TOKEN:-}" ]]; then
+              curl_auth_args=(-H "Authorization: Bearer ${GITHUB_TOKEN}" -H "X-GitHub-Api-Version: 2022-11-28")
+            fi
+
+            "$curl_bin" "${curl_auth_args[@]}" --fail -L "$console_api" \
+              -o "$console_json" --retry 3 --retry-delay 5 --max-time 300 || return 1
+
+            read -r console_tag console_url console_sha256 < <("$python_bin" - "$console_json" <<'PY'
+          import json
+          import re
+          import sys
+
+          with open(sys.argv[1], encoding="utf-8") as handle:
+              release = json.load(handle)
+
+          tag = release.get("tag_name", "") or "unknown"
+
+          for asset in release.get("assets", []):
+              name = asset.get("name", "")
+              digest = asset.get("digest", "")
+              url = asset.get("browser_download_url", "")
+              if name.startswith("rustfs-console-") and name.endswith(".zip") and digest.startswith("sha256:") and url:
+                  sha256 = digest.split(":", 1)[1]
+                  if not re.fullmatch(r"[0-9a-fA-F]{64}", sha256):
+                      raise SystemExit(f"console zip asset has invalid sha256 digest: {sha256}")
+                  sys.stdout.buffer.write(f"{tag} {url} {sha256}\n".encode("utf-8"))
+                  break
+          else:
+              raise SystemExit("no console zip asset with sha256 digest found")
+          PY
+            ) || return 1
+
+            if [[ -z "$console_url" || -z "$console_sha256" ]]; then
+              echo "Console release metadata is missing URL or SHA-256 digest" >&2
+              return 1
+            fi
+
+            echo "Console release: ${console_tag}"
+            echo "Downloading console asset: ${console_url}"
+            "$curl_bin" --fail -L "$console_url" -o console.zip --retry 3 --retry-delay 5 --max-time 300 || return 1
+            verify_sha256 "$console_sha256" console.zip || return 2
+            unzip -o console.zip -d ./rustfs/static || return 2
+          }
+
+          if download_console_assets; then
+            rm -f console.zip console-release.json
+          else
+            status=$?
+            rm -f console.zip console-release.json
+            if [[ "$status" -eq 2 ]]; then
+              echo "Console asset integrity verification failed" >&2
+            fi
+            echo "Failed to download verified console assets" >&2
+            exit 1
+          fi
+
+          if [[ ! -s ./rustfs/static/index.html ]]; then
+            echo "Console asset archive is missing static/index.html" >&2
+            exit 1
+          fi
+
+          asset_count=$(find ./rustfs/static -type f | wc -l | tr -d '[:space:]')
+          echo "Console assets ready: version=${console_tag:-unknown}, ${asset_count} files extracted to ./rustfs/static"
+
       - name: Build RustFS
         # Pin the glibc baseline (`<target>.<glibc>` for cargo-zigbuild) so
         # nightlies run on older distros such as RHEL/Rocky 9 (glibc 2.34); #8062.
-        run: cargo zigbuild --release --locked --target x86_64-unknown-linux-gnu.2.31 -p rustfs --bins
+        shell: bash
+        run: |
+          # Force rebuild by touching build.rs so a restored cache cannot
+          # reuse a rustfs artifact compiled before these console assets existed.
+          touch rustfs/build.rs
+          cargo zigbuild --release --locked --target x86_64-unknown-linux-gnu.2.31 -p rustfs --bin rustfs
 
       - name: Verify glibc symbol baseline
         run: |
           python3 scripts/release/check_glibc.py 2.31 \
-            target/x86_64-unknown-linux-gnu/release/rustfs \
-            target/x86_64-unknown-linux-gnu/release/rustfs-cli
+            target/x86_64-unknown-linux-gnu/release/rustfs
           target/x86_64-unknown-linux-gnu/release/rustfs --version | grep -Fx 'build target : x86_64-unknown-linux-gnu'
 
+      - name: Verify console
+        shell: bash
+        run: |
+          set -euo pipefail
+
+          find_free_port() {
+            python3 - <<'PY'
+          import socket
+
+          with socket.socket() as sock:
+              sock.bind(("127.0.0.1", 0))
+              print(sock.getsockname()[1])
+          PY
+          }
+
+          binary="target/x86_64-unknown-linux-gnu/release/rustfs"
+          data_dir="$(mktemp -d)"
+          log_file="${RUNNER_TEMP}/rustfs-console-smoke.log"
+          server_pid=""
+          trap '
+            if [[ -n "${server_pid:-}" ]]; then
+              kill "$server_pid" 2>/dev/null || true
+              wait "$server_pid" 2>/dev/null || true
+            fi
+            rm -rf "$data_dir"
+          ' EXIT
+
+          if [[ ! -x "$binary" ]]; then
+            echo "rustfs binary not found: ${binary}" >&2
+            exit 1
+          fi
+
+          api_port="$(find_free_port)"
+          console_port="$(find_free_port)"
+          console_url="http://127.0.0.1:${console_port}/rustfs/console/"
+
+          "$binary" server \
+            --address "127.0.0.1:${api_port}" \
+            --console-enable \
+            --console-address "127.0.0.1:${console_port}" \
+            --access-key console-smoke \
+            --secret-key console-smoke-secret \
+            "$data_dir" >"$log_file" 2>&1 &
+          server_pid=$!
+
+          response=""
+          for _ in {1..40}; do
+            response="$(curl --silent --output /dev/null --write-out '%{http_code} %{content_type}' "$console_url" || true)"
+            if [[ "$response" == 200\ text/html* ]]; then
+              exit 0
+            fi
+            sleep 0.25
+          done
+
+          echo "Console endpoint did not return 200 text/html: $response" >&2
+          cat "$log_file"
+          exit 1
+
       - name: Build DEB package
         id: deb
         shell: bash

crates/ecstore/src/disk/local.rs

@@ -11222,8 +11222,16 @@ impl LocalDisk {
 
                 if entry.ends_with(SLASH_SEPARATOR) {
                     if entry.ends_with(GLOBAL_DIR_SUFFIX_WITH_SLASH) {
-                        let entry =
-                            format!("{}{}", entry.as_str().trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH), SLASH_SEPARATOR);
+                        let marker_name = entry.as_str().trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH);
+                        // An empty marker name would schedule the directory being
+                        // scanned as its own child and recurse until the stack
+                        // overflows. Ignore malformed markers left by older writers
+                        // or corrupt disks instead of decoding them.
+                        if marker_name.is_empty() {
+                            item.clear();
+                            continue;
+                        }
+                        let entry = format!("{marker_name}{SLASH_SEPARATOR}");
                         dir_objes.insert(entry.clone());
                         *item = entry;
                         continue;
@@ -26834,6 +26842,49 @@ mod test {
         assert_eq!(objs_returned, 1);
     }
 
+    #[tokio::test]
+    async fn test_scan_dir_ignores_empty_directory_markers() {
+        use rustfs_filemeta::MetacacheReader;
+        use tempfile::tempdir;
+
+        let dir = tempdir().expect("create disk directory");
+        let bucket = "test-bucket";
+        let bucket_dir = dir.path().join(bucket);
+        for object in ["__XLDIR__", "nested/__XLDIR__", "nested/healthy", "folder__XLDIR__"] {
+            let object_dir = bucket_dir.join(object);
+            fs::create_dir_all(&object_dir).await.expect("create object directory");
+            fs::write(object_dir.join("xl.meta"), b"meta")
+                .await
+                .expect("write object metadata");
+        }
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("valid disk endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
+        let (reader, mut writer) = tokio::io::duplex(4096);
+        let mut out = MetacacheWriter::new(&mut writer);
+        let opts = WalkDirOptions {
+            bucket: bucket.to_owned(),
+            recursive: true,
+            ..Default::default()
+        };
+        let mut count = 0;
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            disk.scan_dir(String::new(), String::new(), &opts, &mut out, &mut count, false, None),
+        )
+        .await
+        .expect("malformed directory marker must not recurse into its parent")
+        .expect("scan succeeds despite malformed markers");
+        out.close().await.expect("close metacache stream");
+        let entries = MetacacheReader::new(reader).read_all().await.expect("read listed objects");
+        let names: Vec<_> = entries
+            .into_iter()
+            .filter(|entry| !entry.metadata.is_empty())
+            .map(|entry| entry.name)
+            .collect();
+        assert_eq!(names, vec!["folder/", "nested/healthy"]);
+        assert_eq!(count, 2);
+    }
+
     #[tokio::test]
     async fn test_scan_dir_deduplicates_explicit_dir_marker_recursion() {
         use rustfs_filemeta::MetacacheReader;

crates/ecstore/src/set_disk/ops/object.rs

@@ -9304,6 +9304,60 @@ impl SetDisks {
 }
 
 impl SetDisks {
+    /// Judge the version a versionless destructive delete removes.
+    ///
+    /// For a key that is not versioning-`Enabled`, a delete without a version
+    /// id removes the `null` version, while the preceding Object Lock check
+    /// judged the key's latest entry. When a UUID version or delete marker
+    /// sits above a retained `null` version, the two differ and the retained
+    /// version must be judged on its own. The caller holds the object's write
+    /// lock. Plain unversioned keys and versioning-`Enabled` keys keep a single
+    /// lookup: their latest entry is the version removed.
+    ///
+    /// The lookup future is boxed from a synchronous wrapper so it leaves the
+    /// delete futures' frames: `delete_object` already nests the boxed batch
+    /// path, and inlining a second `get_object_info` overflowed the 4 MiB test
+    /// stack on Linux.
+    fn check_null_version_object_lock_delete<'a>(
+        &'a self,
+        bucket: &'a str,
+        object: &'a str,
+        latest: &'a ObjectInfo,
+        opts: &'a ObjectOptions,
+    ) -> impl Future<Output = Result<()>> + Send + 'a {
+        Box::pin(self.check_null_version_object_lock_delete_inner(bucket, object, latest, opts))
+    }
+
+    async fn check_null_version_object_lock_delete_inner(
+        &self,
+        bucket: &str,
+        object: &str,
+        latest: &ObjectInfo,
+        opts: &ObjectOptions,
+    ) -> Result<()> {
+        if is_meta_bucketname(bucket)
+            || opts.version_id.is_some()
+            || set_disk_delete_creates_delete_marker(opts)
+            || opts.transition.expire_restored
+            || latest.version_id.is_none_or(|version| version.is_nil())
+        {
+            return Ok(());
+        }
+        let null_opts = ObjectOptions {
+            version_id: Some(Uuid::nil().to_string()),
+            no_lock: true,
+            metadata_cache_safe: false,
+            ..opts.clone()
+        };
+        match self.get_object_info(bucket, object, &null_opts).await {
+            Ok(existing) => check_object_lock_delete(&self.ctx, bucket, object, &existing, &null_opts).await,
+            // A null delete marker has no protected payload to remove.
+            Err(StorageError::MethodNotAllowed) => Ok(()),
+            Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(()),
+            Err(err) => Err(err),
+        }
+    }
+
     /// Read current metadata without cache admission or read-side repair. The
     /// outer batch owner already holds the sorted namespace write-lock set.
     pub(crate) async fn conditional_delete_current_snapshot(
@@ -10022,6 +10076,17 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 del_errs[i] = Some(err);
                 continue;
             }
+            // Not gated on `source_missing`: with a delete marker on top the
+            // latest-version lookup reports the key as missing while the null
+            // version below it may still be retained.
+            if object_lock_check_required
+                && let Err(err) = self
+                    .check_null_version_object_lock_delete(bucket, &dobj.object_name, &goi, &check_opts)
+                    .await
+            {
+                del_errs[i] = Some(err);
+                continue;
+            }
 
             if opts.tier_delete_journal_api.is_some()
                 && !source_missing
@@ -11094,6 +11159,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             opts.precondition_check(&goi)?;
             check_object_lock_delete(&self.ctx, bucket, object, &goi, &opts).await?;
         }
+        self.check_null_version_object_lock_delete(bucket, object, &goi, &opts)
+            .await?;
         let delete_marker_purge_candidate = opts
             .version_id
             .as_deref()

crates/ecstore/src/store/init.rs

@@ -27352,4 +27352,201 @@ mod tests {
             .expect("recovery must preserve the local source object");
         assert_eq!(source_after.size, i64::try_from(payload.len()).expect("payload length should fit i64"));
     }
+
+    #[cfg(feature = "test-util")]
+    async fn seed_retained_null_version(
+        store: &Arc<crate::store::ECStore>,
+        bucket: &str,
+        object: &str,
+        protection: &str,
+        latest_is_marker: bool,
+    ) {
+        let metadata = if protection == "legal-hold" {
+            HashMap::from([("x-amz-object-lock-legal-hold".to_string(), "ON".to_string())])
+        } else {
+            HashMap::from([
+                ("x-amz-object-lock-mode".to_string(), protection.to_string()),
+                ("x-amz-object-lock-retain-until-date".to_string(), "2099-01-01T00:00:00Z".to_string()),
+            ])
+        };
+        let mut original = PutObjReader::from_vec(b"protected null version".to_vec());
+        store
+            .put_object(
+                bucket,
+                object,
+                &mut original,
+                &ObjectOptions {
+                    user_defined: metadata,
+                    mod_time: Some(OffsetDateTime::now_utc() - time::Duration::days(1)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed retained null version");
+        let mut latest = PutObjReader::from_vec(b"unprotected latest version".to_vec());
+        store
+            .put_object(
+                bucket,
+                object,
+                &mut latest,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed newer unprotected UUID version");
+        if latest_is_marker {
+            store
+                .delete_object(
+                    bucket,
+                    object,
+                    ObjectOptions {
+                        versioned: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("create a UUID delete marker above the retained null version");
+        }
+    }
+
+    #[cfg(feature = "test-util")]
+    async fn assert_retained_null_version_body(store: &Arc<crate::store::ECStore>, bucket: &str, object: &str) {
+        let mut reader = store
+            .get_object_reader(
+                bucket,
+                object,
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    version_id: Some(Uuid::nil().to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("retained null version must remain readable");
+        let mut body = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut body)
+            .await
+            .expect("read retained null version");
+        assert_eq!(body, b"protected null version");
+    }
+
+    /// A versionless delete removes the null version, so the Object Lock check
+    /// must judge that version even when a UUID version or delete marker is
+    /// the key's latest entry (rustfs/backlog#3017).
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn null_version_object_lock_delete_checks_actual_target() {
+        use crate::bucket::replication::DeleteReplicationConfigSnapshot;
+        use s3s::dto::{BucketVersioningStatus, ExcludedPrefix, VersioningConfiguration};
+
+        let temp = tempfile::tempdir().expect("create null-version delete store");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "null-version-delete-lock", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("null-delete-lock-{}", Uuid::new_v4());
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create Object Lock bucket");
+        let configs = [
+            VersioningConfiguration::default(),
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)),
+                ..Default::default()
+            },
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                excluded_prefixes: Some(vec![ExcludedPrefix {
+                    prefix: Some("archive/".to_string()),
+                }]),
+                ..Default::default()
+            },
+        ];
+        for (case, config) in configs.into_iter().enumerate() {
+            let snapshot = Arc::new(DeleteReplicationConfigSnapshot::from_configs_for_test(config, None));
+            for protection in ["COMPLIANCE", "GOVERNANCE", "legal-hold"] {
+                for latest_is_marker in [false, true] {
+                    let object = format!("archive/{case}-{protection}-{latest_is_marker}");
+                    seed_retained_null_version(&store, &bucket, &object, protection, latest_is_marker).await;
+                    let (_, errors) = store
+                        .delete_objects(
+                            &bucket,
+                            vec![ObjectToDelete {
+                                object_name: object.clone(),
+                                ..Default::default()
+                            }],
+                            ObjectOptions {
+                                delete_replication_config_snapshot: Some(Arc::clone(&snapshot)),
+                                ..Default::default()
+                            },
+                        )
+                        .await;
+                    assert!(
+                        matches!(errors.as_slice(), [Some(StorageError::PrefixAccessDenied(_, _))]),
+                        "batch delete must judge the retained null version: {object}: {errors:?}"
+                    );
+                    let error = store
+                        .delete_object(
+                            &bucket,
+                            &object,
+                            ObjectOptions {
+                                version_suspended: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect_err("suspended single delete must also judge the retained null version");
+                    assert!(matches!(error, StorageError::PrefixAccessDenied(_, _)), "{object}: {error:?}");
+                    assert_retained_null_version_body(&store, &bucket, &object).await;
+                    if protection == "GOVERNANCE" {
+                        let (_, errors) = store
+                            .delete_objects(
+                                &bucket,
+                                vec![ObjectToDelete {
+                                    object_name: object.clone(),
+                                    ..Default::default()
+                                }],
+                                ObjectOptions {
+                                    delete_replication_config_snapshot: Some(Arc::clone(&snapshot)),
+                                    object_lock_delete: Some(crate::storage_api_contracts::object::ObjectLockDeleteOptions {
+                                        bypass_governance: true,
+                                    }),
+                                    ..Default::default()
+                                },
+                            )
+                            .await;
+                        assert!(errors.iter().all(Option::is_none), "authorized governance bypass: {errors:?}");
+                        assert!(
+                            store
+                                .get_object_reader(
+                                    &bucket,
+                                    &object,
+                                    None,
+                                    HeaderMap::new(),
+                                    &ObjectOptions {
+                                        version_id: Some(Uuid::nil().to_string()),
+                                        ..Default::default()
+                                    }
+                                )
+                                .await
+                                .is_err(),
+                            "authorized deletion must make the retained null payload unreadable"
+                        );
+                    }
+                }
+            }
+        }
+    }
 }

crates/iam/src/cache.rs

@@ -15,7 +15,6 @@
 use std::{
     collections::{HashMap, HashSet},
     ops::{Deref, DerefMut},
-    ptr,
     sync::{Arc, Mutex, Weak},
 };
 
@@ -64,21 +63,46 @@ impl Default for CacheState {
 pub struct Cache {
     state: ArcSwap<CacheState>,
     write_lock: Mutex<()>,
+    reload_lock: AsyncMutex<()>,
+    reload_changes: Mutex<Weak<Mutex<ReloadChanges>>>,
     service_account_mutation_lock: AsyncMutex<()>,
-    sts_account_mutation_locks: Arc<StsMutationLockRegistry>,
+    sts_account_mutation_locks: Arc<MutationLockRegistry>,
+    parent_mutation_locks: Arc<MutationLockRegistry>,
 }
 
-struct StsMutationLockRegistry {
-    locks: Mutex<HashMap<String, Weak<AsyncMutex<StsMutationLockState>>>>,
+#[derive(Default)]
+struct MutationLockRegistry {
+    locks: Mutex<HashMap<String, Weak<AsyncMutex<MutationLockState>>>>,
 }
 
-pub(crate) struct StsMutationLockState {
+pub(crate) struct MutationLockState {
     access_key: String,
-    registry: Weak<StsMutationLockRegistry>,
-    lock: Weak<AsyncMutex<StsMutationLockState>>,
+    registry: Weak<MutationLockRegistry>,
+    lock: Weak<AsyncMutex<MutationLockState>>,
 }
 
-impl Drop for StsMutationLockState {
+impl MutationLockRegistry {
+    fn lock_for(self: &Arc<Self>, access_key: &str) -> Arc<AsyncMutex<MutationLockState>> {
+        let mut locks = self.locks.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        if let Some(lock) = locks.get(access_key).and_then(Weak::upgrade) {
+            return lock;
+        }
+
+        let registry = Arc::downgrade(self);
+        let access_key_owned = access_key.to_string();
+        let lock = Arc::new_cyclic(|lock| {
+            AsyncMutex::new(MutationLockState {
+                access_key: access_key_owned,
+                registry,
+                lock: lock.clone(),
+            })
+        });
+        locks.insert(access_key.to_string(), Arc::downgrade(&lock));
+        lock
+    }
+}
+
+impl Drop for MutationLockState {
     fn drop(&mut self) {
         let Some(registry) = self.registry.upgrade() else {
             return;
@@ -95,10 +119,52 @@ impl Default for Cache {
         Self {
             state: ArcSwap::new(Arc::new(CacheState::default())),
             write_lock: Mutex::new(()),
+            reload_lock: AsyncMutex::new(()),
+            reload_changes: Mutex::new(Weak::new()),
             service_account_mutation_lock: AsyncMutex::new(()),
-            sts_account_mutation_locks: Arc::new(StsMutationLockRegistry {
-                locks: Mutex::new(HashMap::new()),
-            }),
+            sts_account_mutation_locks: Arc::new(MutationLockRegistry::default()),
+            parent_mutation_locks: Arc::new(MutationLockRegistry::default()),
+        }
+    }
+}
+
+#[derive(Clone, Copy, Hash, PartialEq, Eq)]
+pub(crate) enum CachePartition {
+    PolicyDocs,
+    Users,
+    UserPolicies,
+    StsAccounts,
+    StsPolicies,
+    Groups,
+    UserGroupMemberships,
+    GroupPolicies,
+}
+
+type ReloadChanges = HashMap<CachePartition, HashSet<String>>;
+
+/// A reload journals only keys changed during its own lifetime, including
+/// deletions. Unrelated STS traffic cannot suppress a persisted revocation.
+pub(crate) struct CacheReload<'a> {
+    _guard: tokio::sync::MutexGuard<'a, ()>,
+    changes: Arc<Mutex<ReloadChanges>>,
+}
+
+impl CacheReload<'_> {
+    pub(crate) fn merge_partition<T: Clone>(
+        &self,
+        partition: CachePartition,
+        loaded: &mut CacheEntity<T>,
+        current: &CacheEntity<T>,
+    ) {
+        let changes = self.changes.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        if let Some(keys) = changes.get(&partition) {
+            for key in keys {
+                if let Some(value) = current.get(key) {
+                    loaded.insert(key.clone(), value.clone());
+                } else {
+                    loaded.remove(key);
+                }
+            }
         }
     }
 }
@@ -106,31 +172,27 @@ impl Default for Cache {
 pub(crate) type CacheSnapshot = Guard<Arc<CacheState>>;
 
 impl Cache {
+    pub(crate) async fn begin_reload(&self) -> CacheReload<'_> {
+        // Registration and publication both take write_lock before the journal
+        // lock, so no mutation can fall between the reload and its journal.
+        let guard = self.reload_lock.lock().await;
+        let changes = Arc::new(Mutex::new(HashMap::new()));
+        let _write = self.write_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        *self.reload_changes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = Arc::downgrade(&changes);
+        CacheReload { _guard: guard, changes }
+    }
+
     pub(crate) fn service_account_mutation_lock(&self) -> &AsyncMutex<()> {
         &self.service_account_mutation_lock
     }
 
-    pub(crate) fn sts_account_mutation_lock(&self, access_key: &str) -> Arc<AsyncMutex<StsMutationLockState>> {
-        let mut locks = self
-            .sts_account_mutation_locks
-            .locks
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
-        if let Some(lock) = locks.get(access_key).and_then(Weak::upgrade) {
-            return lock;
-        }
+    pub(crate) fn sts_account_mutation_lock(&self, access_key: &str) -> Arc<AsyncMutex<MutationLockState>> {
+        self.sts_account_mutation_locks.lock_for(access_key)
+    }
 
-        let registry = Arc::downgrade(&self.sts_account_mutation_locks);
-        let access_key_owned = access_key.to_string();
-        let lock = Arc::new_cyclic(|lock| {
-            AsyncMutex::new(StsMutationLockState {
-                access_key: access_key_owned,
-                registry,
-                lock: lock.clone(),
-            })
-        });
-        locks.insert(access_key.to_string(), Arc::downgrade(&lock));
-        lock
+    /// Orders derived-credential publication with changes to its regular parent.
+    pub(crate) fn parent_mutation_lock(&self, access_key: &str) -> Arc<AsyncMutex<MutationLockState>> {
+        self.parent_mutation_locks.lock_for(access_key)
     }
 
     pub(crate) fn snapshot(&self) -> CacheSnapshot {
@@ -142,7 +204,13 @@ impl Cache {
         let current = self.state.load_full();
         let mut locked = LockedCache {
             state: CacheState::clone(&current),
+            #[cfg(test)]
             current_ptr: Arc::as_ptr(&current),
+            reload_changes: self
+                .reload_changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .upgrade(),
             dirty: false,
         };
         let ret = f(&mut locked);
@@ -215,7 +283,9 @@ impl Cache {
 
 pub(crate) struct LockedCache {
     state: CacheState,
+    #[cfg(test)]
     current_ptr: *const CacheState,
+    reload_changes: Option<Arc<Mutex<ReloadChanges>>>,
     dirty: bool,
 }
 
@@ -224,11 +294,19 @@ impl LockedCache {
         &self.state
     }
 
+    #[cfg(test)]
     pub(crate) fn matches_snapshot(&self, snapshot: &CacheSnapshot) -> bool {
-        ptr::eq(self.current_ptr, Arc::as_ptr(snapshot))
+        std::ptr::eq(self.current_ptr, Arc::as_ptr(snapshot))
     }
 
-    fn exec<T: Clone>(target: &mut Arc<CacheEntity<T>>, t: OffsetDateTime, mut op: impl FnMut(&mut CacheEntity<T>)) -> bool {
+    fn exec<T: Clone>(
+        target: &mut Arc<CacheEntity<T>>,
+        t: OffsetDateTime,
+        changes: Option<&Arc<Mutex<ReloadChanges>>>,
+        partition: CachePartition,
+        key: &str,
+        mut op: impl FnMut(&mut CacheEntity<T>),
+    ) -> bool {
         if target.load_time >= t {
             return false;
         }
@@ -236,6 +314,14 @@ impl LockedCache {
         let mut new = CacheEntity::clone(target);
         op(&mut new);
         *target = Arc::new(new);
+        if let Some(changes) = changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(partition)
+                .or_default()
+                .insert(key.to_string());
+        }
         true
     }
 
@@ -244,139 +330,315 @@ impl LockedCache {
     }
 
     pub(crate) fn replace_policy_docs(&mut self, value: CacheEntity<PolicyDoc>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::PolicyDocs)
+                .or_default()
+                .extend(self.state.policy_docs.keys().chain(value.keys()).cloned());
+        }
         self.state.policy_docs = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn replace_users(&mut self, value: CacheEntity<UserIdentity>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::Users)
+                .or_default()
+                .extend(self.state.users.keys().chain(value.keys()).cloned());
+        }
         self.state.users = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn replace_user_policies(&mut self, value: CacheEntity<MappedPolicy>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::UserPolicies)
+                .or_default()
+                .extend(self.state.user_policies.keys().chain(value.keys()).cloned());
+        }
         self.state.user_policies = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn replace_sts_accounts(&mut self, value: CacheEntity<UserIdentity>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::StsAccounts)
+                .or_default()
+                .extend(self.state.sts_accounts.keys().chain(value.keys()).cloned());
+        }
         self.state.sts_accounts = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn replace_sts_policies(&mut self, value: CacheEntity<MappedPolicy>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::StsPolicies)
+                .or_default()
+                .extend(self.state.sts_policies.keys().chain(value.keys()).cloned());
+        }
         self.state.sts_policies = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn replace_groups(&mut self, value: CacheEntity<GroupInfo>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::Groups)
+                .or_default()
+                .extend(self.state.groups.keys().chain(value.keys()).cloned());
+        }
         self.state.groups = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn replace_group_policies(&mut self, value: CacheEntity<MappedPolicy>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::GroupPolicies)
+                .or_default()
+                .extend(self.state.group_policies.keys().chain(value.keys()).cloned());
+        }
         self.state.group_policies = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn replace_user_group_memberships(&mut self, value: CacheEntity<HashSet<String>>) {
+        if let Some(changes) = &self.reload_changes {
+            changes
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .entry(CachePartition::UserGroupMemberships)
+                .or_default()
+                .extend(self.state.user_group_memberships.keys().chain(value.keys()).cloned());
+        }
         self.state.user_group_memberships = Self::replaced(value);
         self.dirty = true;
     }
 
     pub(crate) fn add_or_update_policy_doc(&mut self, key: &str, value: &PolicyDoc, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.policy_docs, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.policy_docs,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::PolicyDocs,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn add_or_update_user(&mut self, key: &str, value: &UserIdentity, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.users, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.users,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::Users,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn add_or_update_user_policy(&mut self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.user_policies, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.user_policies,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::UserPolicies,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn add_or_update_sts_account(&mut self, key: &str, value: &UserIdentity, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.sts_accounts, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.sts_accounts,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::StsAccounts,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn add_or_update_sts_policy(&mut self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.sts_policies, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.sts_policies,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::StsPolicies,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn add_or_update_group(&mut self, key: &str, value: &GroupInfo, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.groups, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.groups,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::Groups,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn add_or_update_user_group_membership(&mut self, key: &str, value: &HashSet<String>, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.user_group_memberships, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.user_group_memberships,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::UserGroupMemberships,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn add_or_update_group_policy(&mut self, key: &str, value: &MappedPolicy, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.group_policies, t, |map| {
-            map.insert(key.to_string(), value.clone());
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.group_policies,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::GroupPolicies,
+            key,
+            |map| {
+                map.insert(key.to_string(), value.clone());
+            },
+        );
     }
 
     pub(crate) fn delete_policy_doc(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.policy_docs, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.policy_docs,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::PolicyDocs,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn delete_user(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.users, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.users,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::Users,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn delete_user_policy(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.user_policies, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.user_policies,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::UserPolicies,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn delete_user_group_membership(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.user_group_memberships, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.user_group_memberships,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::UserGroupMemberships,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn delete_sts_account(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.sts_accounts, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.sts_accounts,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::StsAccounts,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn delete_sts_policy(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.sts_policies, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.sts_policies,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::StsPolicies,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn delete_group(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.groups, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.groups,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::Groups,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn delete_group_policy(&mut self, key: &str, t: OffsetDateTime) {
-        self.dirty |= Self::exec(&mut self.state.group_policies, t, |map| {
-            map.remove(key);
-        });
+        self.dirty |= Self::exec(
+            &mut self.state.group_policies,
+            t,
+            self.reload_changes.as_ref(),
+            CachePartition::GroupPolicies,
+            key,
+            |map| {
+                map.remove(key);
+            },
+        );
     }
 
     pub(crate) fn build_user_group_memberships(&mut self) {
@@ -494,6 +756,50 @@ mod tests {
     use crate::cache::Cache;
     use crate::store::MappedPolicy;
 
+    #[tokio::test]
+    async fn reload_journal_preserves_same_key_mutations_and_expires_on_cancellation() {
+        use super::{CacheEntity, CachePartition};
+        let cache = Cache::default();
+        let stale = UserIdentity {
+            version: 1,
+            ..Default::default()
+        };
+        let current = UserIdentity {
+            version: 2,
+            ..Default::default()
+        };
+        let reload = cache.begin_reload().await;
+        cache.add_or_update_user("updated", &current, OffsetDateTime::now_utc());
+        cache.add_or_update_user("removed", &current, OffsetDateTime::now_utc());
+        cache.delete_user("removed", OffsetDateTime::now_utc());
+        // Even an absent key's deletion is recorded: storage may have returned
+        // an entry created and deleted entirely during the reload's reads.
+        cache.delete_user("absent", OffsetDateTime::now_utc());
+        let mut loaded = CacheEntity::new(std::collections::HashMap::from([
+            ("updated".to_string(), stale.clone()),
+            ("removed".to_string(), stale.clone()),
+            ("absent".to_string(), stale.clone()),
+            ("disk-only".to_string(), stale),
+        ]));
+        cache.with_write_lock(|locked| {
+            reload.merge_partition(CachePartition::Users, &mut loaded, &locked.state().users);
+            locked.replace_users(loaded);
+        });
+        let snapshot = cache.snapshot();
+        assert_eq!(snapshot.users["updated"].version, 2);
+        assert!(!snapshot.users.contains_key("removed"));
+        assert!(!snapshot.users.contains_key("absent"));
+        assert!(snapshot.users.contains_key("disk-only"));
+        drop(reload);
+        assert!(cache.reload_changes.lock().expect("journal lock").upgrade().is_none());
+        let cancelled = cache.begin_reload().await;
+        cache.delete_user("disk-only", OffsetDateTime::now_utc());
+        drop(cancelled);
+        assert!(cache.reload_changes.lock().expect("journal lock").upgrade().is_none());
+        let next = cache.begin_reload().await;
+        assert!(next.changes.lock().expect("new journal").is_empty());
+    }
+
     #[test]
     fn sts_mutation_locks_are_keyed_and_prune_unused_entries() {
         let cache = Cache::default();

crates/iam/src/federation/mapper.rs

@@ -16,6 +16,7 @@ use super::{FederatedClaims, VerifiedFederatedIdentity};
 use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
 use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
 use rustfs_utils::HashAlgorithm;
+use serde_json::Value;
 use std::{collections::HashMap, sync::Arc};
 use tracing::debug;
 
@@ -66,6 +67,15 @@ impl FederatedAuthorization {
         &self.provider_id
     }
 
+    /// Issuer from the verified upstream identity.
+    pub fn issuer(&self) -> Option<&str> {
+        self.claims
+            .raw
+            .get("iss")
+            .and_then(Value::as_str)
+            .filter(|issuer| !issuer.is_empty())
+    }
+
     pub fn subject(&self) -> &str {
         &self.claims.sub
     }

crates/iam/src/manager.rs

@@ -415,6 +415,16 @@ where
 
         self.api.save_iam_config(IAMFormat::new_version_1(), path).await
     }
+    /// Current built-in IAM group membership, independent of credential snapshots.
+    pub fn user_group_memberships(&self, name: &str) -> Vec<String> {
+        self.cache
+            .snapshot()
+            .user_group_memberships
+            .get(name)
+            .map(|groups| groups.iter().cloned().collect())
+            .unwrap_or_default()
+    }
+
     pub async fn get_user(&self, access_key: &str) -> Option<UserIdentity> {
         let cache = self.cache.snapshot();
         cache
@@ -1005,6 +1015,17 @@ where
     }
 
     pub async fn sts_policy_db_get(&self, name: &str, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
+        self.sts_policy_db_get_with_identity_mapping(name, groups)
+            .await
+            .map(|(policies, _)| policies)
+    }
+
+    /// Resolve STS policies and whether that same lookup found a nonempty identity mapping.
+    pub async fn sts_policy_db_get_with_identity_mapping(
+        &self,
+        name: &str,
+        groups: &Option<Vec<String>>,
+    ) -> Result<(Vec<String>, bool)> {
         if name.is_empty() {
             return Err(Error::InvalidArgument);
         }
@@ -1031,7 +1052,10 @@ where
             }
         };
 
-        self.policy_db_get_with_groups(mapped_policy.to_slice(), groups).await
+        let direct_policies = mapped_policy.to_slice();
+        let has_identity_mapping = !direct_policies.is_empty();
+        let policies = self.policy_db_get_with_groups(direct_policies, groups).await?;
+        Ok((policies, has_identity_mapping))
     }
 
     async fn policy_db_get_with_groups(&self, mut policies: Vec<String>, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
@@ -1158,7 +1182,7 @@ where
         if let Some(groups) = u.credentials.groups.as_ref() {
             for group in groups.iter() {
                 if groups_cache.get(group).filter(|v| v.status == STATUS_DISABLED).is_some() {
-                    return Ok((Vec::new(), OffsetDateTime::now_utc()));
+                    continue;
                 }
 
                 let mp = match group_policies.get(group) {
@@ -1189,7 +1213,7 @@ where
 
         for group in user_group_memberships.get(name).cloned().unwrap_or_default().iter() {
             if groups_cache.get(group).filter(|v| v.status == STATUS_DISABLED).is_some() {
-                return Ok((Vec::new(), OffsetDateTime::now_utc()));
+                continue;
             }
 
             let mp = match group_policies.get(group) {
@@ -1486,7 +1510,27 @@ where
         args: &AddOrUpdateUserReq,
         updated_at: OffsetDateTime,
     ) -> Result<OffsetDateTime> {
+        self.add_user_at_with_secret_change(access_key, args, updated_at)
+            .await
+            .map(|(updated_at, _, _guard)| updated_at)
+    }
+
+    /// Return whether this write replaced a different regular-user secret.
+    pub(crate) async fn add_user_at_with_secret_change(
+        &self,
+        access_key: &str,
+        args: &AddOrUpdateUserReq,
+        updated_at: OffsetDateTime,
+    ) -> Result<(OffsetDateTime, bool, tokio::sync::OwnedMutexGuard<crate::cache::MutationLockState>)> {
+        let parent_lock = self.cache.parent_mutation_lock(access_key);
+        let parent_guard = parent_lock.lock_owned().await;
+        let _service_guard = self.cache.service_account_mutation_lock().lock().await;
+        let sts_lock = self.cache.sts_account_mutation_lock(access_key);
+        let _sts_guard = sts_lock.lock().await;
         let cache = self.cache.snapshot();
+        if cache.sts_accounts.contains_key(access_key) {
+            return Err(Error::AccessKeyAlreadyExists);
+        }
         let users = Arc::clone(&cache.users);
         if let Some(x) = users.get(access_key) {
             debug!(
@@ -1496,10 +1540,13 @@ where
                 state = "existing",
                 "Updating existing IAM user"
             );
-            if x.credentials.is_temp() {
+            if x.credentials.is_temp() || x.credentials.is_service_account() {
                 return Err(Error::AccessKeyAlreadyExists);
             }
         }
+        let secret_changed = users
+            .get(access_key)
+            .is_some_and(|user| user.credentials.secret_key != args.secret_key);
         drop(cache);
         drop(users);
 
@@ -1523,7 +1570,7 @@ where
 
         self.update_user_with_claims(access_key, user_entry)?;
 
-        Ok(updated_at)
+        Ok((updated_at, secret_changed, parent_guard))
     }
 
     pub async fn delete_user(&self, access_key: &str, utype: UserType) -> Result<()> {
@@ -1540,12 +1587,17 @@ where
             return Err(Error::InvalidArgument);
         }
 
+        let parent_lock = (utype == UserType::Reg).then(|| self.cache.parent_mutation_lock(access_key));
+        let _parent_guard = match &parent_lock {
+            Some(lock) => Some(lock.lock().await),
+            None => None,
+        };
         let sts_mutation_lock = (utype == UserType::Sts).then(|| self.cache.sts_account_mutation_lock(access_key));
         let _sts_mutation_guard = match &sts_mutation_lock {
             Some(lock) => Some(lock.lock().await),
             None => None,
         };
-        let _service_account_guard = if utype == UserType::Svc {
+        let _service_account_guard = if matches!(utype, UserType::Svc | UserType::Reg) {
             Some(self.cache.service_account_mutation_lock().lock().await)
         } else {
             None
@@ -1640,17 +1692,46 @@ where
     }
 
     pub async fn update_user_secret_key(&self, access_key: &str, secret_key: &str) -> Result<(OffsetDateTime, AccountStatus)> {
+        self.update_user_secret_key_with_guard(access_key, secret_key, None)
+            .await
+            .map(|(updated_at, status, _guard)| (updated_at, status))
+    }
+
+    pub(crate) async fn update_user_secret_key_with_guard(
+        &self,
+        access_key: &str,
+        secret_key: &str,
+        expected_secret: Option<&str>,
+    ) -> Result<(
+        OffsetDateTime,
+        AccountStatus,
+        tokio::sync::OwnedMutexGuard<crate::cache::MutationLockState>,
+    )> {
         if access_key.is_empty() || secret_key.is_empty() {
             return Err(Error::InvalidArgument);
         }
 
+        let parent_lock = self.cache.parent_mutation_lock(access_key);
+        let parent_guard = parent_lock.lock_owned().await;
         let cache = self.cache.snapshot();
         let users = Arc::clone(&cache.users);
         let u = match users.get(access_key) {
             Some(u) => u,
+            None if expected_secret.is_some() => return Err(Error::IAMActionNotAllowed),
             None => return Err(Error::NoSuchUser(access_key.to_string())),
         };
 
+        if let Some(expected) = expected_secret {
+            use subtle::ConstantTimeEq as _;
+            if !u.credentials.is_valid()
+                || u.credentials.is_temp()
+                || u.credentials.is_service_account()
+                || !bool::from(u.credentials.secret_key.as_bytes().ct_eq(expected.as_bytes()))
+            {
+                return Err(Error::IAMActionNotAllowed);
+            }
+        }
+
         let mut cred = u.credentials.clone();
         cred.secret_key = secret_key.to_string();
 
@@ -1672,7 +1753,7 @@ where
             .await?;
 
         self.update_user_with_claims(access_key, u)?;
-        Ok((updated_at, status))
+        Ok((updated_at, status, parent_guard))
     }
 
     /// Add SSH public key for a user (for SFTP authentication)
@@ -1681,6 +1762,8 @@ where
             return Err(Error::InvalidArgument);
         }
 
+        let parent_lock = self.cache.parent_mutation_lock(access_key);
+        let _parent_guard = parent_lock.lock().await;
         let cache = self.cache.snapshot();
         let users = Arc::clone(&cache.users);
         let u = match users.get(access_key) {
@@ -1720,6 +1803,8 @@ where
             return Err(Error::InvalidArgument);
         }
 
+        let parent_lock = self.cache.parent_mutation_lock(access_key);
+        let _parent_guard = parent_lock.lock().await;
         let cache = self.cache.snapshot();
         let users = Arc::clone(&cache.users);
         let u = match users.get(access_key) {
@@ -2238,6 +2323,11 @@ where
         Ok(())
     }
     pub async fn user_notification_handler(&self, name: &str, user_type: UserType) -> Result<()> {
+        let parent_lock = (user_type == UserType::Reg).then(|| self.cache.parent_mutation_lock(name));
+        let _parent_guard = match &parent_lock {
+            Some(lock) => Some(lock.lock().await),
+            None => None,
+        };
         let sts_mutation_lock = (user_type == UserType::Sts).then(|| self.cache.sts_account_mutation_lock(name));
         let _sts_mutation_guard = match &sts_mutation_lock {
             Some(lock) => Some(lock.lock().await),
@@ -2985,6 +3075,58 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn disabled_group_preserves_direct_and_other_group_policies() {
+        for credential_groups in [false, true] {
+            let iam = build_test_iam_cache(FailingInitialLoadStore);
+            let groups = vec!["disabled".to_string(), "enabled".to_string()];
+            let user = UserIdentity::new(Credentials {
+                access_key: "alice".to_string(),
+                secret_key: "alice-secret".to_string(),
+                status: STATUS_ENABLED.to_string(),
+                groups: credential_groups.then(|| groups.clone()),
+                ..Default::default()
+            });
+            let now = OffsetDateTime::now_utc();
+            iam.cache.add_or_update_user("alice", &user, now);
+            iam.cache
+                .add_or_update_user_policy("alice", &MappedPolicy::new("direct-deny"), now);
+            if !credential_groups {
+                iam.cache
+                    .add_or_update_user_group_membership("alice", &groups.iter().cloned().collect(), now);
+            }
+            for (group, status, policy) in [
+                ("disabled", STATUS_DISABLED, "disabled-grant"),
+                ("enabled", STATUS_ENABLED, "other-group-deny"),
+            ] {
+                iam.cache.add_or_update_group(
+                    group,
+                    &GroupInfo {
+                        status: status.to_string(),
+                        members: vec!["alice".to_string()],
+                        ..Default::default()
+                    },
+                    now,
+                );
+                iam.cache.add_or_update_group_policy(group, &MappedPolicy::new(policy), now);
+            }
+            let policies: HashSet<_> = iam
+                .policy_db_get("alice", &None)
+                .await
+                .expect("user policy lookup")
+                .into_iter()
+                .collect();
+            assert_eq!(policies, HashSet::from(["direct-deny".to_string(), "other-group-deny".to_string()]));
+            assert!(
+                iam.policy_db_get_internal("disabled", true, false)
+                    .await
+                    .expect("disabled group lookup")
+                    .0
+                    .is_empty()
+            );
+        }
+    }
+
     #[tokio::test]
     async fn sts_policy_lookup_ignores_colliding_regular_user_mapping() {
         let iam = build_test_iam_cache(FailingInitialLoadStore);

crates/iam/src/oidc/provider.rs

@@ -18,18 +18,19 @@ use super::{
 use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
 use openidconnect::core::{CoreIdToken, CoreIdTokenClaims, CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
 use openidconnect::{
-    AsyncHttpClient, ClaimsVerificationError, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, NonceVerifier,
+    AsyncHttpClient, ClaimsVerificationError, ClientId, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, NonceVerifier,
     ProviderMetadataWithLogout, SignatureVerificationError, TokenUrl,
 };
 use serde::Deserialize;
 use std::collections::HashMap;
-use std::sync::RwLock;
+use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration as StdDuration, Instant};
 use tokio::time::sleep;
 use tracing::{debug, warn};
 use url::Url;
 
 pub(super) const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60);
+const OIDC_JWKS_REFRESH_COOLDOWN: StdDuration = StdDuration::from_secs(60);
 pub(super) const OIDC_DISCOVERY_TRANSPORT_RETRIES: usize = 3;
 pub(super) const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration::from_millis(50);
 pub(super) const OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY: &str = "JWKS request blocked by outbound policy";
@@ -97,9 +98,14 @@ pub(super) struct ProviderState {
     pub(super) discovered_at: Instant,
 }
 
+type RefreshAttempt = Option<(Instant, Result<ProviderState, String>)>;
+type ProviderRefreshLock = Arc<tokio::sync::Mutex<RefreshAttempt>>;
+
+#[derive(Clone)]
 pub(super) struct ProviderRuntime {
-    states: RwLock<HashMap<String, ProviderState>>,
-    http_client: ReqwestHttpClient,
+    states: Arc<RwLock<HashMap<String, ProviderState>>>,
+    http_client: Arc<ReqwestHttpClient>,
+    refreshes: Arc<Mutex<HashMap<String, ProviderRefreshLock>>>,
 }
 
 // Workload issuers do not implement the browser authorization flow. Keep their
@@ -200,16 +206,25 @@ impl DiscoveredProviderMetadata {
 
     pub(super) fn verifier(&self, config: &OidcProviderConfig) -> CoreIdTokenVerifier<'static> {
         let client_id = ClientId::new(config.client_id.clone());
-        let secret = config.client_secret.as_ref().map(|secret| ClientSecret::new(secret.clone()));
         let (issuer, jwks, algorithms) = match self {
             Self::Console(metadata) => (metadata.issuer(), metadata.jwks(), metadata.id_token_signing_alg_values_supported()),
             Self::Workload(metadata) => (&metadata.issuer, &metadata.jwks, &metadata.id_token_signing_alg_values_supported),
         };
-        let verifier = match secret {
-            Some(secret) => CoreIdTokenVerifier::new_confidential_client(client_id, secret, issuer.clone(), jwks.clone()),
-            None => CoreIdTokenVerifier::new_public_client(client_id, issuer.clone(), jwks.clone()),
-        };
-        verifier.set_allowed_algs(algorithms.clone())
+        // Caller-supplied tokens must prove possession of the issuer's signing
+        // key. A client secret is shared with clients and cannot provide that proof.
+        let algorithms = algorithms
+            .iter()
+            .filter(|algorithm| {
+                !matches!(
+                    algorithm,
+                    CoreJwsSigningAlgorithm::HmacSha256
+                        | CoreJwsSigningAlgorithm::HmacSha384
+                        | CoreJwsSigningAlgorithm::HmacSha512
+                )
+            })
+            .cloned()
+            .collect::<Vec<_>>();
+        CoreIdTokenVerifier::new_public_client(client_id, issuer.clone(), jwks.clone()).set_allowed_algs(algorithms)
     }
 }
 
@@ -273,8 +288,9 @@ impl ProviderState {
 impl ProviderRuntime {
     pub(super) fn new(http_client: ReqwestHttpClient, states: HashMap<String, ProviderState>) -> Self {
         Self {
-            states: RwLock::new(states),
-            http_client,
+            states: Arc::new(RwLock::new(states)),
+            http_client: Arc::new(http_client),
+            refreshes: Arc::new(Mutex::new(HashMap::new())),
         }
     }
 
@@ -326,9 +342,38 @@ impl ProviderRuntime {
         provider_id: &str,
         config: &OidcProviderConfig,
     ) -> Result<ProviderState, String> {
-        self.recover_provider_state(provider_id, config)
+        let refresh = self
+            .refreshes
+            .lock()
+            .map_err(|_| "provider refresh registry is unavailable".to_string())?
+            .entry(provider_id.to_string())
+            .or_default()
+            .clone();
+        let mut attempt = refresh.lock_owned().await;
+        if let Some((completed, result)) = attempt.as_ref()
+            && completed.elapsed() < OIDC_JWKS_REFRESH_COOLDOWN
+        {
+            return result.clone();
+        }
+
+        // The admitted fetch owns the per-provider lock. A disconnected request
+        // cannot abandon discovery while consuming the provider's cooldown.
+        // Waiting callers share its completed result, including genuine failures.
+        let runtime = tokio::runtime::Handle::try_current().map_err(|err| err.to_string())?;
+        let provider_runtime = self.clone();
+        let provider_id = provider_id.to_string();
+        let config = config.clone();
+        runtime
+            .spawn(async move {
+                let result = provider_runtime
+                    .recover_provider_state(&provider_id, &config)
+                    .await
+                    .map_err(|err| err.to_string());
+                *attempt = Some((Instant::now(), result.clone()));
+                result
+            })
             .await
-            .map_err(|err| err.to_string())
+            .map_err(|err| err.to_string())?
     }
 
     pub(super) async fn recover_provider_state(

crates/iam/src/oidc/provider_tests.rs

@@ -180,7 +180,7 @@ fn workload_metadata_rejects_duplicate_fields() {
     }
 }
 #[test]
-fn workload_verifier_preserves_algorithm_secret_and_audience_policy() {
+fn caller_supplied_tokens_cannot_be_signed_with_the_client_secret() {
     let secret = Nonce::new_random().secret().to_string();
     let mut config = build_mocked_oidc_provider_config("workload", "https://issuer.example.com");
     config.client_secret = Some(secret.clone());
@@ -195,7 +195,8 @@ fn workload_verifier_preserves_algorithm_secret_and_audience_policy() {
     let token: CoreIdToken = signed.parse().unwrap();
     for workload in [false, true] {
         for (algorithms, accepted) in [
-            (serde_json::json!(["HS256", "unsupported-future-algorithm"]), true),
+            (serde_json::json!(["RS256", "HS256"]), false),
+            (serde_json::json!(["HS256", "unsupported-future-algorithm"]), false),
             (serde_json::json!(["ES256"]), false),
             (serde_json::json!(["unsupported-future-algorithm"]), false),
         ] {
@@ -216,21 +217,6 @@ fn workload_verifier_preserves_algorithm_secret_and_audience_policy() {
                 accepted,
                 "workload={workload}, algorithms={algorithms}"
             );
-            if accepted {
-                assert!(
-                    token.claims(&metadata.verifier(&config), |_: Option<&Nonce>| Ok(())).is_err(),
-                    "additional audiences require explicit trust"
-                );
-                let mut wrong_secret = config.clone();
-                wrong_secret.client_secret = Some(Nonce::new_random().secret().to_string());
-                let verifier = metadata
-                    .verifier(&wrong_secret)
-                    .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-                assert!(
-                    token.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_err(),
-                    "incorrect client secret must fail"
-                );
-            }
         }
     }
 }
@@ -274,6 +260,19 @@ fn configured_verifier_rejects_malformed_hmac_signature_lengths() {
             let metadata = DiscoveredProviderMetadata::parse(&serde_json::to_vec(&document).expect("encode metadata"), true)
                 .expect("parse configured provider metadata");
             let verifier = metadata.verifier(&config);
+            assert!(
+                verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_err(),
+                "caller-supplied HMAC tokens must be rejected"
+            );
+            // The authorization-code flow obtains tokens directly from the
+            // provider and still supports confidential-client verification.
+            let verifier = openidconnect::core::CoreIdTokenVerifier::new_confidential_client(
+                openidconnect::ClientId::new(config.client_id.clone()),
+                openidconnect::ClientSecret::new(secret.to_owned()),
+                metadata.issuer().clone(),
+                openidconnect::core::CoreJsonWebKeySet::new(vec![]),
+            )
+            .set_allowed_algs(vec![serde_json::from_value(serde_json::json!(name)).expect("parse signing algorithm")]);
             assert!(
                 verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
                 "valid {name} token must pass for workload={workload}"
@@ -780,3 +779,149 @@ async fn test_validate_oidc_provider_config_returns_detailed_errors() {
     assert!(err.contains(&format!("issuer '{base}/application/o/rustfs/'")));
     assert!(handle.join().is_ok());
 }
+
+#[tokio::test]
+async fn jwks_refresh_is_single_flight_and_survives_request_cancellation() {
+    let (initial_key, initial_jwk) = oidc_es256_key_and_jwk("initial");
+    let (rotated_key, rotated_jwk) = oidc_es256_key_and_jwk("rotated");
+    let (wrong_key, _) = oidc_es256_key_and_jwk("wrong");
+    let requests = Arc::new(AtomicUsize::new(0));
+    let started = Arc::new(tokio::sync::Notify::new());
+    let (release_tx, release_rx) = std::sync::mpsc::channel();
+    let server_requests = Arc::clone(&requests);
+    let server_started = Arc::clone(&started);
+    let Some((base, server)) = start_mock_oidc_discovery_server_with_jwks(
+        |base| (base.to_string(), format!("{base}/jwks"), "/jwks".to_string()),
+        6,
+        "ES256",
+        false,
+        move |fetch| {
+            let request = server_requests.fetch_add(1, Ordering::SeqCst);
+            if request == 2 {
+                server_started.notify_one();
+                release_rx
+                    .recv_timeout(StdDuration::from_secs(5))
+                    .expect("release refresh response");
+            }
+            serde_json::json!({"keys": [if fetch == 0 { &initial_jwk } else { &rotated_jwk }]}).to_string()
+        },
+    ) else {
+        return;
+    };
+    let mut config = build_mocked_oidc_provider_config("bounded-refresh", &base);
+    config.issuer = Some(base.clone());
+    let http = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).expect("allow test origin"));
+    let state = discover_provider(&config, &http).await.expect("initial discovery");
+    let sys = Arc::new(OidcSys {
+        configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]),
+        provider_runtime: ProviderRuntime::new(http, HashMap::from([(config.id.clone(), state)])),
+        state_store: OidcStateStore::new(),
+        pending_discovery: Default::default(),
+    });
+    let now = time::OffsetDateTime::now_utc().unix_timestamp();
+    let payload = serde_json::json!({"iss": base, "sub": "user", "aud": config.client_id, "iat": now, "exp": now + 300});
+    let sign = |kid: &str, claims: &serde_json::Value, key: &EncodingKey| {
+        let mut header = Header::new(Algorithm::ES256);
+        header.kid = Some(kid.to_string());
+        jsonwebtoken::encode(&header, claims, key).expect("sign test token")
+    };
+    for (claim, value) in [
+        ("aud", serde_json::json!("wrong-audience")),
+        ("exp", serde_json::json!(now - 300)),
+    ] {
+        let mut invalid = payload.clone();
+        invalid[claim] = value;
+        assert!(
+            sys.verify_web_identity_token(&sign("initial", &invalid, &initial_key))
+                .await
+                .is_err()
+        );
+    }
+    assert_eq!(requests.load(Ordering::SeqCst), 2, "claim failures cannot refresh signing keys");
+    let invalid = sign("rotated", &payload, &wrong_key);
+    let attack = {
+        let sys = Arc::clone(&sys);
+        let invalid = invalid.clone();
+        tokio::spawn(async move { sys.verify_web_identity_token(&invalid).await })
+    };
+    tokio::time::timeout(StdDuration::from_secs(5), started.notified())
+        .await
+        .expect("refresh started");
+    attack.abort();
+    assert!(attack.await.expect_err("request canceled").is_cancelled());
+    let mut retries = Vec::new();
+    for _ in 0..16 {
+        let sys = Arc::clone(&sys);
+        let invalid = invalid.clone();
+        retries.push(tokio::spawn(async move { sys.verify_web_identity_token(&invalid).await }));
+    }
+    release_tx.send(()).expect("release provider");
+    for retry in retries {
+        assert!(
+            tokio::time::timeout(StdDuration::from_secs(5), retry)
+                .await
+                .expect("retry timeout")
+                .expect("retry join")
+                .is_err()
+        );
+    }
+    let rotated = sign("rotated", &payload, &rotated_key);
+    sys.verify_web_identity_token(&rotated)
+        .await
+        .expect("completed refresh must accept rotated keys after the original caller left");
+    assert_eq!(requests.load(Ordering::SeqCst), 4, "one discovery/JWKS pair for all rejected tokens");
+    sys.provider_runtime
+        .refresh_provider_state("independent-provider", &config)
+        .await
+        .expect("provider cooldowns must be independent");
+    assert_eq!(requests.load(Ordering::SeqCst), 6);
+    server.join().expect("mock server");
+}
+
+#[tokio::test]
+async fn jwks_refresh_cools_down_failures_and_recovers_after_the_interval() {
+    let (_, initial_jwk) = oidc_es256_key_and_jwk("initial");
+    let (_, recovered_jwk) = oidc_es256_key_and_jwk("recovered");
+    let requests = Arc::new(AtomicUsize::new(0));
+    let server_requests = Arc::clone(&requests);
+    let Some((base, server)) = start_mock_oidc_discovery_server_with_jwks(
+        |base| (base.to_string(), format!("{base}/jwks"), "/jwks".to_string()),
+        6,
+        "ES256",
+        false,
+        move |fetch| {
+            server_requests.fetch_add(1, Ordering::SeqCst);
+            match fetch {
+                0 => serde_json::json!({"keys": [initial_jwk]}).to_string(),
+                1 => "invalid JWKS".to_string(),
+                _ => serde_json::json!({"keys": [recovered_jwk]}).to_string(),
+            }
+        },
+    ) else {
+        return;
+    };
+    let mut config = build_mocked_oidc_provider_config("recovering-refresh", &base);
+    config.issuer = Some(base.clone());
+    let http = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).expect("allow test origin"));
+    let initial = discover_provider(&config, &http).await.expect("initial discovery");
+    let runtime = ProviderRuntime::new(http, HashMap::from([(config.id.clone(), initial.clone())]));
+    assert!(runtime.refresh_provider_state(&config.id, &config).await.is_err());
+    assert_eq!(requests.load(Ordering::SeqCst), 4);
+    for _ in 0..8 {
+        assert!(runtime.refresh_provider_state(&config.id, &config).await.is_err());
+    }
+    assert_eq!(requests.load(Ordering::SeqCst), 4, "completed failures also consume a bounded cooldown");
+    assert_eq!(
+        runtime.get_provider_state(&config.id).expect("cached keys").discovered_at,
+        initial.discovered_at
+    );
+    let refresh = runtime.refreshes.lock().expect("registry")[&config.id].clone();
+    refresh.lock().await.as_mut().expect("completed attempt").0 = Instant::now() - OIDC_JWKS_REFRESH_COOLDOWN;
+    runtime
+        .refresh_provider_state(&config.id, &config)
+        .await
+        .expect("retry after cooldown");
+    assert_eq!(requests.load(Ordering::SeqCst), 6);
+    assert!(runtime.get_provider_state(&config.id).expect("recovered keys").discovered_at > initial.discovered_at);
+    server.join().expect("mock server");
+}

crates/iam/src/oidc/runtime.rs

@@ -33,8 +33,8 @@ use crate::{
 };
 use openidconnect::core::{CoreAuthenticationFlow, CoreClient, CoreIdToken};
 use openidconnect::{
-    Audience, AuthType, AuthorizationCode, ClientId, ClientSecret, CsrfToken, LogoutRequest, Nonce, PkceCodeChallenge,
-    PkceCodeVerifier, PostLogoutRedirectUrl, RedirectUrl, RequestTokenError, Scope,
+    Audience, AuthType, AuthorizationCode, ClaimsVerificationError, ClientId, ClientSecret, CsrfToken, LogoutRequest, Nonce,
+    PkceCodeChallenge, PkceCodeVerifier, PostLogoutRedirectUrl, RedirectUrl, RequestTokenError, Scope,
 };
 use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
 use rustfs_utils::logging::LogThrottle;
@@ -537,6 +537,9 @@ impl OidcSys {
             .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
         let verified = verify_id_token_claims(id_token, &verifier, &Nonce::new(session.nonce.clone()));
         if let Err(e) = verified {
+            if !matches!(e, ClaimsVerificationError::SignatureVerification(_)) {
+                return Err(format!("ID token verification failed: {e}"));
+            }
             let verification_error = e.to_string();
             warn!(
                 event = EVENT_OIDC_DIAGNOSTICS,
@@ -767,6 +770,9 @@ impl OidcSys {
             .verifier(&config)
             .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
         if let Err(e) = verify_id_token_claims(&id_token, &verifier, |_: Option<&Nonce>| Ok(())) {
+            if !matches!(e, ClaimsVerificationError::SignatureVerification(_)) {
+                return Err(format!("ID token verification failed: {e}"));
+            }
             state = self
                 .provider_runtime
                 .refresh_provider_state(&provider_id, &config)

crates/iam/src/storage_api.rs

@@ -14,6 +14,20 @@
 
 use std::sync::Arc;
 
+pub(crate) fn is_unreadable_bucket_tags(error: &crate::error::Error) -> bool {
+    let crate::error::Error::Io(error) = error else {
+        return false;
+    };
+    let Some(storage_error) = error.get_ref().and_then(|inner| inner.downcast_ref::<IamStorageError>()) else {
+        return false;
+    };
+    rustfs_ecstore::api::bucket::metadata::unreadable_config_refusal(storage_error)
+        .is_some_and(|failure| failure.config_file == rustfs_ecstore::api::bucket::metadata::BUCKET_TAGGING_CONFIG)
+}
+
+#[cfg(test)]
+pub(crate) use rustfs_ecstore::api::bucket::metadata::UnreadableBucketConfig;
+
 pub(crate) use rustfs_ecstore::api::config::RUSTFS_CONFIG_PREFIX as IAM_CONFIG_ROOT_PREFIX;
 use rustfs_ecstore::api::config::com::{
     delete_config as ecstore_delete_config, read_config_no_lock as ecstore_read_config_no_lock,

crates/iam/src/store/object.rs

@@ -1154,7 +1154,7 @@ impl Store for ObjectStore {
         // node-counted lock quorum (rustfs#4304; startup contract rustfs#4056).
         const LOAD_ALL_MODE: LoadMode = LoadMode::BootstrapNoLock;
 
-        let cache_snapshot = cache.snapshot();
+        let reload = cache.begin_reload().await;
         let listed_config_items = self.list_all_iamconfig_items().await?;
 
         let mut policy_docs_cache = CacheEntity::new(get_default_policies());
@@ -1201,26 +1201,20 @@ impl Store for ObjectStore {
         }
 
         // groups
-        let mut groups_cache = None;
+        let mut groups_cache = CacheEntity::default();
         if let Some(item_name_list) = listed_config_items.get(GROUPS_LIST_KEY) {
-            let mut items_cache = CacheEntity::default();
-
             for item in item_name_list.iter() {
                 let name = rustfs_utils::path::dir(item);
                 debug!(group = %name, "IAM group loaded");
-                if let Err(err) = self.load_group_with(&name, &mut items_cache, LOAD_ALL_MODE).await {
+                if let Err(err) = self.load_group_with(&name, &mut groups_cache, LOAD_ALL_MODE).await {
                     return Err(Error::other(format!("load group failed: {err}")));
                 };
             }
-
-            groups_cache = Some(items_cache);
         }
 
         // user policies
-        let mut user_policies_cache = None;
+        let mut user_policies_cache = CacheEntity::default();
         if let Some(item_name_list) = listed_config_items.get(POLICY_DB_USERS_LIST_KEY) {
-            let mut items_cache = CacheEntity::default();
-
             // Load in fixed-size chunks so each mapping is fetched exactly once
             // (backlog#806 — the split_off loop was O(n^2)).
             for chunk in item_name_list.chunks(32) {
@@ -1235,32 +1229,26 @@ impl Store for ObjectStore {
 
                     let name = chunk[idx].trim_end_matches(".json").to_owned();
                     debug!(user = %name, "IAM user policy loaded");
-                    items_cache.insert(name, p);
+                    user_policies_cache.insert(name, p);
                 }
             }
-
-            user_policies_cache = Some(items_cache);
         }
 
         // group policy
-        let mut group_policies_cache = None;
+        let mut group_policies_cache = CacheEntity::default();
         if let Some(item_name_list) = listed_config_items.get(POLICY_DB_GROUPS_LIST_KEY) {
-            let mut items_cache = CacheEntity::default();
-
             for item in item_name_list.iter() {
                 let name = item.trim_end_matches(".json");
 
                 debug!(group = %name, "IAM group policy loaded");
                 if let Err(err) = self
-                    .load_mapped_policy_with(name, UserType::Reg, true, &mut items_cache, LOAD_ALL_MODE)
+                    .load_mapped_policy_with(name, UserType::Reg, true, &mut group_policies_cache, LOAD_ALL_MODE)
                     .await
                     && !is_err_no_such_policy(&err)
                 {
                     return Err(Error::other(format!("load group policy failed: {err}")));
                 };
             }
-
-            group_policies_cache = Some(items_cache);
         }
 
         let mut sts_policies_cache = CacheEntity::default();
@@ -1330,25 +1318,48 @@ impl Store for ObjectStore {
             }
         }
 
+        #[cfg(test)]
+        if let Ok(barrier) = LOAD_ALL_COMMIT_BARRIER.try_with(std::sync::Arc::clone) {
+            barrier.0.notify_one();
+            barrier.1.notified().await;
+        }
+
         cache.with_write_lock(|cache| {
-            if cache.matches_snapshot(&cache_snapshot) {
-                cache.replace_policy_docs(policy_docs_cache);
-                if let Some(groups_cache) = groups_cache {
-                    cache.replace_groups(groups_cache);
-                }
-                if let Some(user_policies_cache) = user_policies_cache {
-                    cache.replace_user_policies(user_policies_cache);
-                }
-                if let Some(group_policies_cache) = group_policies_cache {
-                    cache.replace_group_policies(group_policies_cache);
-                }
-                cache.replace_users(user_items_cache);
-                cache.replace_sts_accounts(sts_items_cache);
-                cache.replace_sts_policies(sts_policies_cache);
-                cache.build_user_group_memberships();
-            } else {
-                warn!("IAM full reload cache commit skipped due to concurrent cache changes");
-            }
+            reload.merge_partition(
+                crate::cache::CachePartition::PolicyDocs,
+                &mut policy_docs_cache,
+                &cache.state().policy_docs,
+            );
+            reload.merge_partition(crate::cache::CachePartition::Groups, &mut groups_cache, &cache.state().groups);
+            reload.merge_partition(
+                crate::cache::CachePartition::UserPolicies,
+                &mut user_policies_cache,
+                &cache.state().user_policies,
+            );
+            reload.merge_partition(
+                crate::cache::CachePartition::GroupPolicies,
+                &mut group_policies_cache,
+                &cache.state().group_policies,
+            );
+            reload.merge_partition(crate::cache::CachePartition::Users, &mut user_items_cache, &cache.state().users);
+            reload.merge_partition(
+                crate::cache::CachePartition::StsAccounts,
+                &mut sts_items_cache,
+                &cache.state().sts_accounts,
+            );
+            reload.merge_partition(
+                crate::cache::CachePartition::StsPolicies,
+                &mut sts_policies_cache,
+                &cache.state().sts_policies,
+            );
+            cache.replace_policy_docs(policy_docs_cache);
+            cache.replace_groups(groups_cache);
+            cache.replace_user_policies(user_policies_cache);
+            cache.replace_group_policies(group_policies_cache);
+            cache.replace_users(user_items_cache);
+            cache.replace_sts_accounts(sts_items_cache);
+            cache.replace_sts_policies(sts_policies_cache);
+            cache.build_user_group_memberships();
         });
 
         Ok(())
@@ -1363,6 +1374,11 @@ fn map_delete_user_identity_error(name: &str, err: Error) -> Error {
     }
 }
 
+#[cfg(test)]
+tokio::task_local! {
+    static LOAD_ALL_COMMIT_BARRIER: std::sync::Arc<(tokio::sync::Notify, tokio::sync::Notify)>;
+}
+
 #[cfg(test)]
 mod tests {
     use super::{DecryptSource, LoadMode, ObjectStore, map_delete_user_identity_error};
@@ -1372,6 +1388,99 @@ mod tests {
     use serial_test::serial;
     use temp_env::with_vars;
 
+    #[test]
+    #[serial]
+    fn test_load_all_clears_partitions_with_empty_prefix() {
+        use crate::cache::Cache;
+        use crate::store::{GroupInfo, MappedPolicy, Store, UserType};
+
+        std::thread::Builder::new()
+            .stack_size(8 * 1024 * 1024)
+            .spawn(|| {
+                let runtime = tokio::runtime::Builder::new_multi_thread()
+                    .worker_threads(2)
+                    .enable_all()
+                    .build()
+                    .expect("build IAM reload test runtime");
+                runtime.block_on(async {
+                    let _credentials = test_cred();
+                    let temp = tempfile::tempdir().expect("create IAM reload store directory");
+                    let env = Box::pin(
+                        rustfs_test_utils::TestECStoreEnv::builder()
+                            .base_dir(temp.path())
+                            .init_bucket_metadata(false)
+                            .build(),
+                    )
+                    .await;
+                    let store = ObjectStore::new(env.ecstore);
+                    store
+                        .save_group_info("reload-group", GroupInfo::new(vec!["reload-user".to_string()]))
+                        .await
+                        .expect("seed group membership");
+                    store
+                        .save_mapped_policy("reload-user", UserType::Reg, false, MappedPolicy::new("readwrite"), None)
+                        .await
+                        .expect("seed user policy");
+                    store
+                        .save_mapped_policy("reload-group", UserType::Reg, true, MappedPolicy::new("readonly"), None)
+                        .await
+                        .expect("seed group policy");
+
+                    let cache = Cache::default();
+                    store.load_all(&cache).await.expect("load nonempty IAM partitions");
+                    {
+                        let snapshot = cache.snapshot();
+                        assert!(snapshot.groups.contains_key("reload-group"));
+                        assert_eq!(snapshot.user_policies["reload-user"].policies, "readwrite");
+                        assert_eq!(snapshot.group_policies["reload-group"].policies, "readonly");
+                        assert!(snapshot.user_group_memberships["reload-user"].contains("reload-group"));
+                    }
+
+                    // Mutate durable state without notifying this cache, as on
+                    // a peer that missed the last mapping/group deletion.
+                    store
+                        .delete_mapped_policy("reload-user", UserType::Reg, false)
+                        .await
+                        .expect("remove last user mapping");
+                    store
+                        .delete_mapped_policy("reload-group", UserType::Reg, true)
+                        .await
+                        .expect("remove last group mapping");
+                    store.delete_group_info("reload-group").await.expect("remove last group");
+                    assert!(cache.snapshot().groups.contains_key("reload-group"));
+                    // Simulate a missed peer revocation while STS publication
+                    // and a separate user update overlap this full disk reload.
+                    let stale = rustfs_policy::auth::UserIdentity {
+                        version: 1,
+                        ..Default::default()
+                    };
+                    cache.add_or_update_user("revoked-user", &stale, time::OffsetDateTime::now_utc());
+                    let barrier = std::sync::Arc::new((tokio::sync::Notify::new(), tokio::sync::Notify::new()));
+                    let loading = super::LOAD_ALL_COMMIT_BARRIER.scope(std::sync::Arc::clone(&barrier), store.load_all(&cache));
+                    let concurrent = async {
+                        barrier.0.notified().await;
+                        cache.add_or_update_sts_account("new-session", &stale, time::OffsetDateTime::now_utc());
+                        cache.add_or_update_user("new-service", &stale, time::OffsetDateTime::now_utc());
+                        barrier.1.notify_one();
+                    };
+                    let (loaded, ()) = tokio::join!(loading, concurrent);
+                    loaded.expect("reconcile revoked identities despite STS traffic");
+                    let snapshot = cache.snapshot();
+                    assert!(!snapshot.users.contains_key("revoked-user"));
+                    assert!(snapshot.users.contains_key("new-service"));
+                    assert!(snapshot.sts_accounts.contains_key("new-session"));
+                    assert!(snapshot.groups.is_empty());
+                    assert!(snapshot.user_policies.is_empty());
+                    assert!(snapshot.group_policies.is_empty());
+                    assert!(snapshot.user_group_memberships.is_empty());
+                    assert!(!snapshot.policy_docs.is_empty(), "built-in policies must remain available");
+                });
+            })
+            .expect("spawn IAM reload test thread")
+            .join()
+            .expect("IAM reload test should complete");
+    }
+
     /// rustfs#4304 / startup contract rustfs#4056: the bulk snapshot load
     /// (`load_all`) must never depend on the node-counted namespace lock
     /// quorum, so its reads have to carry `no_lock = true` down to the

crates/policy/src/arn.rs

@@ -20,7 +20,7 @@ const ARN_PARTITION_RUSTFS: &str = "rustfs";
 const ARN_SERVICE_IAM: &str = "iam";
 const ARN_RESOURCE_TYPE_ROLE: &str = "role";
 
-#[derive(Debug, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct ARN {
     pub partition: String,
     pub service: String,

crates/policy/src/policy/action.rs

@@ -165,9 +165,10 @@ impl<'de> Deserialize<'de> for ActionSet {
     }
 }
 
-#[derive(Serialize, Deserialize, Hash, PartialEq, Eq, Clone, Debug, Copy)]
+#[derive(Deserialize, Hash, PartialEq, Eq, Clone, Debug, Copy)]
 #[serde(try_from = "&str", untagged)]
 pub enum Action {
+    All,
     S3Action(S3Action),
     AdminAction(AdminAction),
     StsAction(StsAction),
@@ -175,12 +176,29 @@ pub enum Action {
     None,
 }
 
+impl Serialize for Action {
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        match self {
+            Self::None => serializer.serialize_unit(),
+            _ => serializer.serialize_str(self.into()),
+        }
+    }
+}
+
 impl Action {
     pub fn is_match(&self, action: &Action) -> bool {
         self.is_match_for_effect(action, false)
     }
 
     pub fn is_match_for_effect(&self, action: &Action, deny: bool) -> bool {
+        if matches!(self, Self::All) {
+            // Preserve the historical S3-only Allow while allowing deny-all and
+            // NotAction exclusions to cover every action family.
+            return deny || Self::S3Action(S3Action::AllActions).is_match_for_effect(action, false);
+        }
         // Force-delete bypasses emptiness and version checks. A blanket `s3:*`
         // / `*` Allow (including canned consoleAdmin) must not confer it; the
         // statement has to name `s3:ForceDeleteBucket` or `s3:ForceDeleteObject`.
@@ -202,6 +220,7 @@ fn action_requires_explicit_grant(action: &Action) -> bool {
 impl From<&Action> for &str {
     fn from(value: &Action) -> &'static str {
         match value {
+            Action::All => "*",
             Action::S3Action(s) => s.into(),
             Action::AdminAction(s) => s.into(),
             Action::StsAction(s) => s.into(),
@@ -221,9 +240,8 @@ impl Action {
 impl TryFrom<&str> for Action {
     type Error = Error;
     fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
-        // Support wildcard "*" which matches all S3 actions (AWS S3 standard)
         if value == "*" {
-            return Ok(Self::S3Action(S3Action::AllActions));
+            return Ok(Self::All);
         }
         if value.starts_with(Self::S3_PREFIX) {
             Ok(Self::S3Action(
@@ -822,7 +840,7 @@ mod tests {
     #[test]
     fn blanket_s3_wildcard_does_not_grant_force_delete() {
         let wildcard = Action::try_from("s3:*").expect("s3:* parses");
-        let star = Action::try_from("*").expect("* parses as s3:*");
+        let star = Action::try_from("*").expect("* parses");
         let force_bucket = Action::try_from("s3:ForceDeleteBucket").expect("force bucket action parses");
         let force_object = Action::try_from("s3:ForceDeleteObject").expect("force object action parses");
         let delete_object = Action::try_from("s3:DeleteObject").expect("delete object parses");
@@ -841,9 +859,10 @@ mod tests {
 
     #[test]
     fn test_action_wildcard_parsing() {
-        // Test that "*" parses to S3Action::AllActions
+        // Preserve the bare wildcard through serialization.
         let action = Action::try_from("*").expect("Should parse wildcard");
-        assert!(matches!(action, Action::S3Action(S3Action::AllActions)));
+        assert!(matches!(action, Action::All));
+        assert_eq!(serde_json::to_string(&action).expect("serialize wildcard"), r#""*""#);
     }
 
     #[test]
@@ -929,15 +948,13 @@ mod tests {
 
     #[test]
     fn test_actionset_wildcard_serialization() {
-        // Wildcard action should serialize as array for S3 specification compliance
-        let actionset = ActionSet(vec![Action::try_from("*").expect("Should parse wildcard")]);
-
-        let json = serde_json::to_string(&actionset).expect("Should serialize");
-        let parsed: serde_json::Value = serde_json::from_str(&json).expect("Should parse");
-        assert!(parsed.is_array(), "Should serialize as array");
-        let arr = parsed.as_array().expect("Should be array");
-        assert_eq!(arr.len(), 1);
-        assert_eq!(arr[0].as_str().unwrap(), "s3:*");
+        for wildcard in ["*", "s3:*"] {
+            let actionset = ActionSet(vec![Action::try_from(wildcard).expect("parse wildcard")]);
+            let json = serde_json::to_string(&actionset).expect("serialize action set");
+            assert_eq!(json, format!(r#"["{wildcard}"]"#));
+            let round_trip: ActionSet = serde_json::from_str(&json).expect("deserialize action set");
+            assert_eq!(round_trip, actionset);
+        }
     }
 
     #[test]

crates/policy/src/policy/function/addr.rs

@@ -15,7 +15,7 @@
 use super::func::InnerFunc;
 use ipnetwork::IpNetwork;
 use serde::{Deserialize, Serialize, de::Visitor};
-use std::{borrow::Cow, collections::HashMap, net::IpAddr};
+use std::{collections::HashMap, net::IpAddr};
 
 pub type AddrFunc = InnerFunc<AddrFuncValue>;
 
@@ -30,7 +30,16 @@ impl AddrFunc {
                 };
 
                 for ip_net in inner.values.0.iter() {
-                    if ip_net.contains(ip) {
+                    let canonical = ip.to_canonical();
+                    let matches_mapped_network = match (ip_net, canonical) {
+                        (IpNetwork::V6(network), IpAddr::V4(address))
+                            if network.prefix() >= 96 && network.ip().to_ipv4_mapped().is_some() =>
+                        {
+                            network.contains(address.to_ipv6_mapped())
+                        }
+                        _ => false,
+                    };
+                    if ip_net.contains(ip) || ip_net.contains(canonical) || matches_mapped_network {
                         return true;
                     }
                 }
@@ -81,13 +90,7 @@ impl<'de> Deserialize<'de> for AddrFuncValue {
 
         impl AddrFuncValueVisitor {
             fn cidr<E: serde::de::Error>(v: &str) -> Result<IpNetwork, E> {
-                let mut cidr_str = Cow::from(v);
-                if v.find('/').is_none() {
-                    cidr_str.to_mut().push_str("/32");
-                }
-
-                cidr_str
-                    .parse::<IpNetwork>()
+                v.parse::<IpNetwork>()
                     .map_err(|_| E::custom(format!("{v} can not be parsed to CIDR")))
             }
         }
@@ -107,6 +110,40 @@ mod tests {
     };
     use test_case::test_case;
 
+    #[test]
+    fn source_ip_matches_ipv4_and_mapped_forms_across_protocols() {
+        for network in ["203.0.113.0/24", "::ffff:203.0.113.0/120"] {
+            let function: AddrFunc =
+                serde_json::from_str(&format!(r#"{{"aws:SourceIp":"{network}"}}"#)).expect("parse source IP condition");
+            for address in ["203.0.113.9", "::ffff:203.0.113.9"] {
+                assert!(
+                    function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![address.to_string()]),])),
+                    "{network} must match {address}"
+                );
+            }
+            for address in ["198.51.100.9", "::ffff:198.51.100.9", "2001:db8::9"] {
+                assert!(
+                    !function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![address.to_string()]),])),
+                    "{network} must not match {address}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn bare_address_matches_only_the_named_host() {
+        for (host, neighbor, prefix) in [
+            ("2001:db8:85a3::8a2e:370:7334", "2001:db8:ffff::1", 128),
+            ("203.0.113.7", "203.0.113.8", 32),
+        ] {
+            let function: AddrFunc =
+                serde_json::from_str(&format!(r#"{{"aws:SourceIp":"{host}"}}"#)).expect("parse bare IP address condition");
+            assert_eq!(function.0[0].values.0[0].prefix(), prefix);
+            assert!(function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![host.to_string()])])));
+            assert!(!function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![neighbor.to_string()])])));
+        }
+    }
+
     fn new_func(name: KeyName, variable: Option<String>, value: Vec<&str>) -> AddrFunc {
         AddrFunc {
             0: vec![FuncKeyValue {
@@ -119,20 +156,20 @@ mod tests {
     #[test_case(r#"{"aws:SourceIp": "203.0.113.0/24"}"#, new_func(Aws(AWSSourceIP), None, vec!["203.0.113.0/24"]); "1")]
     #[test_case(r#"{"aws:SourceIp": "203.0.113.0"}"#, new_func(Aws(AWSSourceIP), None, vec!["203.0.113.0/32"]); "2")]
     #[test_case(r#"{"aws:SourceIp": "2001:DB8:1234:5678::/64"}"#, new_func(Aws(AWSSourceIP),None, vec!["2001:DB8:1234:5678::/64"]); "3")]
-    #[test_case(r#"{"aws:SourceIp": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/32"]); "4")]
+    #[test_case(r#"{"aws:SourceIp": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/128"]); "4")]
     #[test_case(r#"{"aws:SourceIp": ["203.0.113.0/24","203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), None, vec!["203.0.113.0/24", "203.0.113.0/32"]); "5")]
     #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::/64","203.0.113.0/24"]}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/64", "203.0.113.0/24"]); "6")]
-    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),None, vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/32"]); "7")]
-    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/32", "203.0.113.0/32"]); "8")]
+    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),None, vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/128"]); "7")]
+    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/128", "203.0.113.0/32"]); "8")]
     #[test_case(r#"{"aws:SourceIp/a": "203.0.113.0/24"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/24"]); "9")]
     #[test_case(r#"{"aws:SourceIp/a": "203.0.113.0/24"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/24"]); "10")]
     #[test_case(r#"{"aws:SourceIp/a": "203.0.113.0"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/32"]); "11")]
     #[test_case(r#"{"aws:SourceIp/a": "2001:DB8:1234:5678::/64"}"#, new_func(Aws(AWSSourceIP),Some("a".into()), vec!["2001:DB8:1234:5678::/64"]); "12")]
-    #[test_case(r#"{"aws:SourceIp/a": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/32"]); "13")]
+    #[test_case(r#"{"aws:SourceIp/a": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/128"]); "13")]
     #[test_case(r#"{"aws:SourceIp/a": ["203.0.113.0/24", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/24", "203.0.113.0/32"]); "14")]
     #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::/64", "203.0.113.0/24"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/64", "203.0.113.0/24"]); "15")]
-    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),Some("a".into()), vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/32"]); "16")]
-    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/32", "203.0.113.0/32"]); "17")]
+    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),Some("a".into()), vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/128"]); "16")]
+    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/128", "203.0.113.0/32"]); "17")]
     fn test_deser(input: &str, expect: AddrFunc) -> Result<(), serde_json::Error> {
         let v: AddrFunc = serde_json::from_str(input)?;
         assert_eq!(v, expect);

crates/policy/src/policy/function/condition.rs

@@ -61,6 +61,11 @@ pub enum Condition {
 
 impl Condition {
     pub fn from_deserializer<'a, D: MapAccess<'a>>(key: &str, d: &mut D) -> Result<Self, D::Error> {
+        // Two suffixes are retained for existing serialized policies. Bound the
+        // chain before parsing so Clone, evaluation and Drop stay bounded too.
+        if key.ends_with("IfExistsIfExistsIfExists") {
+            return Err(Error::custom("condition operator has too many IfExists suffixes"));
+        }
         Ok(match key {
             "StringEquals" => Self::StringEquals(d.next_value()?),
             "StringNotEquals" => Self::StringNotEquals(d.next_value()?),
@@ -262,18 +267,18 @@ impl Condition {
                 Null(s) => s.evaluate_null(values),
                 Bool(s) => s.evaluate_bool(values),
                 NumericEquals(s) => s.evaluate(i64::eq, false, values),
-                NumericNotEquals(s) => s.evaluate(i64::ne, false, values),
+                NumericNotEquals(s) => s.evaluate(i64::ne, true, values),
                 NumericLessThan(s) => s.evaluate(i64::lt, false, values),
                 NumericLessThanEquals(s) => s.evaluate(i64::le, false, values),
                 NumericGreaterThan(s) => s.evaluate(i64::gt, false, values),
                 NumericGreaterThanIfExists(s) => s.evaluate(i64::ge, true, values),
                 NumericGreaterThanEquals(s) => s.evaluate(i64::ge, false, values),
-                DateEquals(s) => s.evaluate(OffsetDateTime::eq, values),
-                DateNotEquals(s) => s.evaluate(OffsetDateTime::ne, values),
-                DateLessThan(s) => s.evaluate(OffsetDateTime::lt, values),
-                DateLessThanEquals(s) => s.evaluate(OffsetDateTime::le, values),
-                DateGreaterThan(s) => s.evaluate(OffsetDateTime::gt, values),
-                DateGreaterThanEquals(s) => s.evaluate(OffsetDateTime::ge, values),
+                DateEquals(s) => s.evaluate(OffsetDateTime::eq, false, values),
+                DateNotEquals(s) => s.evaluate(OffsetDateTime::ne, true, values),
+                DateLessThan(s) => s.evaluate(OffsetDateTime::lt, false, values),
+                DateLessThanEquals(s) => s.evaluate(OffsetDateTime::le, false, values),
+                DateGreaterThan(s) => s.evaluate(OffsetDateTime::gt, false, values),
+                DateGreaterThanEquals(s) => s.evaluate(OffsetDateTime::ge, false, values),
                 IfExists(inner) => {
                     if !inner.has_any_key_in(values) {
                         return true;
@@ -486,6 +491,31 @@ mod tests {
         assert_eq!(functions, deserialized);
     }
 
+    #[test]
+    fn repeated_if_exists_suffixes_are_rejected_before_recursion() {
+        for count in [3, 10_000] {
+            let key = format!("StringEquals{}", "IfExists".repeat(count));
+            let document = serde_json::json!({key: {"s3:x-amz-server-side-encryption": "aws:kms"}});
+            let result = serde_json::from_slice::<Functions>(&serde_json::to_vec(&document).expect("serialize condition"));
+            assert!(result.is_err(), "{count} suffixes must be rejected without constructing a deep chain");
+        }
+    }
+
+    #[tokio::test]
+    async fn numeric_and_date_not_equals_match_absent_keys() {
+        for (operator, key, value, absent) in [
+            ("NumericNotEquals", "s3:max-keys", "30", true),
+            ("NumericEquals", "s3:max-keys", "30", false),
+            ("DateNotEquals", "s3:object-lock-retain-until-date", "2030-01-01T00:00:00Z", true),
+            ("DateEquals", "s3:object-lock-retain-until-date", "2030-01-01T00:00:00Z", false),
+        ] {
+            let document = serde_json::json!({operator: {key: value}});
+            let functions: Functions =
+                serde_json::from_slice(&serde_json::to_vec(&document).expect("serialize condition")).expect("parse condition");
+            assert_eq!(functions.evaluate(&HashMap::new()).await, absent, "{operator}");
+        }
+    }
+
     #[test]
     fn test_nested_if_exists_serde_round_trip() {
         // Verifies that nested IfExists(IfExists(StringEquals)) serializes with the

crates/policy/src/policy/function/date.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::func::InnerFunc;
+use super::key_name::{AwsKeyName, KeyName};
 use serde::{Deserialize, Deserializer, Serialize, de};
 use std::{collections::HashMap, fmt};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
@@ -20,14 +21,27 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 pub type DateFunc = InnerFunc<DateFuncValue>;
 
 impl DateFunc {
-    pub fn evaluate(&self, op: impl Fn(&OffsetDateTime, &OffsetDateTime) -> bool, values: &HashMap<String, Vec<String>>) -> bool {
+    pub fn evaluate(
+        &self,
+        op: impl Fn(&OffsetDateTime, &OffsetDateTime) -> bool,
+        absent: bool,
+        values: &HashMap<String, Vec<String>>,
+    ) -> bool {
         for inner in self.0.iter() {
             let v = match values.get(inner.key.name().as_str()).and_then(|x| x.first()) {
                 Some(x) => x,
+                None if absent => continue,
                 None => return false,
             };
 
-            let Ok(rv) = OffsetDateTime::parse(v, &Rfc3339) else {
+            let parsed = if inner.key.is(&KeyName::Aws(AwsKeyName::AWSEpochTime)) {
+                v.parse::<i64>()
+                    .ok()
+                    .and_then(|seconds| OffsetDateTime::from_unix_timestamp(seconds).ok())
+            } else {
+                OffsetDateTime::parse(v, &Rfc3339).ok()
+            };
+            let Some(rv) = parsed else {
                 return false;
             };
 
@@ -88,7 +102,7 @@ impl<'de> Deserialize<'de> for DateFuncValue {
 
 #[cfg(test)]
 mod tests {
-    use super::{DateFunc, DateFuncValue};
+    use super::{AwsKeyName, DateFunc, DateFuncValue};
     use crate::policy::function::func::FuncKeyValue;
     use crate::policy::function::{
         key::Key,
@@ -124,15 +138,50 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn date_conditions_compare_epoch_seconds_only_for_epoch_time() {
+        let function = new_func(KeyName::Aws(AwsKeyName::AWSEpochTime), None, "2020-01-01T00:00:00Z");
+        for (value, equal, later) in [
+            ("1577836799", false, false),
+            ("1577836800", true, false),
+            ("1577836801", false, true),
+        ] {
+            let values = HashMap::from([("EpochTime".to_string(), vec![value.to_string()])]);
+            assert_eq!(function.evaluate(OffsetDateTime::eq, false, &values), equal);
+            assert_eq!(function.evaluate(OffsetDateTime::gt, false, &values), later);
+            assert_eq!(function.evaluate(OffsetDateTime::lt, false, &values), !equal && !later);
+        }
+        for invalid in ["2020-01-01T00:00:00Z", "invalid", "9223372036854775807"] {
+            let values = HashMap::from([("EpochTime".to_string(), vec![invalid.to_string()])]);
+            assert!(!function.evaluate(OffsetDateTime::ne, true, &values));
+        }
+        let header_date = new_func(S3(S3ObjectLockRetainUntilDate), None, "2020-01-01T00:00:00Z");
+        let values = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["1577836800".to_string()])]);
+        assert!(!header_date.evaluate(OffsetDateTime::eq, false, &values));
+    }
+
+    #[test]
+    fn absent_date_key_does_not_skip_other_keys() {
+        let mut function = new_func(S3(S3ObjectLockRetainUntilDate), None, "2030-01-01T00:00:00Z");
+        function
+            .0
+            .extend(new_func(KeyName::Aws(AwsKeyName::AWSCurrentTime), None, "2030-01-01T00:00:00Z").0);
+        let values = HashMap::from([("CurrentTime".to_string(), vec!["2030-01-01T00:00:00Z".to_string()])]);
+        assert!(!function.evaluate(OffsetDateTime::ne, true, &values));
+        assert!(function.evaluate(OffsetDateTime::eq, true, &values));
+        function.0.reverse();
+        assert!(!function.evaluate(OffsetDateTime::ne, true, &values));
+    }
+
     #[test]
     fn evaluate_compares_request_date_to_policy_date() {
         let function = new_func(S3(S3ObjectLockRetainUntilDate), None, "2030-01-01T00:00:00Z");
         let later = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["2099-01-01T00:00:00Z".to_string()])]);
         let earlier = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["2029-01-01T00:00:00Z".to_string()])]);
 
-        assert!(function.evaluate(OffsetDateTime::gt, &later));
-        assert!(!function.evaluate(OffsetDateTime::gt, &earlier));
-        assert!(function.evaluate(OffsetDateTime::lt, &earlier));
-        assert!(!function.evaluate(OffsetDateTime::lt, &later));
+        assert!(function.evaluate(OffsetDateTime::gt, false, &later));
+        assert!(!function.evaluate(OffsetDateTime::gt, false, &earlier));
+        assert!(function.evaluate(OffsetDateTime::lt, false, &earlier));
+        assert!(!function.evaluate(OffsetDateTime::lt, false, &later));
     }
 }

crates/policy/src/policy/function/number.rs

@@ -26,11 +26,12 @@ pub type NumberFunc = InnerFunc<NumberFuncValue>;
 pub struct NumberFuncValue(i64);
 
 impl NumberFunc {
-    pub fn evaluate(&self, op: impl Fn(&i64, &i64) -> bool, if_exists: bool, values: &HashMap<String, Vec<String>>) -> bool {
+    pub fn evaluate(&self, op: impl Fn(&i64, &i64) -> bool, absent: bool, values: &HashMap<String, Vec<String>>) -> bool {
         for inner in self.0.iter() {
             let v = match values.get(inner.key.name().as_str()).and_then(|x| x.first()) {
                 Some(x) => x,
-                None => return if_exists,
+                None if absent => continue,
+                None => return false,
             };
 
             let Ok(rv) = v.parse::<i64>() else {
@@ -146,6 +147,20 @@ mod tests {
         assert_eq!(limit.evaluate(i64::eq, false, &values), equals);
     }
 
+    #[test]
+    fn absent_numeric_key_does_not_skip_other_keys() {
+        let mut function = new_func(S3(S3MaxKeys), None, 30);
+        function
+            .0
+            .extend(new_func(S3(S3ObjectLockRemainingRetentionDays), None, 30).0);
+        let values = HashMap::from([("object-lock-remaining-retention-days".to_string(), vec!["30".to_string()])]);
+        assert!(!function.evaluate(i64::ne, true, &values));
+        assert!(!function.evaluate(i64::gt, true, &values));
+        assert!(function.evaluate(i64::eq, true, &values));
+        function.0.reverse();
+        assert!(!function.evaluate(i64::ne, true, &values));
+    }
+
     #[test]
     fn remaining_retention_days_numeric_condition_without_value_uses_if_exists() {
         let limit = new_func(S3(S3ObjectLockRemainingRetentionDays), None, 30);

crates/policy/src/policy/policy.rs

@@ -3416,6 +3416,277 @@ mod test {
         assert_eq!(round_trip.statements[0].effect, policy.statements[0].effect);
     }
 
+    #[tokio::test]
+    async fn global_wildcard_deny_covers_all_families_without_broadening_allow() {
+        for action in [
+            "s3:GetObject",
+            "s3:ForceDeleteObject",
+            "admin:CreateUser",
+            "sts:AssumeRole",
+            "kms:Configure",
+            "kms:Decrypt",
+        ] {
+            let parsed_action = Action::try_from(action).expect("action parses");
+            let conditions = HashMap::new();
+            let claims = HashMap::new();
+            let args = Args {
+                account: "user",
+                groups: &None,
+                action: parsed_action,
+                bucket: if action.starts_with("s3:") { "bucket" } else { "" },
+                object: if action.starts_with("s3:") { "object" } else { "" },
+                conditions: &conditions,
+                claims: &claims,
+                is_owner: false,
+                deny_only: false,
+            };
+            for (effect, selector, wildcard, expected) in [
+                ("Deny", "Action", "*", false),
+                ("Deny", "Action", "s3:*", !action.starts_with("s3:")),
+                ("Deny", "NotAction", "*", true),
+            ] {
+                let document = serde_json::json!({
+                    "Version": "2012-10-17", "Statement": [
+                        {"Effect": "Allow", "Action": action, "Resource": "*"},
+                        {"Effect": effect, selector: wildcard, "Resource": "*"}
+                    ]
+                });
+                let policy = Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy"))
+                    .expect("parse wildcard policy");
+                assert_eq!(policy.is_allowed(&args).await, expected, "{effect} {selector} {wildcard} for {action}");
+                let serialized = serde_json::to_value(&policy).expect("serialize policy");
+                assert_eq!(serialized["Statement"][1][selector][0], wildcard);
+            }
+            for selector in ["Action", "NotAction"] {
+                let document = serde_json::json!({
+                    "Version": "2012-10-17", "Statement": [{"Effect": "Allow", selector: "*", "Resource": "*"}]
+                });
+                let policy = Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy"))
+                    .expect("parse wildcard Allow");
+                assert_eq!(policy.is_allowed(&args).await, selector == "Action" && action == "s3:GetObject");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn global_wildcard_deny_still_obeys_resource_and_condition_scope() {
+        let document = serde_json::json!({
+            "Version": "2012-10-17", "Statement": [
+                {"Effect": "Allow", "Action": "admin:CreateUser", "Resource": "*"},
+                {"Effect": "Deny", "Action": "*", "Resource": "arn:aws:s3:::bucket/*"}
+            ]
+        });
+        let policy = Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy")).expect("parse scoped Deny");
+        let claims = HashMap::new();
+        let conditions = HashMap::new();
+        let mut args = Args {
+            account: "user",
+            groups: &None,
+            action: Action::try_from("admin:CreateUser").expect("admin action"),
+            bucket: "",
+            object: "",
+            conditions: &conditions,
+            claims: &claims,
+            is_owner: false,
+            deny_only: false,
+        };
+        assert!(policy.is_allowed(&args).await, "bucket-scoped Deny must not become a global admin Deny");
+        let mut document = document;
+        document["Statement"][1]["Resource"] = serde_json::json!("*");
+        document["Statement"][1]["Condition"] = serde_json::json!({"StringEquals": {"aws:username": "blocked"}});
+        let conditional =
+            Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy")).expect("parse conditional Deny");
+        assert!(conditional.is_allowed(&args).await);
+        let conditions = HashMap::from([("username".to_string(), vec!["blocked".to_string()])]);
+        args.conditions = &conditions;
+        assert!(!conditional.is_allowed(&args).await);
+    }
+
+    #[tokio::test]
+    async fn security_conditions_enforce_identity_and_bucket_denies() {
+        for (operator, key, policy_value, request_value, denied) in [
+            ("NumericNotEquals", "s3:object-lock-remaining-retention-days", "30", None, true),
+            ("NumericNotEquals", "s3:object-lock-remaining-retention-days", "30", Some("30"), false),
+            ("NumericNotEquals", "s3:object-lock-remaining-retention-days", "30", Some("10"), true),
+            ("DateNotEquals", "s3:object-lock-retain-until-date", "2030-01-01T00:00:00Z", None, true),
+            (
+                "DateNotEquals",
+                "s3:object-lock-retain-until-date",
+                "2030-01-01T00:00:00Z",
+                Some("2030-01-01T00:00:00Z"),
+                false,
+            ),
+            ("DateGreaterThan", "aws:EpochTime", "2020-01-01T00:00:00Z", Some("1577836801"), true),
+            ("DateGreaterThan", "aws:EpochTime", "2020-01-01T00:00:00Z", Some("1577836799"), false),
+            (
+                "NotIpAddress",
+                "aws:SourceIp",
+                "2001:db8:85a3::8a2e:370:7334",
+                Some("2001:db8:ffff::1"),
+                true,
+            ),
+            (
+                "NotIpAddress",
+                "aws:SourceIp",
+                "2001:db8:85a3::8a2e:370:7334",
+                Some("2001:db8:85a3::8a2e:370:7334"),
+                false,
+            ),
+            (
+                "IpAddress",
+                "aws:SourceIp",
+                "2001:db8:85a3::8a2e:370:7334",
+                Some("2001:db8:ffff::1"),
+                false,
+            ),
+            ("IpAddress", "aws:SourceIp", "2001:db8::/32", Some("2001:db8:ffff::1"), true),
+        ] {
+            let mut document = serde_json::json!({
+                "Version": "2012-10-17", "Statement": [
+                    {"Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::bucket/*"},
+                    {"Effect": "Deny", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::bucket/*",
+                     "Condition": {operator: {key: policy_value}}}
+                ]
+            });
+            let policy =
+                Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy")).expect("parse identity policy");
+            let conditions = request_value
+                .map(|value| {
+                    HashMap::from([(key.split_once(':').expect("condition prefix").1.to_string(), vec![value.to_string()])])
+                })
+                .unwrap_or_default();
+            let claims = HashMap::new();
+            let args = Args {
+                account: "user",
+                groups: &None,
+                action: Action::S3Action(S3Action::PutObjectAction),
+                bucket: "bucket",
+                object: "object",
+                conditions: &conditions,
+                claims: &claims,
+                is_owner: false,
+                deny_only: false,
+            };
+            assert_eq!(policy.is_allowed(&args).await, !denied, "identity {operator} {key} {request_value:?}");
+            for statement in document["Statement"].as_array_mut().expect("statements array") {
+                statement["Principal"] = serde_json::json!("*");
+            }
+            let policy: BucketPolicy = serde_json::from_slice(&serde_json::to_vec(&document).expect("serialize bucket policy"))
+                .expect("parse bucket policy");
+            assert_eq!(
+                policy
+                    .is_allowed(&BucketPolicyArgs {
+                        account: "user",
+                        groups: &None,
+                        action: args.action,
+                        bucket: "bucket",
+                        object: "object",
+                        conditions: &conditions,
+                        is_owner: false,
+                    })
+                    .await,
+                !denied,
+                "bucket {operator} {key} {request_value:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn service_principal_never_grants_access_to_a_named_access_key() {
+        for principal in [
+            serde_json::json!({"Service": "logging.s3.amazonaws.com"}),
+            serde_json::json!({"Service": "*"}),
+            serde_json::json!({"Service": "logging.s3.amazonaws.com", "AWS": "authorized-user"}),
+        ] {
+            let document = serde_json::json!({
+                "Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": principal,
+                "Action": ["s3:GetObject", "s3:PutObject"], "Resource": "arn:aws:s3:::bucket/*"}]
+            });
+            let policy: BucketPolicy = serde_json::from_slice(&serde_json::to_vec(&document).unwrap()).unwrap();
+            for action in [S3Action::GetObjectAction, S3Action::PutObjectAction] {
+                for account in ["logging.s3.amazonaws.com", "authorized-user"] {
+                    let allowed = policy
+                        .is_allowed(&BucketPolicyArgs {
+                            account,
+                            groups: &None,
+                            action: Action::S3Action(action),
+                            bucket: "bucket",
+                            object: "log",
+                            conditions: &HashMap::new(),
+                            is_owner: false,
+                        })
+                        .await;
+                    assert_eq!(allowed, account == "authorized-user" && principal.get("AWS").is_some());
+                }
+            }
+            assert_eq!(serde_json::to_value(&policy).unwrap()["Statement"][0]["Principal"], principal);
+        }
+    }
+
+    #[tokio::test]
+    async fn directory_objects_obey_prefix_denies_in_identity_and_bucket_policies() {
+        let mut document = serde_json::json!({
+            "Version": "2012-10-17", "Statement": [
+                {"Effect": "Allow", "Action": "s3:*", "Resource": "arn:aws:s3:::bucket/*"},
+                {"Effect": "Deny", "Action": "s3:*", "Resource": "arn:aws:s3:::bucket/private/*"}
+            ]
+        });
+        let identity = Policy::parse_config(&serde_json::to_vec(&document).unwrap()).unwrap();
+        for statement in document["Statement"].as_array_mut().unwrap() {
+            statement["Principal"] = serde_json::json!("*");
+        }
+        let bucket: BucketPolicy = serde_json::from_slice(&serde_json::to_vec(&document).unwrap()).unwrap();
+        for action in [
+            S3Action::GetObjectAction,
+            S3Action::PutObjectAction,
+            S3Action::DeleteObjectAction,
+            S3Action::DeleteObjectVersionAction,
+        ] {
+            for (object, allowed) in [
+                ("private/", false),
+                ("private/child", false),
+                ("public/", true),
+                ("private", true),
+            ] {
+                let conditions = HashMap::new();
+                assert_eq!(
+                    identity
+                        .is_allowed(&Args {
+                            account: "user",
+                            groups: &None,
+                            action: Action::S3Action(action),
+                            bucket: "bucket",
+                            object,
+                            conditions: &conditions,
+                            claims: &HashMap::new(),
+                            is_owner: false,
+                            deny_only: false,
+                        })
+                        .await,
+                    allowed,
+                    "identity {action:?} {object}"
+                );
+                for account in ["user", ""] {
+                    assert_eq!(
+                        bucket
+                            .is_allowed(&BucketPolicyArgs {
+                                account,
+                                groups: &None,
+                                action: Action::S3Action(action),
+                                bucket: "bucket",
+                                object,
+                                conditions: &conditions,
+                                is_owner: false,
+                            })
+                            .await,
+                        allowed,
+                        "bucket {account} {action:?} {object}"
+                    );
+                }
+            }
+        }
+    }
+
     #[tokio::test]
     async fn force_delete_actions_require_an_explicit_grant() {
         let wildcard = Policy::parse_config(

crates/policy/src/policy/principal.rs

@@ -19,7 +19,7 @@ use std::collections::HashSet;
 
 /// Principal that serializes AWS field as single string when containing only "*",
 /// or as an array otherwise (matching AWS S3 API format).
-/// Also supports Service principals (e.g., "logging.s3.amazonaws.com").
+/// Service principals round-trip for compatibility but do not match access keys.
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct Principal {
     aws: HashSet<String>,
@@ -131,11 +131,6 @@ impl Principal {
                 return true;
             }
         }
-        for pattern in self.service.iter() {
-            if wildcard::is_simple_match(pattern, principal) {
-                return true;
-            }
-        }
         false
     }
 }
@@ -225,7 +220,7 @@ mod test {
             service: HashSet::from(["logging.s3.amazonaws.com".to_string()]),
         };
 
-        assert!(principal.is_match("logging.s3.amazonaws.com"));
+        assert!(!principal.is_match("logging.s3.amazonaws.com"));
         assert!(!principal.is_match("replication.s3.amazonaws.com"));
         assert!(!principal.is_match("arn:aws:iam::123456789012:root"));
     }
@@ -238,7 +233,7 @@ mod test {
         };
 
         assert!(principal.is_match("arn:aws:iam::123456789012:root"));
-        assert!(principal.is_match("logging.s3.amazonaws.com"));
+        assert!(!principal.is_match("logging.s3.amazonaws.com"));
         assert!(!principal.is_match("other-principal"));
     }
 }

crates/policy/src/policy/resource.rs

@@ -17,7 +17,11 @@ use serde::{
     Deserialize, Deserializer, Serialize,
     de::{self, Error as DeError, Visitor},
 };
-use std::{collections::HashMap, fmt, ops::Deref};
+use std::{
+    collections::{HashMap, HashSet},
+    fmt,
+    ops::Deref,
+};
 
 use super::{
     Error as IamError, Validator,
@@ -50,12 +54,6 @@ impl ResourceSet {
         self.0.is_empty()
     }
 
-    fn push_unique(&mut self, resource: Resource) {
-        if !self.0.contains(&resource) {
-            self.0.push(resource);
-        }
-    }
-
     pub async fn is_match(&self, resource: &str, conditions: &HashMap<String, Vec<String>>) -> bool {
         self.is_match_with_resolver(resource, conditions, None).await
     }
@@ -130,9 +128,7 @@ impl<'de> Deserialize<'de> for ResourceSet {
                 E: de::Error,
             {
                 let resource = Resource::try_from(value).map_err(|e| E::custom(format!("invalid resource: {}", e)))?;
-                let mut resources = ResourceSet::default();
-                resources.push_unique(resource);
-                Ok(resources)
+                Ok(ResourceSet(vec![resource]))
             }
 
             fn visit_seq<A>(self, mut seq: A) -> std::result::Result<Self::Value, A::Error>
@@ -140,11 +136,14 @@ impl<'de> Deserialize<'de> for ResourceSet {
                 A: de::SeqAccess<'de>,
                 A::Error: DeError,
             {
-                let mut resources = ResourceSet(Vec::with_capacity(seq.size_hint().unwrap_or(0)));
+                let mut resources = ResourceSet::default();
+                let mut seen = HashSet::new();
                 while let Some(value) = seq.next_element::<String>()? {
                     match Resource::try_from(value.as_str()) {
                         Ok(resource) => {
-                            resources.push_unique(resource);
+                            if seen.insert(resource.clone()) {
+                                resources.0.push(resource);
+                            }
                         }
                         Err(e) => {
                             return Err(A::Error::custom(format!("invalid resource: {}", e)));
@@ -224,7 +223,14 @@ impl Resource {
                 return true;
             }
 
-            if cp != "." && wildcard::is_match(resolved_pattern, &cp) {
+            // Keep directory objects distinct from their bare key for prefix
+            // matching, while retaining path cleaning and exact-match compatibility.
+            let wildcard_resource = if matches!(self, Resource::S3(_)) && resource.ends_with('/') && cp.contains('/') {
+                format!("{cp}/")
+            } else {
+                cp.clone()
+            };
+            if cp != "." && wildcard::is_match(resolved_pattern, &wildcard_resource) {
                 return true;
             }
         }
@@ -314,6 +320,20 @@ mod tests {
     use std::collections::HashMap;
     use test_case::test_case;
 
+    #[test]
+    fn large_resource_sets_deduplicate_without_changing_first_seen_order() {
+        let unique = (0..40_000)
+            .map(|index| format!("arn:aws:s3:::bucket/key-{index}"))
+            .collect::<Vec<_>>();
+        let mut repeated = unique.clone();
+        repeated.extend(unique.iter().rev().cloned());
+        let resources: super::ResourceSet = serde_json::from_slice(&serde_json::to_vec(&repeated).expect("encode resources"))
+            .expect("large valid resource list parses");
+        assert_eq!(resources.len(), unique.len());
+        assert_eq!(serde_json::to_value(resources).expect("serialize resources"), serde_json::json!(unique));
+        assert!(serde_json::from_str::<super::ResourceSet>(r#"["arn:aws:s3:::bucket/key","invalid"]"#).is_err());
+    }
+
     #[test_case("*", "mybucket" => true; "all buckets")]
     #[test_case("*", "mybucket/myobject" => true; "all objects")]
     #[test_case("*", "key/mykey" => true; "all keys")]
@@ -335,6 +355,10 @@ mod tests {
     #[test_case("arn:aws:s3:::mybucket","mybucket/myobject" => false; "15")]
     #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/../victim-bucket/evil.txt" => false; "16")]
     #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/safe/../../victim-bucket/evil.txt" => false; "17")]
+    #[test_case("arn:aws:s3:::bucket/secret/*","bucket/secret/" => true; "directory prefix")]
+    #[test_case("arn:aws:s3:::bucket/secret","bucket/secret/" => true; "directory exact compatibility")]
+    #[test_case("arn:aws:s3:::bucket/*","bucket/" => false; "bucket is not a directory object")]
+    #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/../victim-bucket/" => false; "directory traversal")]
     #[test_case("arn:aws:kms:::key/mykey","key/mykey" => true; "kms exact key")]
     #[test_case("arn:aws:kms:::key/mykey","key/otherkey" => false; "kms wrong key")]
     #[test_case("arn:aws:kms:::key/*","key/mykey" => true; "kms key wildcard")]

crates/policy/src/policy/statement.rs

@@ -144,7 +144,7 @@ impl Statement {
 
         for action in self.actions.iter() {
             match action {
-                Action::S3Action(_) => saw_s3 = true,
+                Action::All | Action::S3Action(_) => saw_s3 = true,
                 Action::AdminAction(_) => saw_admin = true,
                 Action::StsAction(_) => saw_sts = true,
                 Action::KmsAction(_) => saw_kms = true,

crates/policy/src/policy/variables.rs

@@ -259,35 +259,11 @@ async fn resolve_aws_variables_with_depth(pattern: &str, resolver: &dyn PolicyVa
     if depth >= 10 {
         return vec![pattern.to_string()];
     }
-    let mut results = vec![pattern.to_string()];
-
-    let mut changed = true;
-    let max_iterations = 10; // Prevent infinite loops
-    let mut iteration = 0;
-
-    while changed && iteration < max_iterations {
-        changed = false;
-        iteration += 1;
-
-        let mut new_results = Vec::new();
-        for result in &results {
-            let resolved = resolve_single_pass(result, resolver, depth).await;
-            if resolved.len() > 1 || (resolved.len() == 1 && &resolved[0] != result) {
-                changed = true;
-            }
-            new_results.extend(resolved);
-        }
-
-        // Remove duplicates while preserving order
-        results.clear();
-        let mut seen = std::collections::HashSet::new();
-        for result in new_results {
-            if seen.insert(result.clone()) {
-                results.push(result);
-            }
-        }
-    }
-
+    // Values from credentials and claims are data, not policy source. Expanding
+    // substituted placeholders again allows attacker-controlled exponential growth.
+    let mut results = resolve_single_pass(pattern, resolver, depth).await;
+    let mut seen = std::collections::HashSet::new();
+    results.retain(|result| seen.insert(result.clone()));
     results
 }
 
@@ -370,8 +346,8 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
                             let suffix = &results[i].0[end_pos + 1..];
 
                             for value in values {
-                                // Resume after the substitution; a generated placeholder
-                                // belongs to the next bounded pass, not this one.
+                                // Resume after the substitution; generated placeholders
+                                // are literal data and must never be expanded.
                                 new_results.push((format!("{prefix}{value}{suffix}"), actual_pos + value.len()));
                             }
 
@@ -445,8 +421,20 @@ mod tests {
             calls: AtomicUsize::new(0),
         };
         let result = resolve_aws_variables("${aws:AccountId}{aws:username}", &resolver).await;
-        assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
-        assert_eq!(result, vec!["${aws:AccountId}{aws:username}".to_string()]);
+        assert_eq!(resolver.calls.load(Ordering::Relaxed), 1);
+        assert_eq!(result, vec!["$${aws:username}".to_string()]);
+    }
+
+    #[tokio::test]
+    async fn substituted_account_id_placeholders_are_literal() {
+        let account = "${aws:AccountId}".repeat(2);
+        let resolver = VariableResolver::new(VariableContext {
+            account_id: Some(account.clone()),
+            username: Some("alice".to_string()),
+            ..Default::default()
+        });
+        let result = resolve_aws_variables("prefix/${aws:AccountId}/${aws:username}", &resolver).await;
+        assert_eq!(result, vec![format!("prefix/{account}/alice")]);
     }
 
     #[tokio::test]

rustfs/src/admin/auth.rs

@@ -293,6 +293,18 @@ pub async fn authenticate_request(
     result
 }
 
+/// The policy client address comes only from verified middleware context.
+/// Keep the original socket peer in request extensions for auditing.
+pub(crate) fn policy_source_addr(extensions: &http::Extensions) -> Option<std::net::SocketAddr> {
+    let peer = extensions
+        .get::<Option<RemoteAddr>>()
+        .and_then(|peer| peer.map(|addr| addr.0));
+    extensions
+        .get::<rustfs_trusted_proxies::ClientInfo>()
+        .map(|client| std::net::SocketAddr::new(client.real_ip, peer.map_or(0, |addr| addr.port())))
+        .or(peer)
+}
+
 /// Full admin gate over an `S3Request`: extract the request credentials,
 /// authenticate them ([`authenticate_request`]), then authorize the caller for
 /// `actions` ([`validate_admin_request`], allowing on the first permitted
@@ -306,7 +318,7 @@ pub async fn authorize_admin_request(req: &S3Request<Body>, actions: Vec<Action>
 
     let (cred, owner) = authenticate_request(&req.headers, &req.uri, input_cred).await?;
 
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+    let remote_addr = policy_source_addr(&req.extensions);
     validate_admin_request(&req.headers, &cred, owner, false, actions, remote_addr).await?;
 
     Ok(cred)
@@ -547,6 +559,163 @@ mod tests {
         );
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn admin_and_sts_handlers_use_verified_client_ip_and_bucket_scope() {
+        crate::app::gating_test_env::run_large_stack_test("admin-client-conditions", || async {
+            use crate::admin::router::Operation;
+            use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface, publish_test_app_context};
+            use rustfs_iam::store::object::ObjectStore;
+            use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
+            use std::sync::Arc;
+            struct ReadyIam(Arc<rustfs_iam::sys::IamSys<ObjectStore>>);
+            impl IamInterface for ReadyIam {
+                fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<ObjectStore>> {
+                    self.0.clone()
+                }
+                fn is_ready(&self) -> bool {
+                    true
+                }
+            }
+            struct TestKms;
+            impl KmsInterface for TestKms {
+                fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
+                    Arc::new(rustfs_kms::KmsServiceManager::new())
+                }
+            }
+            let previous = crate::runtime_sources::current_app_context();
+            let (_directory, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+            ObjectStore::new(store.clone())
+                .save_iam_config(
+                    serde_json::json!({"version": 1}),
+                    format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+                )
+                .await
+                .unwrap();
+            let iam = rustfs_iam::build_iam_sys(store.clone()).await.unwrap();
+            for (key, document) in [
+                (
+                    "proxy-admin",
+                    serde_json::json!({"Version":"2012-10-17","Statement":[
+                        {"Effect":"Allow","Action":"admin:*"},
+                        {"Effect":"Allow","Action":"sts:AssumeRole"},
+                        {"Effect":"Deny","Action":"admin:*","Condition":{"NotIpAddress":{"aws:SourceIp":"10.0.0.0/8"}}},
+                        {"Effect":"Deny","Action":"sts:AssumeRole","Condition":{"NotIpAddress":{"aws:SourceIp":"10.0.0.0/8"}}}
+                    ]}),
+                ),
+                (
+                    "scoped-usage",
+                    serde_json::json!({"Version":"2012-10-17","Statement":[
+                        {"Effect":"Allow","Action":"s3:ListBucket","Resource":"arn:aws:s3:::usage-allowed"},
+                        {"Effect":"Deny","Action":"s3:ListBucket","NotResource":"arn:aws:s3:::usage-allowed"}
+                    ]}),
+                ),
+            ] {
+                iam.create_user(
+                    key,
+                    &AddOrUpdateUserReq {
+                        secret_key: "test-user-secret".to_owned(),
+                        policy: None,
+                        status: AccountStatus::Enabled,
+                    },
+                )
+                .await
+                .unwrap();
+                iam.set_policy(key, Policy::parse_config(&serde_json::to_vec(&document).unwrap()).unwrap())
+                    .await
+                    .unwrap();
+                iam.policy_db_set(key, UserType::Reg, false, key).await.unwrap();
+            }
+            let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(iam)), Arc::new(TestKms)));
+            assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+                access_key: "proxy-test-root".to_owned(),
+                secret_key: "proxy-root-secret".to_owned(),
+                status: "on".to_owned(),
+                ..Default::default()
+            }));
+            publish_test_app_context(context);
+            let request = |key: &str, client_ip: Option<&str>| {
+                let mut extensions = http::Extensions::new();
+                extensions.insert(Some(RemoteAddr("10.1.2.3:4567".parse().unwrap())));
+                if let Some(ip) = client_ip {
+                    extensions.insert(rustfs_trusted_proxies::ClientInfo::direct(format!("{ip}:7890").parse().unwrap()));
+                }
+                let mut headers = HeaderMap::new();
+                headers.insert("x-forwarded-for", "10.1.2.3".parse().unwrap());
+                headers.insert("x-real-ip", "10.1.2.3".parse().unwrap());
+                S3Request {
+                    input: Body::from(String::new()),
+                    method: http::Method::GET,
+                    uri: "/".parse().unwrap(),
+                    headers,
+                    extensions,
+                    credentials: Some(s3s::auth::Credentials {
+                        access_key: key.to_owned(),
+                        secret_key: s3s::auth::SecretKey::from("test-user-secret".to_owned()),
+                    }),
+                    region: None,
+                    service: None,
+                    trailing_headers: None,
+                }
+            };
+            for (client, allowed) in [(Some("203.0.113.17"), false), (Some("10.8.7.6"), true), (None, true)] {
+                let req = request("proxy-admin", client);
+                assert_eq!(policy_source_addr(&req.extensions).unwrap().port(), 4567);
+                let result = authorize_admin_request(&req, vec![admin_action()]).await;
+                if allowed {
+                    result.expect("verified private client is allowed");
+                } else {
+                    assert_eq!(result.unwrap_err().code(), &s3s::S3ErrorCode::AccessDenied);
+                }
+                let mut req = request("proxy-admin", client);
+                req.method = http::Method::POST;
+                req.input = Body::from("Action=AssumeRole&Version=invalid-version".to_owned());
+                let error = crate::admin::handlers::sts::AssumeRoleHandle {}
+                    .call(req, matchit::Params::new())
+                    .await
+                    .unwrap_err();
+                if allowed {
+                    assert_eq!(error.code(), &s3s::S3ErrorCode::InvalidArgument);
+                    assert_eq!(
+                        error.message(),
+                        Some("not support version"),
+                        "only an authorized caller reaches version validation"
+                    );
+                } else {
+                    assert_eq!(error.code(), &s3s::S3ErrorCode::AccessDenied);
+                }
+                let response = crate::admin::handlers::is_admin::IsAdminHandler {}
+                    .call(request("proxy-admin", client), matchit::Params::new())
+                    .await
+                    .unwrap();
+                let mut body = response.output.1;
+                let json: serde_json::Value = serde_json::from_slice(&body.store_all_limited(4096).await.unwrap()).unwrap();
+                assert_eq!(json["is_admin"], allowed);
+            }
+            let mut routes = matchit::Router::new();
+            routes.insert("/usage/{bucket}", ()).unwrap();
+            for (bucket, allowed) in [("usage-allowed", true), ("usage-secret", false)] {
+                let path = format!("/usage/{bucket}");
+                let matched = routes.at(&path).unwrap();
+                let mut req = request("scoped-usage", None);
+                req.uri = format!("{path}?unexpected=1").parse().unwrap();
+                let error = crate::admin::handlers::usage_prefix::BucketPrefixUsageHandler {}
+                    .call(req, matched.params)
+                    .await
+                    .unwrap_err();
+                if allowed {
+                    assert_eq!(error.code(), &s3s::S3ErrorCode::InvalidArgument);
+                    assert_eq!(error.message(), Some("unknown query parameter: unexpected"));
+                } else {
+                    assert_eq!(error.code(), &s3s::S3ErrorCode::AccessDenied);
+                }
+            }
+            if let Some(previous) = previous {
+                publish_test_app_context(previous);
+            }
+        });
+    }
+
     /// Owner (root/admin) credential is authorized for an admin action.
     #[tokio::test]
     async fn admin_owner_credential_is_allowed() {

rustfs/src/admin/handlers/account.rs

@@ -42,7 +42,6 @@ use crate::admin::service::caller_identity::{CallerIdentity, oidc_profile_fields
 use crate::admin::storage_api::s3::{self, Body, S3ErrorCode, S3Request, S3Response, S3Result};
 use crate::admin::utils::read_compatible_admin_body;
 use crate::auth::constant_time_eq;
-use crate::server::RemoteAddr;
 use http::StatusCode;
 use hyper::Method;
 use matchit::Params;
@@ -162,6 +161,13 @@ pub struct ChangeOwnPasswordHandler {}
 #[async_trait::async_trait]
 impl Operation for ChangeOwnPasswordHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let signing_secret = req
+            .credentials
+            .as_ref()
+            .ok_or_else(|| s3::error(S3ErrorCode::AccessDenied, "authentication required"))?
+            .secret_key
+            .expose()
+            .to_string();
         let caller = CallerIdentity::resolve(&req).await?;
         let audit = AccountAuditContext::from_request(&req);
 
@@ -203,7 +209,9 @@ impl Operation for ChangeOwnPasswordHandler {
             return Err(s3::error(S3ErrorCode::InvalidRequest, "the calling identity no longer exists"));
         };
 
-        if !constant_time_eq(&request.current_secret_key, &stored.credentials.secret_key) {
+        if !constant_time_eq(&request.current_secret_key, &stored.credentials.secret_key)
+            || !constant_time_eq(&request.current_secret_key, &signing_secret)
+        {
             warn!(
                 component = LOG_COMPONENT_ADMIN,
                 subsystem = LOG_SUBSYSTEM_ACCOUNT,
@@ -256,31 +264,16 @@ impl Operation for ChangeOwnPasswordHandler {
             let iam_store =
                 current_ready_iam_handle().map_err(|_| s3::error(S3ErrorCode::InternalError, "iam is not initialized"))?;
 
-            let (updated_at, status) = iam_store
-                .set_user_secret_key(&access_key, &new_secret_key)
+            let (updated_at, status, revoked) = iam_store
+                .change_own_secret_key(&access_key, &signing_secret, &new_secret_key)
                 .await
-                .map_err(iam_error_to_s3_error)?;
-
-            // Sessions minted under the old secret must not outlive it. A
-            // failure here is reported but does not undo the rotation: the new
-            // secret is already authoritative, and re-running the revocation is
-            // safe, whereas rolling the secret back would resurrect it.
-            let revoked = match iam_store.revoke_sts_sessions_for_parent(&access_key).await {
-                Ok(revoked) => revoked,
-                Err(err) => {
-                    warn!(
-                        component = LOG_COMPONENT_ADMIN,
-                        subsystem = LOG_SUBSYSTEM_ACCOUNT,
-                        event = EVENT_ADMIN_ACCOUNT_STATE,
-                        action = "change_own_password",
-                        access_key = %MaskedAccessKey(&access_key),
-                        result = "session_revocation_incomplete",
-                        error = ?err,
-                        "admin account state"
-                    );
-                    0
-                }
-            };
+                .map_err(|error| {
+                    if matches!(error, rustfs_iam::error::Error::IAMActionNotAllowed) {
+                        s3::error(S3ErrorCode::InvalidRequest, "the current secret key is incorrect")
+                    } else {
+                        iam_error_to_s3_error(error)
+                    }
+                })?;
 
             emit_audit(
                 &audit_for_task,
@@ -378,7 +371,7 @@ impl Operation for SetUserSecretKeyHandler {
             caller.is_owner,
             false,
             vec![Action::AdminAction(AdminAction::CreateUserAdminAction)],
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await
         .inspect_err(|_| {
@@ -424,28 +417,11 @@ impl Operation for SetUserSecretKeyHandler {
                 ));
             }
 
-            let (updated_at, status) = iam_store
-                .set_user_secret_key(&target, &request.secret_key)
+            let (updated_at, status, revoked) = iam_store
+                .rotate_user_secret_key(&target, &request.secret_key)
                 .await
                 .map_err(iam_error_to_s3_error)?;
 
-            let revoked = match iam_store.revoke_sts_sessions_for_parent(&target).await {
-                Ok(revoked) => revoked,
-                Err(err) => {
-                    warn!(
-                        component = LOG_COMPONENT_ADMIN,
-                        subsystem = LOG_SUBSYSTEM_ACCOUNT,
-                        event = EVENT_ADMIN_ACCOUNT_STATE,
-                        action = "set_user_secret_key",
-                        access_key = %MaskedAccessKey(&target),
-                        result = "session_revocation_incomplete",
-                        error = ?err,
-                        "admin account state"
-                    );
-                    0
-                }
-            };
-
             emit_audit(
                 &audit_for_task,
                 AccountAuditRecord::success(AccountAuditOperation::ResetUserPassword, &target, identity_type)
@@ -648,16 +624,23 @@ mod tests {
     #[test]
     fn both_rotation_handlers_broadcast_after_revoking_sessions() {
         let src = include_str!("account.rs");
-        for marker in [
-            "impl Operation for ChangeOwnPasswordHandler",
-            "impl Operation for SetUserSecretKeyHandler",
+        // Session revocation now runs inside the IAM rotation call, under the
+        // per-key mutation lock, so the handler must reach that call before it
+        // broadcasts the new secret to peer sites.
+        for (marker, rotation) in [
+            ("impl Operation for ChangeOwnPasswordHandler", "change_own_secret_key("),
+            ("impl Operation for SetUserSecretKeyHandler", "rotate_user_secret_key("),
         ] {
             let start = src.find(marker).expect("handler should exist");
             let block = &src[start..];
             let block = &block[..block.find("\n}\n").expect("handler block end")];
             let revoke = block
-                .find("revoke_sts_sessions_for_parent")
-                .expect("handler must revoke sessions");
+                .find(rotation)
+                .expect("handler must rotate through the IAM call that revokes sessions");
+            assert!(
+                !block.contains("revoke_sts_sessions_for_parent"),
+                "{marker}: revocation belongs to the locked IAM rotation, not a separate handler step"
+            );
             let broadcast = block
                 .find("broadcast_secret_key_rotation(")
                 .expect("handler must broadcast the rotation to peer sites");

rustfs/src/admin/handlers/account_info.rs

@@ -22,7 +22,7 @@ use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOption
 use crate::admin::storage_api::data_usage::{apply_bucket_usage_memory_overlay, load_admin_data_usage_from_backend_cached};
 use crate::admin::storage_api::metadata_sys;
 use crate::auth::get_condition_values;
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};
 use matchit::Params;
@@ -146,7 +146,7 @@ impl Operation for AccountInfoHandler {
         let claims = cred.claims.as_ref().unwrap_or(&default_claims);
 
         let cred_clone = cred.clone();
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
         let conditions = get_condition_values(&req.headers, &cred_clone, None, None, remote_addr);
         let cred_clone = Arc::new(cred_clone);
         let conditions = Arc::new(conditions);

rustfs/src/admin/handlers/durability.rs

@@ -28,7 +28,6 @@ use crate::admin::storage_api::bucket::metadata::BUCKET_DURABILITY_CONFIG;
 use crate::admin::storage_api::bucket::metadata_sys;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::server::ADMIN_PREFIX;
-use crate::server::RemoteAddr;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_policy::policy::action::{Action, AdminAction};
@@ -127,7 +126,7 @@ async fn authenticate_admin(req: &S3Request<Body>) -> S3Result<()> {
 
     let (cred, owner) = check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-    let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+    let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
     validate_admin_request(
         &req.headers,
         &cred,

rustfs/src/admin/handlers/group.rs

@@ -434,7 +434,7 @@ impl Operation for SetGroupStatus {
 pub struct UpdateGroupMembers {}
 #[async_trait::async_trait]
 impl Operation for UpdateGroupMembers {
-    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!(
             event = EVENT_ADMIN_GROUP_STATE,
             component = LOG_COMPONENT_ADMIN,
@@ -448,9 +448,16 @@ impl Operation for UpdateGroupMembers {
             return Err(s3_error!(InvalidRequest, "authentication required"));
         }
 
-        authorize_admin_request(&req, vec![Action::AdminAction(AdminAction::AddUserToGroupAdminAction)]).await?;
+        authorize_admin_request(
+            &req,
+            vec![
+                Action::AdminAction(AdminAction::AddUserToGroupAdminAction),
+                Action::AdminAction(AdminAction::RemoveUserFromGroupAdminAction),
+            ],
+        )
+        .await?;
 
-        let mut input = req.input;
+        let mut input = std::mem::take(&mut req.input);
         let body = match input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await {
             Ok(b) => b,
             Err(e) => {
@@ -470,6 +477,13 @@ impl Operation for UpdateGroupMembers {
         let args: GroupAddRemove = serde_json::from_slice(&body)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid JSON: {e}")))?;
 
+        let action = if args.is_remove {
+            AdminAction::RemoveUserFromGroupAdminAction
+        } else {
+            AdminAction::AddUserToGroupAdminAction
+        };
+        authorize_admin_request(&req, vec![Action::AdminAction(action)]).await?;
+
         warn!(
             event = EVENT_ADMIN_GROUP_STATE,
             component = LOG_COMPONENT_ADMIN,
@@ -719,6 +733,143 @@ mod tests {
         &block[..end]
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn group_members_handler_enforces_add_and_remove_permissions() {
+        use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface, publish_test_app_context};
+        use rustfs_iam::store::{Store as _, UserType, object::ObjectStore};
+        use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
+        use rustfs_policy::policy::Policy;
+        use std::sync::Arc;
+
+        struct ReadyIam(Arc<rustfs_iam::sys::IamSys<ObjectStore>>);
+        impl IamInterface for ReadyIam {
+            fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<ObjectStore>> {
+                Arc::clone(&self.0)
+            }
+            fn is_ready(&self) -> bool {
+                true
+            }
+        }
+        struct TestKms;
+        impl KmsInterface for TestKms {
+            fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
+                Arc::new(rustfs_kms::KmsServiceManager::new())
+            }
+        }
+
+        let (_directory, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+        ObjectStore::new(store.clone())
+            .save_iam_config(
+                serde_json::json!({"version": 1}),
+                format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+            )
+            .await
+            .expect("seed IAM format");
+        let iam = rustfs_iam::build_iam_sys(store.clone()).await.expect("initialize IAM");
+        for key in ["group-adder", "group-remover", "group-member"] {
+            iam.create_user(
+                key,
+                &AddOrUpdateUserReq {
+                    secret_key: "group-user-secret".to_owned(),
+                    policy: None,
+                    status: AccountStatus::Enabled,
+                },
+            )
+            .await
+            .expect("create group test user");
+        }
+        for (key, allow, deny) in [
+            ("group-adder", "admin:AddUserToGroup", "admin:RemoveUserFromGroup"),
+            ("group-remover", "admin:RemoveUserFromGroup", "admin:AddUserToGroup"),
+        ] {
+            let document = serde_json::json!({"Version": "2012-10-17", "Statement": [
+                {"Effect": "Allow", "Action": [allow]}, {"Effect": "Deny", "Action": [deny]}
+            ]});
+            iam.set_policy(
+                key,
+                Policy::parse_config(&serde_json::to_vec(&document).expect("policy JSON")).expect("parse policy"),
+            )
+            .await
+            .expect("store delegated policy");
+            iam.policy_db_set(key, UserType::Reg, false, key)
+                .await
+                .expect("attach delegated policy");
+        }
+        iam.add_users_to_group("guardrail", vec!["group-member".to_owned()])
+            .await
+            .expect("seed protected group");
+        let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(Arc::clone(&iam))), Arc::new(TestKms)));
+        assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+            access_key: "group-test-root".to_owned(),
+            secret_key: "group-root-secret".to_owned(),
+            status: "on".to_owned(),
+            ..Default::default()
+        }));
+        publish_test_app_context(context);
+
+        for (key, remove, allowed) in [
+            ("group-adder", true, false),
+            ("group-remover", false, false),
+            ("group-remover", true, true),
+            ("group-adder", false, true),
+        ] {
+            let before = iam
+                .get_group_description("guardrail")
+                .await
+                .expect("read original membership")
+                .members;
+            let mut req = credential_less_request(Method::PUT, "/rustfs/admin/v3/update-group-members");
+            req.credentials = Some(s3s::auth::Credentials {
+                access_key: key.to_owned(),
+                secret_key: s3s::auth::SecretKey::from("group-user-secret".to_owned()),
+            });
+            req.input = Body::from(
+                serde_json::to_vec(&serde_json::json!({
+                    "group": "guardrail", "members": ["group-member"], "groupStatus": "enabled", "isRemove": remove
+                }))
+                .expect("group update JSON"),
+            );
+            let result = UpdateGroupMembers {}.call(req, Params::new()).await;
+            let after = iam
+                .get_group_description("guardrail")
+                .await
+                .expect("read resulting membership")
+                .members;
+            if allowed {
+                result.expect("matching delegated action permits the mutation");
+                assert_eq!(after.contains(&"group-member".to_owned()), !remove);
+            } else {
+                let error = result.expect_err("the opposite action must not authorize the mutation");
+                assert_eq!(error.code(), &S3ErrorCode::AccessDenied);
+                assert_eq!(after, before, "a rejected mutation must preserve the group");
+            }
+        }
+    }
+
+    #[test]
+    fn update_group_members_authorizes_the_decoded_mutation() {
+        let source = include_str!("group.rs")
+            .split("\n#[cfg(test)]\n")
+            .next()
+            .expect("production source");
+        let handler = source_block(source, "impl Operation for UpdateGroupMembers");
+        assert_eq!(handler.matches("authorize_admin_request(").count(), 2);
+        let decoded = handler
+            .find("let args: GroupAddRemove")
+            .expect("decode request before choosing action");
+        let action = handler
+            .find("let action = if args.is_remove")
+            .expect("body controls requested action");
+        let gate = handler.rfind("authorize_admin_request(").expect("mutation authorization");
+        assert!(decoded < action && action < gate);
+        assert!(handler[action..gate].contains("AdminAction::RemoveUserFromGroupAdminAction"));
+        assert!(handler[action..gate].contains("AdminAction::AddUserToGroupAdminAction"));
+        for mutation in [".remove_users_from_group(", ".add_users_to_group("] {
+            assert!(gate < handler.find(mutation).expect("group mutation"));
+        }
+    }
+
     #[test]
     fn group_handlers_use_the_shared_admin_gate_with_their_actions() {
         let production = include_str!("group.rs")
@@ -731,7 +882,6 @@ mod tests {
             ("GetGroup", "GetGroupAdminAction"),
             ("DeleteGroup", "RemoveUserFromGroupAdminAction"),
             ("SetGroupStatus", "EnableGroupAdminAction"),
-            ("UpdateGroupMembers", "AddUserToGroupAdminAction"),
         ] {
             let block = source_block(production, &format!("impl Operation for {handler}"));
             assert_eq!(

rustfs/src/admin/handlers/heal.rs

@@ -1474,7 +1474,7 @@ async fn authenticate_heal_admin_request(req: &S3Request<Body>) -> S3Result<Stri
         owner,
         false,
         vec![Action::AdminAction(AdminAction::HealAdminAction)],
-        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        crate::admin::auth::policy_source_addr(&req.extensions),
     )
     .await?;
     Ok(cred.access_key)

rustfs/src/admin/handlers/idp_compat.rs

@@ -48,7 +48,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{current_app_context, current_ready_iam_handle, current_server_config_for_context};
 use crate::admin::utils::is_compat_admin_request;
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use http::StatusCode;
 use hyper::Method;
 use matchit::Params;
@@ -981,7 +981,7 @@ async fn authorize_action(req: &S3Request<Body>, cred: &StoredCredentials, owner
         owner,
         false,
         vec![Action::AdminAction(action)],
-        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        crate::admin::auth::policy_source_addr(&req.extensions),
     )
     .await
 }

rustfs/src/admin/handlers/is_admin.rs

@@ -66,7 +66,8 @@ impl Operation for IsAdminHandler {
             let empty_claims = HashMap::new();
             let iam_store = crate::admin::runtime_sources::current_ready_iam_handle()
                 .map_err(|_| s3_error!(InternalError, "iam not init"))?;
-            let conditions = get_condition_values(&req.headers, &cred, None, None, None);
+            let conditions =
+                get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
             iam_store
                 .is_allowed(&Args {
                     account: &cred.access_key,

rustfs/src/admin/handlers/kms_backup.rs

@@ -46,7 +46,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{current_deployment_id, current_kms_runtime_service_manager};
 use crate::admin::utils::json_response;
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use base64_simd::STANDARD as BASE64;
 use hyper::{Method, StatusCode};
 use matchit::Params;
@@ -923,7 +923,7 @@ async fn authorize(
             owner,
             false,
             actions,
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await,
         operation,

rustfs/src/admin/handlers/kms_dynamic.rs

@@ -28,7 +28,7 @@ use crate::admin::storage_api::error::StorageError;
 use crate::admin::storage_api::runtime::ECStore;
 use crate::admin::storage_api::s3::{S3ErrorCode, error as admin_s3_error};
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -97,6 +97,118 @@ fn kms_service_control_actions() -> Vec<Action> {
     vec![Action::KmsAction(KmsAction::ServiceControlAction)]
 }
 
+// A remote KMS administrator may reuse operator-provisioned files, but cannot
+// select new files or redirect the credentials read from them.
+struct VaultConnection<'a> {
+    address: &'a str,
+    namespace: Option<&'a str>,
+    auth: &'a rustfs_kms::VaultAuthMethod,
+    tls: Option<&'a rustfs_kms::config::TlsConfig>,
+}
+
+fn vault_connection(config: &KmsConfig) -> Option<VaultConnection<'_>> {
+    match &config.backend_config {
+        rustfs_kms::BackendConfig::VaultKv2(vault) => Some(VaultConnection {
+            address: &vault.address,
+            namespace: vault.namespace.as_deref(),
+            auth: &vault.auth_method,
+            tls: vault.tls.as_ref(),
+        }),
+        rustfs_kms::BackendConfig::VaultTransit(vault) => Some(VaultConnection {
+            address: &vault.address,
+            namespace: vault.namespace.as_deref(),
+            auth: &vault.auth_method,
+            tls: vault.tls.as_ref(),
+        }),
+        _ => None,
+    }
+}
+
+fn vault_auth_uses_file(auth: &rustfs_kms::VaultAuthMethod) -> bool {
+    use rustfs_kms::VaultAuthMethod;
+    matches!(
+        auth,
+        VaultAuthMethod::AppRole {
+            secret_id_file: Some(_),
+            ..
+        } | VaultAuthMethod::Kubernetes { .. }
+            | VaultAuthMethod::TokenFile { .. }
+    )
+}
+
+fn same_vault_file_auth(requested: &rustfs_kms::VaultAuthMethod, existing: &rustfs_kms::VaultAuthMethod) -> bool {
+    use rustfs_kms::VaultAuthMethod;
+    match (requested, existing) {
+        (
+            VaultAuthMethod::AppRole {
+                role_id,
+                secret_id_file,
+                mount,
+                ..
+            },
+            VaultAuthMethod::AppRole {
+                role_id: old_role,
+                secret_id_file: old_file,
+                mount: old_mount,
+                ..
+            },
+        ) => role_id == old_role && secret_id_file == old_file && mount == old_mount,
+        (
+            VaultAuthMethod::Kubernetes {
+                role, jwt_path, mount, ..
+            },
+            VaultAuthMethod::Kubernetes {
+                role: old_role,
+                jwt_path: old_path,
+                mount: old_mount,
+                ..
+            },
+        ) => role == old_role && jwt_path == old_path && mount == old_mount,
+        (VaultAuthMethod::TokenFile { path, .. }, VaultAuthMethod::TokenFile { path: old_path, .. }) => path == old_path,
+        _ => false,
+    }
+}
+
+fn validate_vault_local_file_access(
+    requested: &KmsConfig,
+    existing: Option<&KmsConfig>,
+    inherited_auth: bool,
+) -> Result<(), String> {
+    let Some(requested) = vault_connection(requested) else {
+        return Ok(());
+    };
+    let auth_file = vault_auth_uses_file(requested.auth);
+    let tls_files = requested
+        .tls
+        .is_some_and(|tls| tls.ca_cert_path.is_some() || tls.client_cert_path.is_some() || tls.client_key_path.is_some());
+    if !auth_file && !tls_files && !inherited_auth {
+        return Ok(());
+    }
+    let error = "Vault server-local files and reused credentials must retain their operator-configured connection";
+    let existing = existing.and_then(vault_connection).ok_or_else(|| error.to_string())?;
+    if requested.address != existing.address || requested.namespace != existing.namespace {
+        return Err(error.to_string());
+    }
+    if auth_file && !same_vault_file_auth(requested.auth, existing.auth) {
+        return Err(error.to_string());
+    }
+    let tls_settings = |tls: Option<&rustfs_kms::config::TlsConfig>| {
+        tls.map(|tls| {
+            (
+                tls.ca_cert_path.clone(),
+                tls.client_cert_path.clone(),
+                tls.client_key_path.clone(),
+                tls.skip_verify,
+            )
+        })
+        .unwrap_or_default()
+    };
+    if tls_settings(requested.tls) != tls_settings(existing.tls) {
+        return Err(error.to_string());
+    }
+    Ok(())
+}
+
 fn normalize_configure_request_secrets(
     request: &mut ConfigureKmsRequest,
     existing_config: Option<&KmsConfig>,
@@ -142,7 +254,7 @@ fn normalize_configure_request_secrets(
     };
 
     if !needs_existing_auth {
-        return Ok(());
+        return validate_vault_local_file_access(&request.to_kms_config(), existing_config, false);
     }
 
     let existing_auth = existing_config
@@ -157,7 +269,7 @@ fn normalize_configure_request_secrets(
         ConfigureKmsRequest::Aws(_) => {}
     }
 
-    Ok(())
+    validate_vault_local_file_access(&request.to_kms_config(), existing_config, true)
 }
 
 fn ensure_kms_config_persistable(config: &KmsConfig) -> Result<(), String> {
@@ -747,7 +859,7 @@ impl Operation for ConfigureKmsHandler {
                 owner,
                 false,
                 kms_configure_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAdminOperation::Configure,
@@ -900,7 +1012,7 @@ impl Operation for StartKmsHandler {
                 owner,
                 false,
                 kms_service_control_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAdminOperation::Start,
@@ -1057,7 +1169,7 @@ impl Operation for StopKmsHandler {
                 owner,
                 false,
                 kms_service_control_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAdminOperation::Stop,
@@ -1159,7 +1271,7 @@ impl Operation for ReloadKmsHandler {
                 owner,
                 false,
                 kms_service_control_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAdminOperation::Reload,
@@ -1253,7 +1365,7 @@ impl Operation for GetKmsStatusHandler {
             owner,
             false,
             kms_service_control_actions(),
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await?;
 
@@ -1347,7 +1459,7 @@ impl Operation for ReconfigureKmsHandler {
                 owner,
                 false,
                 kms_configure_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAdminOperation::Reconfigure,
@@ -1954,6 +2066,147 @@ mod tests {
         }
     }
 
+    fn vault_request(backend: &str, auth: serde_json::Value) -> serde_json::Value {
+        serde_json::json!({
+            "backend_type": backend,
+            "address": "https://vault.internal:8200",
+            "auth_method": auth,
+        })
+    }
+
+    fn decode_vault_request(value: serde_json::Value) -> rustfs_kms::ConfigureKmsRequest {
+        serde_json::from_value(value).expect("valid Vault configure request")
+    }
+
+    fn vault_file_auth_methods() -> [serde_json::Value; 3] {
+        [
+            serde_json::json!({"TokenFile": {"path": "/etc/rustfs/vault-token"}}),
+            serde_json::json!({"AppRole": {
+                "role_id": "rustfs", "secret_id": "", "secret_id_file": "/etc/rustfs/secret-id"
+            }}),
+            serde_json::json!({"Kubernetes": {"role": "rustfs", "jwt_path": "/etc/rustfs/service-account"}}),
+        ]
+    }
+
+    #[test]
+    fn vault_local_files_require_existing_operator_configuration() {
+        for backend in ["VaultKV2", "VaultTransit"] {
+            for auth in vault_file_auth_methods() {
+                let mut request = decode_vault_request(vault_request(backend, auth));
+                normalize_configure_request_secrets(&mut request, None)
+                    .expect_err("remote requests cannot bootstrap server-local credential reads");
+                let existing = vault_config("https://vault.internal:8200", VAULT_TOKEN);
+                normalize_configure_request_secrets(&mut request, Some(&existing))
+                    .expect_err("inline-token configuration grants no local-file access");
+            }
+            // Omitting jwt_path selects the pod's projected token, which is
+            // still a server-local credential and must not bypass the guard.
+            let mut request = decode_vault_request(vault_request(backend, serde_json::json!({"Kubernetes": {"role": "rustfs"}})));
+            normalize_configure_request_secrets(&mut request, None)
+                .expect_err("default Kubernetes path also needs operator configuration");
+        }
+    }
+
+    #[test]
+    fn vault_local_files_allow_unchanged_connections_but_reject_redirects() {
+        for backend in ["VaultKV2", "VaultTransit"] {
+            for auth in vault_file_auth_methods() {
+                let value = vault_request(backend, auth);
+                let existing = decode_vault_request(value.clone()).to_kms_config();
+                let mut request = decode_vault_request(value.clone());
+                normalize_configure_request_secrets(&mut request, Some(&existing))
+                    .expect("existing operator-provisioned credentials remain usable");
+
+                for (field, replacement) in [
+                    ("address", serde_json::json!("https://attacker.invalid:8200")),
+                    ("namespace", serde_json::json!("attacker")),
+                    ("skip_tls_verify", serde_json::json!(true)),
+                    ("ca_cert_path", serde_json::json!("/tmp/attacker-ca.pem")),
+                ] {
+                    let mut changed = value.clone();
+                    changed[field] = replacement;
+                    normalize_configure_request_secrets(&mut decode_vault_request(changed), Some(&existing))
+                        .expect_err("file credentials cannot be redirected or lose their trust boundary");
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn vault_local_files_reject_changed_auth_paths_and_routes() {
+        for backend in ["VaultKV2", "VaultTransit"] {
+            for (auth, variant, path) in [
+                (vault_file_auth_methods()[0].clone(), "TokenFile", "path"),
+                (vault_file_auth_methods()[1].clone(), "AppRole", "secret_id_file"),
+                (vault_file_auth_methods()[2].clone(), "Kubernetes", "jwt_path"),
+            ] {
+                let value = vault_request(backend, auth);
+                let existing = decode_vault_request(value.clone()).to_kms_config();
+                let mut changed = value.clone();
+                changed["auth_method"][variant][path] = serde_json::json!("/proc/self/environ");
+                normalize_configure_request_secrets(&mut decode_vault_request(changed), Some(&existing))
+                    .expect_err("the request cannot select an arbitrary server file");
+                if variant != "TokenFile" {
+                    let mut changed = value.clone();
+                    changed["auth_method"][variant]["mount"] = serde_json::json!("attacker-auth");
+                    normalize_configure_request_secrets(&mut decode_vault_request(changed), Some(&existing))
+                        .expect_err("file contents cannot be sent to a different authentication mount");
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn vault_local_files_reject_new_or_repurposed_tls_paths() {
+        for backend in ["VaultKV2", "VaultTransit"] {
+            for field in ["ca_cert_path", "client_cert_path", "client_key_path"] {
+                let mut value = vault_request(backend, serde_json::json!({"Token": {"token": "explicit-token"}}));
+                value[field] = serde_json::json!("/etc/rustfs/operator.pem");
+                let mut request = decode_vault_request(value.clone());
+                normalize_configure_request_secrets(&mut request, None).expect_err("TLS paths cannot be provisioned remotely");
+                let existing = request.to_kms_config();
+                normalize_configure_request_secrets(&mut request, Some(&existing)).expect("unchanged TLS paths remain usable");
+                value[field] = serde_json::json!("/proc/self/environ");
+                normalize_configure_request_secrets(&mut decode_vault_request(value), Some(&existing))
+                    .expect_err("TLS paths cannot be changed remotely");
+            }
+        }
+    }
+
+    #[test]
+    fn vault_local_files_validate_inherited_auth_after_normalization() {
+        for backend in ["VaultKV2", "VaultTransit"] {
+            for auth in vault_file_auth_methods()
+                .into_iter()
+                .chain([serde_json::json!({"Token": {"token": VAULT_TOKEN}})])
+            {
+                let existing = decode_vault_request(vault_request(backend, auth)).to_kms_config();
+                let value = vault_request(backend, serde_json::json!({"Token": {"token": "  "}}));
+                let mut request = decode_vault_request(value.clone());
+                normalize_configure_request_secrets(&mut request, Some(&existing))
+                    .expect("blank token preserves credentials for the same connection");
+                let mut changed = value;
+                changed["address"] = serde_json::json!("http://attacker.invalid:8200");
+                normalize_configure_request_secrets(&mut decode_vault_request(changed), Some(&existing))
+                    .expect_err("blank token cannot forward existing credentials elsewhere");
+            }
+        }
+    }
+
+    #[test]
+    fn vault_local_files_allow_explicit_credentials_without_files() {
+        for backend in ["VaultKV2", "VaultTransit"] {
+            for auth in [
+                serde_json::json!({"Token": {"token": "explicit-token"}}),
+                serde_json::json!({"AppRole": {"role_id": "role", "secret_id": "explicit-secret"}}),
+            ] {
+                let mut request = decode_vault_request(vault_request(backend, auth));
+                normalize_configure_request_secrets(&mut request, None)
+                    .expect("explicit credentials need no server-local access");
+            }
+        }
+    }
+
     #[test]
     fn config_fingerprint_input_carries_no_credentials() {
         let canonical = redacted_canonical_config(&vault_config("https://vault.internal:8200", VAULT_TOKEN))

rustfs/src/admin/handlers/kms_key_lifecycle.rs

@@ -21,7 +21,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_kms_runtime_service_manager;
 use crate::admin::utils::{extract_query_params, json_response};
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -270,7 +270,7 @@ async fn handle_lifecycle_request(
             owner,
             false,
             actions,
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
             scoped.as_deref().unwrap_or_default(),
         )
         .await,

rustfs/src/admin/handlers/kms_key_metadata.rs

@@ -25,7 +25,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_kms_runtime_service_manager;
 use crate::admin::utils::json_response;
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -327,7 +327,7 @@ async fn handle_metadata_request(
             owner,
             false,
             actions,
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
             scoped.as_deref().unwrap_or_default(),
         )
         .await,

rustfs/src/admin/handlers/kms_keys.rs

@@ -22,7 +22,7 @@ use crate::admin::storage_api::s3;
 use crate::admin::utils::extract_query_params;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::kms_deletion_gate::current_key_impact;
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use hyper::{HeaderMap, Method, StatusCode};
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
@@ -325,7 +325,7 @@ impl Operation for CreateKeyHandler {
                 owner,
                 false,
                 kms_create_key_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAuditOperation::CreateKey,
@@ -418,7 +418,7 @@ impl Operation for DescribeKeyHandler {
                 owner,
                 false,
                 kms_describe_key_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
                 requested_key_id.as_deref().unwrap_or_default(),
             )
             .await,
@@ -1488,7 +1488,7 @@ impl Operation for ListKeysHandler {
                 owner,
                 false,
                 kms_list_keys_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAuditOperation::ListKeys,
@@ -1583,7 +1583,7 @@ impl Operation for GenerateDataKeyHandler {
                 owner,
                 false,
                 kms_generate_data_key_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
                 parsed.as_ref().map(|request| request.key_id.as_str()).unwrap_or_default(),
             )
             .await,
@@ -1664,7 +1664,7 @@ impl Operation for CreateKmsKeyHandler {
                 owner,
                 false,
                 kms_create_key_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAuditOperation::CreateKey,
@@ -1943,7 +1943,7 @@ impl Operation for DeleteKmsKeyHandler {
                 owner,
                 false,
                 kms_delete_key_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
                 scoped.as_deref().unwrap_or_default(),
             )
             .await,
@@ -2111,7 +2111,7 @@ impl Operation for CancelKmsKeyDeletionHandler {
                 owner,
                 false,
                 kms_delete_key_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
                 scoped.as_deref().unwrap_or_default(),
             )
             .await,
@@ -2263,7 +2263,7 @@ impl Operation for ListKmsKeysHandler {
                 owner,
                 false,
                 kms_list_keys_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
             )
             .await,
             KmsAuditOperation::ListKeys,
@@ -2459,7 +2459,7 @@ impl Operation for DescribeKmsKeyHandler {
                 owner,
                 false,
                 kms_describe_key_actions(),
-                req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                crate::admin::auth::policy_source_addr(&req.extensions),
                 params.get("key_id").unwrap_or_default(),
             )
             .await,

rustfs/src/admin/handlers/mfa.rs

@@ -51,7 +51,6 @@ use crate::admin::storage_api::runtime::ECStore;
 use crate::admin::storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
 use crate::admin::utils::read_compatible_admin_body;
 use crate::auth::constant_time_eq;
-use crate::server::RemoteAddr;
 use http::StatusCode;
 use hyper::Method;
 use matchit::Params;
@@ -484,7 +483,7 @@ impl Operation for UserMfaStatusHandler {
             caller.is_owner,
             false,
             vec![Action::AdminAction(AdminAction::GetUserAdminAction)],
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await?;
 
@@ -521,7 +520,7 @@ impl Operation for UserMfaResetHandler {
             caller.is_owner,
             false,
             vec![Action::AdminAction(AdminAction::EnableUserAdminAction)],
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await
         .inspect_err(|_| {

rustfs/src/admin/handlers/plugin_delegation.rs

@@ -11,7 +11,7 @@ use crate::admin::{
 };
 use crate::{
     auth::{check_key_valid_with_context, get_session_token},
-    server::{ADMIN_PREFIX, RemoteAddr},
+    server::ADMIN_PREFIX,
 };
 use http::{HeaderMap, HeaderValue, Method, StatusCode};
 use matchit::Params;
@@ -97,7 +97,7 @@ pub(super) async fn owner(
         &credentials,
         false,
         Action::AdminAction(action),
-        req.extensions.get::<Option<RemoteAddr>>().and_then(|a| a.map(|a| a.0)),
+        crate::admin::auth::policy_source_addr(&req.extensions),
         AdminResourceScope::bucket_object("", ""),
     )
     .await?;

rustfs/src/admin/handlers/quota.rs

@@ -30,7 +30,6 @@ use crate::admin::storage_api::contract::bucket::{BucketOperations as _, BucketO
 use crate::auth::{check_key_valid, get_session_token};
 use crate::error::ApiError;
 use crate::server::ADMIN_PREFIX;
-use crate::server::RemoteAddr;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
@@ -397,7 +396,7 @@ impl Operation for SetBucketQuotaHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
         validate_admin_request(
             &req.headers,
             &cred,
@@ -538,7 +537,7 @@ impl Operation for GetBucketQuotaHandler {
         if bucket.is_empty() {
             return Err(s3_error!(InvalidRequest, "bucket name is required"));
         }
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
         validate_admin_request_with_bucket(
             &req.headers,
             &cred,
@@ -607,7 +606,7 @@ impl Operation for ClearBucketQuotaHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
         validate_admin_request(
             &req.headers,
             &cred,
@@ -726,7 +725,7 @@ impl Operation for GetBucketQuotaStatsHandler {
             return Err(s3_error!(InvalidRequest, "bucket name is required"));
         }
 
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
         validate_admin_request_with_bucket(
             &req.headers,
             &cred,
@@ -795,7 +794,7 @@ impl Operation for CheckBucketQuotaHandler {
             return Err(s3_error!(InvalidRequest, "bucket name is required"));
         }
 
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
         validate_admin_request_with_bucket(
             &req.headers,
             &cred,
@@ -890,7 +889,7 @@ impl Operation for MigrateBucketQuotaProtocolHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
         validate_admin_request(
             &req.headers,
             &cred,

rustfs/src/admin/handlers/service_account.rs

@@ -18,7 +18,7 @@ use crate::admin::handlers::site_replication::{encode_service_account_replicatio
 use crate::admin::runtime_sources::current_action_credentials;
 use crate::admin::utils::{encode_compatible_admin_payload, has_space_be, is_compat_admin_request, read_compatible_admin_body};
 use crate::auth::{constant_time_eq, get_condition_values, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use crate::{
     admin::router::{AdminOperation, Operation, S3Router},
     auth::check_key_valid,
@@ -84,6 +84,25 @@ fn parse_new_service_account_policy(policy: Option<&serde_json::Value>) -> S3Res
     Ok((!policy.id.is_empty() || !policy.version.is_empty() || !policy.statements.is_empty()).then_some(policy))
 }
 
+/// Service accounts cannot delegate beyond their own lifetime. STS sessions
+/// retain the existing ability to create independently expiring service accounts.
+fn bounded_service_account_expiration(caller: &StoredCredentials, requested: Option<OffsetDateTime>) -> Option<OffsetDateTime> {
+    match caller.expiration.filter(|_| caller.is_service_account()) {
+        Some(limit) => Some(requested.map_or(limit, |expiration| expiration.min(limit))),
+        None => requested,
+    }
+}
+
+fn set_service_account_duration_condition(
+    conditions: &mut HashMap<String, Vec<String>>,
+    expiration: Option<OffsetDateTime>,
+    now: OffsetDateTime,
+) {
+    // An unlimited credential must not bypass a finite NumericGreaterThan Deny.
+    let seconds = expiration.map_or(i64::MAX, |expiration| (expiration - now).whole_seconds().max(0));
+    conditions.insert("DurationSeconds".to_string(), vec![seconds.to_string()]);
+}
+
 fn compat_time_sentinel() -> OffsetDateTime {
     OffsetDateTime::UNIX_EPOCH
 }
@@ -322,9 +341,13 @@ impl Operation for AddServiceAccount {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &req_cred.access_key).await?;
 
+        if !constant_time_eq(&cred.secret_key, req_cred.secret_key.expose()) {
+            return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "access denied"));
+        }
+
         let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
 
-        let create_req: AddServiceAccountReq =
+        let mut create_req: AddServiceAccountReq =
             serde_json::from_slice(&body[..]).map_err(|e| s3_error!(InvalidRequest, "unmarshal body failed, e: {:?}", e))?;
 
         // create_req.expiration = create_req.expiration.and_then(|expire| expire.replace_millisecond(0).ok());
@@ -370,6 +393,11 @@ impl Operation for AddServiceAccount {
             return Err(s3_error!(InvalidRequest, "iam not init"));
         };
 
+        create_req.expiration = bounded_service_account_expiration(&cred, create_req.expiration);
+        let mut conditions =
+            get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
+        set_service_account_duration_condition(&mut conditions, create_req.expiration, OffsetDateTime::now_utc());
+
         // This family deliberately calls `is_allowed` directly instead of going
         // through `validate_admin_request`, and must keep doing so
         // (backlog#1886). The shared helper returns as soon as *any* candidate
@@ -383,13 +411,7 @@ impl Operation for AddServiceAccount {
                 groups: &cred.groups,
                 action: Action::AdminAction(AdminAction::CreateServiceAccountAdminAction),
                 bucket: "",
-                conditions: &get_condition_values(
-                    &req.headers,
-                    &cred,
-                    None,
-                    None,
-                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
-                ),
+                conditions: &conditions,
                 is_owner: owner,
                 object: "",
                 claims: cred.claims_or_empty(),
@@ -444,17 +466,9 @@ impl Operation for AddServiceAccount {
         let replication_description = opts.description.clone().unwrap_or_default();
         let replication_expiration = opts.expiration;
 
-        let create_result = if is_svc_acc {
-            iam_store
-                .new_service_account_from_caller(&target_user, target_groups, opts, &cred)
-                .await
-        } else {
-            let replication_claims = opts.claims.clone().unwrap_or_default();
-            iam_store
-                .new_service_account(&target_user, target_groups, opts)
-                .await
-                .map(|(credentials, updated_at)| (credentials, updated_at, replication_claims))
-        };
+        let create_result = iam_store
+            .new_service_account_from_caller(&target_user, target_groups, opts, &cred)
+            .await;
         let (new_cred, updated_at, replication_claims) = create_result.map_err(|e| {
             debug!(
                 component = LOG_COMPONENT_ADMIN,
@@ -466,7 +480,8 @@ impl Operation for AddServiceAccount {
                 "admin service account state"
             );
             match e {
-                rustfs_iam::error::Error::InvalidAccessKeyLength
+                rustfs_iam::error::Error::InvalidArgument
+                | rustfs_iam::error::Error::InvalidAccessKeyLength
                 | rustfs_iam::error::Error::InvalidSecretKeyLength
                 | rustfs_iam::error::Error::AccessKeyAlreadyExists => iam_error_to_s3_error(e),
                 rustfs_iam::error::Error::IAMActionNotAllowed => s3_error!(AccessDenied, "access denied"),
@@ -599,6 +614,9 @@ impl Operation for UpdateServiceAccount {
             return Err(s3_error!(InvalidRequest, "get cred failed"));
         };
 
+        let (cred, owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
         let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
             return Err(s3_error!(InvalidRequest, "iam not init"));
         };
@@ -608,17 +626,29 @@ impl Operation for UpdateServiceAccount {
         //     s3_error!(InternalError, "get service account failed")
         // })?;
 
-        let body =
-            read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), input_cred.secret_key.expose())
-                .await?;
+        let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;
 
-        let update_req: UpdateServiceAccountReq =
+        let mut update_req: UpdateServiceAccountReq =
             serde_json::from_slice(&body[..]).map_err(|e| s3_error!(InvalidRequest, "unmarshal body failed, e: {:?}", e))?;
 
         update_req.validate().map_err(|e| S3Error::with_message(InvalidRequest, e))?;
 
-        let (cred, owner) =
-            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+        // Keep lookup errors behind authorization even though lifetime conditions
+        // need the existing expiration when the request omits it.
+        let existing_account = iam_store.get_service_account(&access_key).await;
+
+        if update_req.new_expiration.is_some() {
+            update_req.new_expiration = bounded_service_account_expiration(&cred, update_req.new_expiration);
+        }
+        let mut conditions =
+            get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
+        set_service_account_duration_condition(
+            &mut conditions,
+            update_req
+                .new_expiration
+                .or_else(|| existing_account.as_ref().ok().and_then(|(account, _)| account.expiration)),
+            OffsetDateTime::now_utc(),
+        );
 
         if !iam_store
             .is_allowed(&Args {
@@ -626,13 +656,7 @@ impl Operation for UpdateServiceAccount {
                 groups: &cred.groups,
                 action: Action::AdminAction(AdminAction::UpdateServiceAccountAdminAction),
                 bucket: "",
-                conditions: &get_condition_values(
-                    &req.headers,
-                    &cred,
-                    None,
-                    None,
-                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
-                ),
+                conditions: &conditions,
                 is_owner: owner,
                 object: "",
                 claims: cred.claims_or_empty(),
@@ -643,10 +667,7 @@ impl Operation for UpdateServiceAccount {
             return Err(s3_error!(AccessDenied, "access denied"));
         }
 
-        let (svc_account, _) = iam_store
-            .get_service_account(&access_key)
-            .await
-            .map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
+        let (svc_account, _) = existing_account.map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
 
         // The admin action permits updates within a caller's scope; only an
         // owner may cross service-account parent boundaries.
@@ -723,11 +744,6 @@ impl Operation for InfoServiceAccount {
             return Err(s3_error!(InvalidRequest, "iam not init"));
         };
 
-        let (svc_account, session_policy) = iam_store
-            .get_service_account(&access_key)
-            .await
-            .map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
-
         let Some(input_cred) = req.credentials else {
             return Err(s3_error!(InvalidRequest, "get cred failed"));
         };
@@ -735,6 +751,11 @@ impl Operation for InfoServiceAccount {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
 
+        let (svc_account, session_policy) = iam_store
+            .get_service_account(&access_key)
+            .await
+            .map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;
+
         if !iam_store
             .is_allowed(&Args {
                 account: &cred.access_key,
@@ -746,7 +767,7 @@ impl Operation for InfoServiceAccount {
                     &cred,
                     None,
                     None,
-                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                    crate::admin::auth::policy_source_addr(&req.extensions),
                 ),
                 is_owner: owner,
                 object: "",
@@ -811,7 +832,7 @@ impl Operation for TemporaryAccountInfo {
                     &cred,
                     None,
                     None,
-                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                    crate::admin::auth::policy_source_addr(&req.extensions),
                 ),
                 is_owner: owner,
                 object: "",
@@ -872,13 +893,8 @@ impl Operation for InfoAccessKey {
 
         let target_cred = iam_store.get_user(&access_key).await.map(|identity| identity.credentials);
 
-        let conditions = get_condition_values(
-            &req.headers,
-            &cred,
-            None,
-            None,
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
-        );
+        let conditions =
+            get_condition_values(&req.headers, &cred, None, None, crate::admin::auth::policy_source_addr(&req.extensions));
         let claims = cred.claims_or_empty();
 
         if !iam_store
@@ -1007,7 +1023,7 @@ impl Operation for ListServiceAccount {
                         &cred,
                         None,
                         None,
-                        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                        crate::admin::auth::policy_source_addr(&req.extensions),
                     ),
                     is_owner: owner,
                     object: "",
@@ -1151,7 +1167,7 @@ impl Operation for ListAccessKeysBulk {
                         &cred,
                         None,
                         None,
-                        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                        crate::admin::auth::policy_source_addr(&req.extensions),
                     ),
                     is_owner: owner,
                     object: "",
@@ -1174,7 +1190,7 @@ impl Operation for ListAccessKeysBulk {
                     &cred,
                     None,
                     None,
-                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                    crate::admin::auth::policy_source_addr(&req.extensions),
                 ),
                 is_owner: owner,
                 object: "",
@@ -1355,7 +1371,7 @@ impl Operation for DeleteServiceAccount {
                     &cred,
                     None,
                     None,
-                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+                    crate::admin::auth::policy_source_addr(&req.extensions),
                 ),
                 is_owner: owner,
                 object: "",
@@ -1444,12 +1460,168 @@ mod tests {
     use serde_json::json;
     use serde_urlencoded::from_bytes;
 
+    #[test]
+    #[serial_test::serial]
+    fn service_account_handlers_authenticate_before_decryption_or_lookup() {
+        crate::app::gating_test_env::run_large_stack_test("service-account-auth-first", || async {
+            use crate::admin::runtime_sources::IamInterface;
+            use crate::admin::runtime_sources::{AppContext, publish_test_app_context};
+            use rustfs_iam::store::object::ObjectStore;
+            use std::sync::Arc;
+            struct ReadyIam(Arc<rustfs_iam::sys::IamSys<ObjectStore>>);
+            impl IamInterface for ReadyIam {
+                fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<ObjectStore>> {
+                    self.0.clone()
+                }
+                fn is_ready(&self) -> bool {
+                    true
+                }
+            }
+            let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+            let (_directory, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+            ObjectStore::new(store.clone())
+                .save_iam_config(
+                    serde_json::json!({"version": 1}),
+                    format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+                )
+                .await
+                .expect("initialize IAM storage");
+            let iam = rustfs_iam::build_iam_sys(store.clone()).await.expect("initialized IAM");
+            let disabled_key = format!("disabled-{}", uuid::Uuid::new_v4());
+            iam.create_user(
+                &disabled_key,
+                &rustfs_madmin::AddOrUpdateUserReq {
+                    secret_key: "disabled-user-secret".to_owned(),
+                    policy: None,
+                    status: rustfs_madmin::AccountStatus::Disabled,
+                },
+            )
+            .await
+            .expect("create disabled IAM caller");
+            let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(iam)), ambient.kms()));
+            assert!(context.publish_action_credentials(StoredCredentials {
+                access_key: "auth-first-root".to_owned(),
+                secret_key: "auth-first-root-secret".to_owned(),
+                status: "on".to_owned(),
+                ..Default::default()
+            }));
+            publish_test_app_context(context);
+            for caller in ["keystone:forged", disabled_key.as_str()] {
+                for operation in ["update-service-account", "info-service-account"] {
+                    let req = S3Request {
+                        input: Body::from(vec![0u8; 64]),
+                        method: if operation == "update-service-account" {
+                            http::Method::POST
+                        } else {
+                            http::Method::GET
+                        },
+                        uri: format!("/minio/admin/v3/{operation}?accessKey=missing-target")
+                            .parse()
+                            .unwrap(),
+                        headers: HeaderMap::new(),
+                        extensions: http::Extensions::new(),
+                        credentials: Some(s3s::auth::Credentials {
+                            access_key: caller.to_owned(),
+                            secret_key: s3s::auth::SecretKey::from("disabled-user-secret"),
+                        }),
+                        region: None,
+                        service: None,
+                        trailing_headers: None,
+                    };
+                    let result = if operation == "update-service-account" {
+                        UpdateServiceAccount {}.call(req, Params::new()).await
+                    } else {
+                        InfoServiceAccount {}.call(req, Params::new()).await
+                    };
+                    assert_eq!(
+                        result
+                            .expect_err("invalid identity precedes KDF and missing target lookup")
+                            .code(),
+                        &s3s::S3ErrorCode::InvalidAccessKeyId,
+                        "{caller}: {operation}"
+                    );
+                }
+            }
+            publish_test_app_context(ambient);
+        });
+    }
+
     #[test]
     fn access_key_query_supports_external_alias() {
         let query: AccessKeyQuery = from_bytes(b"access-key=test-access-key").expect("parse query");
         assert_eq!(query.access_key, "test-access-key");
     }
 
+    #[test]
+    fn service_account_cannot_delegate_a_longer_lifetime() {
+        let deadline = OffsetDateTime::UNIX_EPOCH + time::Duration::days(10);
+        let earlier = deadline - time::Duration::hours(1);
+        let later = deadline + time::Duration::days(365);
+        let mut caller = StoredCredentials {
+            access_key: "expiring-service".to_string(),
+            parent_user: "parent".to_string(),
+            expiration: Some(deadline),
+            claims: Some(HashMap::from([("sa-policy".to_string(), json!("inherited-policy"))])),
+            ..Default::default()
+        };
+        assert_eq!(bounded_service_account_expiration(&caller, None), Some(deadline));
+        assert_eq!(bounded_service_account_expiration(&caller, Some(later)), Some(deadline));
+        assert_eq!(bounded_service_account_expiration(&caller, Some(earlier)), Some(earlier));
+        caller.expiration = None;
+        assert_eq!(bounded_service_account_expiration(&caller, None), None);
+        caller.expiration = Some(deadline);
+        caller.claims = None;
+        caller.session_token = "signed-sts".to_string();
+        assert_eq!(
+            bounded_service_account_expiration(&caller, None),
+            None,
+            "STS retains independent service lifetime"
+        );
+        assert_eq!(bounded_service_account_expiration(&caller, Some(later)), Some(later));
+    }
+
+    #[tokio::test]
+    async fn service_account_duration_condition_enforces_finite_and_unlimited_lifetimes() {
+        let policy = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"],"Resource":"*"},
+            {"Effect":"Deny","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"],"Resource":"*",
+             "Condition":{"NumericGreaterThan":{"svc:DurationSeconds":"3600"}}}
+        ]}"#,
+        )
+        .expect("service lifetime policy");
+        let now = OffsetDateTime::UNIX_EPOCH;
+        for (expiration, allowed) in [
+            (Some(now + time::Duration::seconds(3600)), true),
+            (Some(now + time::Duration::seconds(3601)), false),
+            (None, false),
+        ] {
+            let mut conditions = HashMap::from([("DurationSeconds".to_string(), vec!["1".to_string()])]);
+            set_service_account_duration_condition(&mut conditions, expiration, now);
+            for action in [
+                AdminAction::CreateServiceAccountAdminAction,
+                AdminAction::UpdateServiceAccountAdminAction,
+            ] {
+                assert_eq!(
+                    policy
+                        .is_allowed(&Args {
+                            account: "user",
+                            groups: &None,
+                            action: Action::AdminAction(action),
+                            conditions: &conditions,
+                            is_owner: false,
+                            claims: &HashMap::new(),
+                            deny_only: false,
+                            bucket: "",
+                            object: "",
+                        })
+                        .await,
+                    allowed
+                );
+            }
+        }
+    }
+
     #[test]
     fn non_admin_delete_is_fail_closed_on_unloadable_target() {
         // Regression (backlog#806): a non-admin caller may delete ONLY a service

rustfs/src/admin/handlers/sts.rs

@@ -27,7 +27,6 @@ use crate::{
     },
     auth::{check_key_valid, get_session_token},
     server::ADMIN_PREFIX,
-    server::RemoteAddr,
 };
 use http::StatusCode;
 use http::header::HeaderValue;
@@ -54,6 +53,7 @@ use s3s::{
 use serde::Deserialize;
 use serde_json::Value;
 use serde_urlencoded::from_bytes;
+use std::collections::HashMap;
 use time::{Duration, OffsetDateTime};
 use tracing::{debug, error, warn};
 
@@ -82,6 +82,12 @@ fn clamp_assume_role_duration(duration_seconds: usize) -> usize {
     }
 }
 
+fn set_assume_role_duration_condition(conditions: &mut HashMap<String, Vec<String>>, requested: usize) -> usize {
+    let duration = clamp_assume_role_duration(requested);
+    conditions.insert("DurationSeconds".to_string(), vec![duration.to_string()]);
+    duration
+}
+
 /// Record that AssumeRole assembled its session claims without echoing any claim values.
 ///
 /// Claims carry caller-supplied identity material (parent user, session policy, arbitrary
@@ -186,7 +192,7 @@ impl Operation for AssumeRoleHandle {
 
         match body.action.as_str() {
             ASSUME_ROLE_ACTION => {
-                let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+                let remote_addr = crate::admin::auth::policy_source_addr(&req.extensions);
                 handle_assume_role(req.credentials, req.uri, req.headers, remote_addr, body, store, &audit).await
             }
             ASSUME_ROLE_WITH_WEB_IDENTITY_ACTION => handle_assume_role_with_web_identity(body).await,
@@ -216,6 +222,12 @@ async fn handle_assume_role(
 
     let (cred, owner) = check_key_valid(get_session_token(&uri, &headers).unwrap_or_default(), &user.access_key).await?;
 
+    // The signature was verified before the request body was consumed. A fresh
+    // IAM lookup must not silently replace the secret that authenticated it.
+    if !crate::auth::constant_time_eq(&cred.secret_key, user.secret_key.expose()) {
+        return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied"));
+    }
+
     if cred.is_temp() || cred.is_service_account() {
         return Err(s3_error!(AccessDenied, "Access Denied"));
     }
@@ -223,7 +235,8 @@ async fn handle_assume_role(
     let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
         return Err(s3_error!(InvalidRequest, "iam not init"));
     };
-    let conditions = crate::auth::get_condition_values(&headers, &cred, None, None, remote_addr);
+    let mut conditions = crate::auth::get_condition_values(&headers, &cred, None, None, remote_addr);
+    let exp = set_assume_role_duration_condition(&mut conditions, body.duration_seconds);
     if !iam_store
         .is_allowed(&Args {
             account: &cred.access_key,
@@ -260,7 +273,7 @@ async fn handle_assume_role(
     // behaviour of every existing deployment is unchanged.
     let mfa_verified = enforce_second_factor(&cred.access_key, &body, store, audit).await?;
 
-    let mut claims = cred.claims.unwrap_or_default();
+    let mut claims = cred.claims.clone().unwrap_or_default();
 
     if mfa_verified {
         // Recorded on the session so a later policy condition can require it,
@@ -271,8 +284,6 @@ async fn handle_assume_role(
 
     populate_session_policy(&mut claims, &body.policy)?;
 
-    let exp = clamp_assume_role_duration(body.duration_seconds);
-
     claims.insert(
         "exp".to_string(),
         Value::Number(serde_json::Number::from(
@@ -309,9 +320,9 @@ async fn handle_assume_role(
     );
 
     let updated_at = iam_store
-        .set_temp_user(&new_cred.access_key, &new_cred, None)
+        .set_temp_user_from_parent(&new_cred, &cred)
         .await
-        .map_err(|_| s3_error!(InternalError, "set_temp_user failed"))?;
+        .map_err(super::iam_error::iam_error_to_s3_error)?;
 
     let root_access_key = current_action_credentials().map(|cred| cred.access_key);
     if root_access_key.as_deref() != Some(new_cred.parent_user.as_str())
@@ -604,6 +615,40 @@ mod tests {
         assert_eq!(clamp(999999), 43200); // clamped to max
     }
 
+    #[tokio::test]
+    async fn assume_role_duration_condition_enforces_policy_after_clamping() {
+        use rustfs_policy::policy::Policy;
+        let policy = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[{
+            "Effect":"Deny","Action":"sts:AssumeRole","Resource":"*",
+            "Condition":{"NumericGreaterThan":{"sts:DurationSeconds":"3600"}}
+        }]}"#,
+        )
+        .expect("duration policy");
+        for (requested, allowed) in [(0, true), (1, true), (3600, true), (3601, false), (usize::MAX, false)] {
+            let mut conditions = HashMap::from([("DurationSeconds".to_string(), vec!["1".to_string()])]);
+            let lifetime = set_assume_role_duration_condition(&mut conditions, requested);
+            assert_eq!(conditions["DurationSeconds"], [lifetime.to_string()]);
+            assert_eq!(
+                policy
+                    .is_allowed(&Args {
+                        account: "user",
+                        groups: &None,
+                        action: Action::StsAction(StsAction::AssumeRoleAction),
+                        conditions: &conditions,
+                        is_owner: false,
+                        claims: &HashMap::new(),
+                        deny_only: true,
+                        bucket: "",
+                        object: "",
+                    })
+                    .await,
+                allowed,
+                "requested={requested}"
+            );
+        }
+    }
+
     #[test]
     fn test_assume_role_duration_is_clamped_to_max() {
         // Regression: the standard AssumeRole path previously used the raw client-supplied

rustfs/src/admin/handlers/table_catalog/mod.rs

@@ -24,7 +24,7 @@ use crate::admin::{
 };
 use crate::auth::{check_key_valid_with_context, get_session_token};
 use crate::error::ApiError;
-use crate::server::{RemoteAddr, TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX};
+use crate::server::{TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX};
 use crate::table_catalog::{DEFAULT_WAREHOUSE_ID, TableCatalogStore};
 use bytes::Bytes;
 use futures::{StreamExt, TryStreamExt, stream};
@@ -1141,7 +1141,7 @@ async fn authorize_table_catalog_request(req: &S3Request<Body>, action: AdminAct
         &principal.credentials,
         principal.owner,
         Action::AdminAction(action),
-        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        crate::admin::auth::policy_source_addr(&req.extensions),
         AdminResourceScope::bucket(""),
     )
     .await
@@ -1255,7 +1255,7 @@ async fn authorize_table_catalog_object_for_principal(
         &principal.credentials,
         principal.owner,
         Action::AdminAction(action),
-        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        crate::admin::auth::policy_source_addr(&req.extensions),
         AdminResourceScope::bucket_object(bucket, object),
     )
     .await

rustfs/src/admin/handlers/usage_prefix.rs

@@ -19,7 +19,7 @@
 //! usage caches, with a one-level sub-prefix breakdown — the data console
 //! buckets view MinIO serves from `loadPrefixUsageFromBackend`.
 
-use crate::admin::auth::authorize_admin_request;
+use crate::admin::auth::{authenticate_request, policy_source_addr, validate_admin_request_with_bucket};
 use crate::admin::handlers::system::data_usage_info_gate_actions;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::server::ADMIN_PREFIX;
@@ -69,15 +69,24 @@ fn parse_usage_prefix_query(query: Option<&str>) -> S3Result<(String, usize)> {
 #[async_trait::async_trait]
 impl Operation for BucketPrefixUsageHandler {
     async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        // The shared gate reports the same `InvalidRequest` "get cred failed" this
-        // handler has always returned for a credential-less request, so it needs no
-        // message-preserving pre-check.
-        authorize_admin_request(&req, data_usage_info_gate_actions()).await?;
-
+        let Some(input_cred) = req.credentials.as_ref() else {
+            return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "get cred failed"));
+        };
+        let (cred, owner) = authenticate_request(&req.headers, &req.uri, input_cred).await?;
         let bucket = params.get("bucket").unwrap_or_default().to_string();
         if bucket.is_empty() {
             return Err(s3_error!(InvalidRequest, "bucket path parameter is required"));
         }
+        validate_admin_request_with_bucket(
+            &req.headers,
+            &cred,
+            owner,
+            false,
+            data_usage_info_gate_actions(),
+            policy_source_addr(&req.extensions),
+            &bucket,
+        )
+        .await?;
         let (prefix, max_entries) = parse_usage_prefix_query(req.uri.query())?;
 
         // Authorization is bucket-scoped by the same any-of gate as the

rustfs/src/admin/handlers/user.rs

@@ -23,7 +23,6 @@ use crate::{
         utils::{encode_compatible_admin_payload, has_space_be, read_compatible_admin_body},
     },
     auth::{check_key_valid, constant_time_eq, get_session_token},
-    server::RemoteAddr,
 };
 use http::{HeaderMap, StatusCode};
 use matchit::Params;
@@ -31,7 +30,7 @@ use rustfs_config::{MAX_ADMIN_REQUEST_BODY_SIZE, MAX_IAM_IMPORT_EXPANDED_SIZE, M
 use rustfs_credentials::Credentials;
 use rustfs_iam::{
     store::{GroupInfo, MappedPolicy, UserType},
-    sys::{NewServiceAccountOpts, UpdateServiceAccountOpts},
+    sys::{NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts},
 };
 use rustfs_madmin::{
     AccountStatus, AddOrUpdateUserReq, IAMEntities, IAMErrEntities, IAMErrEntity, IAMErrPolicyEntity, SITE_REPL_API_VERSION,
@@ -196,6 +195,19 @@ fn imported_service_account_parent_scope_failure(
     })
 }
 
+fn retain_exportable_service_accounts(accounts: &mut HashMap<String, rustfs_policy::auth::UserIdentity>) {
+    accounts.retain(|key, account| {
+        key != SITE_REPLICATOR_SERVICE_ACCOUNT && account.credentials.access_key != SITE_REPLICATOR_SERVICE_ACCOUNT
+    });
+}
+
+fn imported_service_account_reserved_failure(access_key: &str) -> Option<IAMErrEntity> {
+    (access_key == SITE_REPLICATOR_SERVICE_ACCOUNT).then(|| IAMErrEntity {
+        name: access_key.to_owned(),
+        error: "service account name is reserved for site replication".to_owned(),
+    })
+}
+
 fn imported_service_account_access_key_failure(entry_access_key: &str, payload_access_key: &str) -> Option<IAMErrEntity> {
     (entry_access_key != payload_access_key).then(|| IAMErrEntity {
         name: entry_access_key.to_string(),
@@ -208,6 +220,7 @@ pub struct AddUser {}
 fn map_add_user_create_error(err: rustfs_iam::error::Error) -> S3Error {
     match err {
         rustfs_iam::error::Error::AccessKeyAlreadyExists => iam_error_to_s3_error(err),
+        rustfs_iam::error::Error::IAMActionNotAllowed => S3Error::new(S3ErrorCode::AccessDenied),
         err => S3Error::with_message(S3ErrorCode::InternalError, format!("failed to create user: {err}")),
     }
 }
@@ -243,7 +256,7 @@ impl Operation for AddUser {
         // let body_bytes = decrypt_data(input_cred.secret_key.expose().as_bytes(), &body)
         //     .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, format!("decrypt_data err {}", e)))?;
 
-        let args: AddOrUpdateUserReq = serde_json::from_slice(&body)
+        let mut args: AddOrUpdateUserReq = serde_json::from_slice(&body)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid JSON: {e}")))?;
 
         if args.secret_key.is_empty() {
@@ -261,7 +274,7 @@ impl Operation for AddUser {
         };
 
         if let Some(user) = iam_store.get_user(ak).await {
-            if (user.credentials.is_temp() || user.credentials.is_service_account()) && cred.parent_user == ak {
+            if user.credentials.is_temp() || user.credentials.is_service_account() {
                 return Err(s3_error!(InvalidArgument, "cannot create a user with a service account access key"));
             }
         } else if has_space_be(ak) {
@@ -306,11 +319,22 @@ impl Operation for AddUser {
             owner,
             check_deny_only,
             vec![Action::AdminAction(AdminAction::CreateUserAdminAction)],
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await?;
 
-        let updated_at = iam_store.create_user(ak, &args).await.map_err(map_add_user_create_error)?;
+        let updated_at = if check_deny_only {
+            let (updated_at, status, _) = iam_store
+                .change_own_secret_key(ak, input_cred.secret_key.expose(), &args.secret_key)
+                .await
+                .map_err(map_add_user_create_error)?;
+            args.status = status;
+            // A self-service secret rotation does not change policy attachment.
+            args.policy = None;
+            updated_at
+        } else {
+            iam_store.create_user(ak, &args).await.map_err(map_add_user_create_error)?
+        };
 
         if let Err(err) = site_replication_iam_change_hook(SRIAMItem {
             r#type: "iam-user".to_string(),
@@ -619,7 +643,7 @@ impl Operation for GetUserInfo {
             owner,
             check_deny_only,
             vec![Action::AdminAction(AdminAction::GetUserAdminAction)],
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await?;
 
@@ -749,12 +773,9 @@ impl Operation for ExportIam {
                         .await
                         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
 
+                    retain_exportable_service_accounts(&mut service_accounts);
                     let mut svc_accts: HashMap<String, SRSvcAccCreate> = HashMap::new();
                     for (k, acc) in service_accounts {
-                        if k == "siteReplicatorSvcAcc" {
-                            continue;
-                        }
-
                         let claims = iam_store
                             .get_claims_for_svc_acc(&acc.credentials.access_key)
                             .await
@@ -904,7 +925,7 @@ impl Operation for ImportIam {
             owner,
             false,
             vec![Action::AdminAction(AdminAction::ImportIAMAction)],
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            crate::admin::auth::policy_source_addr(&req.extensions),
         )
         .await?;
 
@@ -1089,6 +1110,11 @@ impl Operation for ImportIam {
                         return Err(s3_error!(InvalidArgument, "has space be {ak}"));
                     }
 
+                    if let Some(err) = imported_service_account_reserved_failure(&ak) {
+                        failed.service_accounts.push(err);
+                        continue;
+                    }
+
                     if let Some(err) = imported_service_account_access_key_failure(&ak, &req.access_key) {
                         failed.service_accounts.push(err);
                         continue;
@@ -1463,6 +1489,74 @@ mod tests {
         }
     }
 
+    #[test]
+    fn iam_export_excludes_site_replicator_credentials() {
+        let mut accounts = HashMap::from([
+            (
+                "site-replicator-0".to_owned(),
+                rustfs_policy::auth::UserIdentity::new(Credentials {
+                    access_key: "site-replicator-0".to_owned(),
+                    secret_key: "replicator-secret".to_owned(),
+                    ..Default::default()
+                }),
+            ),
+            (
+                "ordinary-service".to_owned(),
+                rustfs_policy::auth::UserIdentity::new(Credentials {
+                    access_key: "ordinary-service".to_owned(),
+                    secret_key: "ordinary-secret".to_owned(),
+                    ..Default::default()
+                }),
+            ),
+            (
+                "mismatched-entry".to_owned(),
+                rustfs_policy::auth::UserIdentity::new(Credentials {
+                    access_key: "site-replicator-0".to_owned(),
+                    secret_key: "aliased-replicator-secret".to_owned(),
+                    ..Default::default()
+                }),
+            ),
+        ]);
+        super::retain_exportable_service_accounts(&mut accounts);
+        assert_eq!(accounts.len(), 1);
+        assert_eq!(accounts["ordinary-service"].credentials.secret_key, "ordinary-secret");
+        let exported = serde_json::to_string(&accounts).expect("serialize selected export records");
+        assert!(!exported.contains("replicator"));
+    }
+
+    #[test]
+    fn iam_archives_preserve_the_reserved_site_replicator_identity() {
+        let failure = super::imported_service_account_reserved_failure(rustfs_iam::sys::SITE_REPLICATOR_SERVICE_ACCOUNT)
+            .expect("the site replicator must never be replaced by an IAM import");
+        assert_eq!(failure.name, "site-replicator-0");
+        assert_eq!(failure.error, "service account name is reserved for site replication");
+        assert!(super::imported_service_account_reserved_failure("ordinary-service-account").is_none());
+        let source = include_str!("user.rs")
+            .split("\n#[cfg(test)]\n")
+            .next()
+            .expect("production source");
+        let export = source_block(source, "impl Operation for ExportIam");
+        let guard = export
+            .find("retain_exportable_service_accounts(&mut service_accounts)")
+            .expect("export uses the canonical reserved identity");
+        assert!(
+            guard
+                < export
+                    .find("let svc_acc_create_req = SRSvcAccCreate")
+                    .expect("exported credentials")
+        );
+        let import = source_block(source, "impl Operation for ImportIam");
+        let guard = import
+            .find("imported_service_account_reserved_failure(&ak)")
+            .expect("import rejects the reserved identity");
+        assert!(
+            guard
+                < import
+                    .find(".delete_service_account(&ak")
+                    .expect("replacement deletes the original account")
+        );
+    }
+
     #[test]
     fn import_iam_enqueues_a_site_replication_snapshot() {
         let body = source_block(include_str!("user.rs"), "impl Operation for ImportIam");

rustfs/src/admin/route_policy.rs

@@ -18,7 +18,6 @@ use rustfs_security_governance::{
 };
 
 const ALL_ADMIN: AdminActionRef = AdminActionRef::new("AllAdminActions");
-const ADD_USER_TO_GROUP: AdminActionRef = AdminActionRef::new("AddUserToGroupAdminAction");
 const ATTACH_POLICY: AdminActionRef = AdminActionRef::new("AttachPolicyAdminAction");
 const CONFIG_UPDATE: AdminActionRef = AdminActionRef::new("ConfigUpdateAdminAction");
 const TLS_CERTIFICATE_PUBLISH: AdminActionRef = AdminActionRef::new("TLSCertificatePublishAdminAction");
@@ -187,12 +186,6 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         RouteRiskLevel::High,
     ),
     admin(HttpMethod::Put, "/rustfs/admin/v3/set-group-status", ENABLE_GROUP, RouteRiskLevel::High),
-    admin(
-        HttpMethod::Put,
-        "/rustfs/admin/v3/update-group-members",
-        ADD_USER_TO_GROUP,
-        RouteRiskLevel::High,
-    ),
     admin(
         HttpMethod::Get,
         "/rustfs/admin/v3/temporary-account-info",
@@ -1806,6 +1799,11 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
 ];
 
 pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[
+    deferred(
+        HttpMethod::Put,
+        "/rustfs/admin/v3/update-group-members",
+        DeferredRoutePolicyReason::MultipleActions,
+    ),
     deferred(HttpMethod::Get, "/rustfs/admin/v3/accountinfo", DeferredRoutePolicyReason::S3Action),
     // The self-service account routes act on the caller, never on a target
     // named in the request, so they gate on possession of the credential (plus,

rustfs/src/admin/service/federated_identity.rs

@@ -20,7 +20,7 @@ use crate::admin::{
 use rustfs_iam::{
     federation::{FederatedSessionBinding, FederatedSessionBindingError, FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM},
     store::{MappedPolicy, object::ObjectStore},
-    sys::{IamSys, is_safe_claim_policy_name},
+    sys::{IamSys, OIDC_ISSUER_CLAIM, is_safe_claim_policy_name},
 };
 use rustfs_madmin::{SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC, SRIAMItem, SRSTSCredential};
 use rustfs_policy::auth::get_new_credentials_with_metadata;
@@ -111,6 +111,9 @@ fn build_oidc_token_claims(transaction: &FederatedSessionTransaction) -> HashMap
     let mut token_claims = HashMap::new();
     token_claims.insert("sub".to_string(), Value::String(authorization.subject().to_string()));
     token_claims.insert("iss".to_string(), Value::String("rustfs-oidc".to_string()));
+    if let Some(issuer) = authorization.issuer() {
+        token_claims.insert(OIDC_ISSUER_CLAIM.to_string(), Value::String(issuer.to_string()));
+    }
     token_claims.insert("oidc_provider".to_string(), Value::String(authorization.provider_id().to_string()));
 
     if !authorization.email().is_empty() {
@@ -329,14 +332,31 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding {
         let parent_user = authorization.oidc_virtual_parent().ok_or_else(|| {
             FederatedSessionBindingError::InvalidRequest("verified OIDC identity is missing issuer or subject".to_string())
         })?;
-        let (selected_policy_names, ignorable_policy_names) = match iam_store
-            .sts_policy_db_get(&parent_user, &Some(authorization.mapped_iam_groups().to_vec()))
+        let (mapped_policy_names, has_identity_mapping) = iam_store
+            .sts_policy_db_get_with_identity_mapping(&parent_user, &Some(authorization.mapped_iam_groups().to_vec()))
             .await
-            .map_err(|_| FederatedSessionBindingError::Internal("failed to resolve OIDC policy mapping".to_string()))?
-        {
-            mapped_policy_names if !mapped_policy_names.is_empty() => (mapped_policy_names, &[][..]),
-            _ => (authorization.policies().to_vec(), authorization.group_claim_policies()),
+            .map_err(|_| FederatedSessionBindingError::Internal("failed to resolve OIDC policy mapping".to_string()))?;
+        // An explicit identity mapping overrides token policies. Group mappings
+        // are additive and must not discard claim-derived Deny policies.
+        let ignorable_policy_names: Vec<String> = if has_identity_mapping {
+            Vec::new()
+        } else {
+            authorization
+                .group_claim_policies()
+                .iter()
+                .filter(|name| !mapped_policy_names.contains(name))
+                .cloned()
+                .collect()
         };
+        let mut selected_policy_names = mapped_policy_names;
+        if !has_identity_mapping {
+            for name in authorization.policies() {
+                if !selected_policy_names.contains(name) {
+                    selected_policy_names.push(name.clone());
+                }
+            }
+        }
+
         // A name may be skipped only after storage confirms the policy is absent. A failed lookup
         // must not be mistaken for absence, or an unreadable Deny policy would be dropped from the
         // signed session and could not be restored at request time.
@@ -368,7 +388,7 @@ impl FederatedSessionBinding for DefaultFederatedSessionBinding {
         let resolved_policy_names = MappedPolicy::new(&resolved_policy_mapping).to_slice();
         let selection = select_oidc_policies(
             &selected_policy_names,
-            ignorable_policy_names,
+            &ignorable_policy_names,
             &resolved_policy_names,
             &missing_policy_names,
         )
@@ -543,6 +563,7 @@ mod tests {
             HashMap::from([
                 ("sub".to_string(), serde_json::json!("subject")),
                 ("iss".to_string(), serde_json::json!("rustfs-oidc")),
+                (OIDC_ISSUER_CLAIM.to_string(), serde_json::json!("https://idp.example.test")),
                 ("oidc_provider".to_string(), serde_json::json!("default")),
                 ("email".to_string(), serde_json::json!("user@example.com")),
                 ("preferred_username".to_string(), serde_json::json!("user")),
@@ -566,6 +587,7 @@ mod tests {
             HashMap::from([
                 ("sub".to_string(), serde_json::json!("subject")),
                 ("iss".to_string(), serde_json::json!("rustfs-oidc")),
+                (OIDC_ISSUER_CLAIM.to_string(), serde_json::json!("https://idp.example.test")),
                 ("oidc_provider".to_string(), serde_json::json!("default")),
             ])
         );
@@ -641,6 +663,7 @@ mod tests {
             HashMap::from([
                 ("sub".to_string(), serde_json::json!("subject")),
                 ("iss".to_string(), serde_json::json!("rustfs-oidc")),
+                (OIDC_ISSUER_CLAIM.to_string(), serde_json::json!("https://idp.example.test")),
                 ("oidc_provider".to_string(), serde_json::json!("default")),
                 ("email".to_string(), serde_json::json!("user@example.com")),
                 ("preferred_username".to_string(), serde_json::json!("user")),
@@ -684,6 +707,153 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn binding_group_policies_preserve_claim_denies() {
+        let iam = ready_test_iam().await;
+        let deny = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[{
+            "Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::payroll/*"
+        }]}"#,
+        )
+        .expect("parse payroll Deny");
+        iam.set_policy("claim-deny-payroll", deny).await.expect("store claim Deny");
+        iam.add_users_to_group("claim-policy-group", Vec::new())
+            .await
+            .expect("create mapped group");
+        iam.policy_db_set("claim-policy-group", UserType::Reg, true, "readwrite")
+            .await
+            .expect("attach group Allow");
+        let authorization = || {
+            let mut fixture = authorization_fixture();
+            fixture.sub = "claim-deny-subject".to_string();
+            fixture.policies = vec!["readwrite".to_string(), "claim-deny-payroll".to_string()];
+            fixture.mapped_iam_groups = vec!["claim-policy-group".to_string()];
+            fixture
+        };
+        let credentials = DefaultFederatedSessionBinding
+            .bind(&transaction_from_fixture(authorization()))
+            .await
+            .expect("bind group plus claim policies");
+        let claims = get_claims_for_test(&credentials);
+        let selected = MappedPolicy::new(claims.get("policy").and_then(Value::as_str).expect("policy claim")).to_slice();
+        assert!(selected.contains(&"claim-deny-payroll".to_string()));
+        let conditions = HashMap::new();
+        let args = Args {
+            account: &credentials.access_key,
+            groups: &credentials.groups,
+            action: Action::S3Action(S3Action::GetObjectAction),
+            bucket: "payroll",
+            object: "salary.csv",
+            conditions: &conditions,
+            claims: &claims,
+            is_owner: false,
+            deny_only: false,
+        };
+        assert!(!iam.is_allowed(&args).await, "group Allow must not discard claim Deny");
+        let mut authorization = authorization();
+        authorization.policies.push("missing-required-policy".to_string());
+        assert!(
+            DefaultFederatedSessionBinding
+                .bind(&transaction_from_fixture(authorization))
+                .await
+                .is_err(),
+            "group mappings must not hide unresolved required claim policies"
+        );
+    }
+
+    fn get_claims_for_test(credentials: &rustfs_credentials::Credentials) -> HashMap<String, Value> {
+        let signing_key = current_token_signing_key().expect("test signing key");
+        rustfs_iam::sys::get_claims_from_token_with_secret(&credentials.session_token, &signing_key)
+            .expect("verify issued session token")
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn group_conditions_follow_live_parent_membership_for_all_credentials() {
+        let iam = ready_test_iam().await;
+        let user = "live-group-condition-user";
+        let group = "live-restricted-group";
+        iam.create_user(
+            user,
+            &AddOrUpdateUserReq {
+                secret_key: "live-group-user-secret".to_string(),
+                policy: None,
+                status: AccountStatus::Enabled,
+            },
+        )
+        .await
+        .expect("create group condition user");
+        let policy = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*"},
+            {"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*",
+             "Condition":{"StringEquals":{"aws:groups":"live-restricted-group"}}}
+        ]}"#,
+        )
+        .expect("parse group condition policy");
+        for member in [true, false, true] {
+            if member {
+                iam.add_users_to_group(group, vec![user.to_string()])
+                    .await
+                    .expect("add membership");
+            } else {
+                iam.remove_users_from_group(group, vec![user.to_string()])
+                    .await
+                    .expect("remove membership");
+            }
+            for kind in ["regular", "sts", "service"] {
+                let mut credentials = rustfs_credentials::Credentials {
+                    access_key: if kind == "regular" {
+                        user
+                    } else {
+                        "derived-group-condition-key"
+                    }
+                    .to_string(),
+                    secret_key: "valid-derived-secret".to_string(),
+                    status: "on".to_string(),
+                    groups: Some(vec!["credential-group".to_string()]),
+                    ..Default::default()
+                };
+                if kind != "regular" {
+                    credentials.parent_user = user.to_string();
+                    credentials.session_token = "signed-token".to_string();
+                }
+                credentials.claims = Some(HashMap::from([("groups".to_string(), serde_json::json!(["token-group"]))]));
+                if kind == "service" {
+                    credentials
+                        .claims
+                        .as_mut()
+                        .expect("claims")
+                        .insert("sa-policy".to_string(), serde_json::json!("inherited-policy"));
+                }
+                let conditions = crate::auth::get_condition_values(&http::HeaderMap::new(), &credentials, None, None, None);
+                let groups = conditions.get("groups").expect("credential and token groups");
+                assert!(groups.contains(&"token-group".to_string()));
+                assert!(groups.contains(&"credential-group".to_string()));
+                assert_eq!(groups.contains(&group.to_string()), member, "{kind}");
+                let claims = HashMap::new();
+                assert_eq!(
+                    policy
+                        .is_allowed(&Args {
+                            account: &credentials.access_key,
+                            groups: &credentials.groups,
+                            action: Action::S3Action(S3Action::GetObjectAction),
+                            bucket: "bucket",
+                            object: "object",
+                            conditions: &conditions,
+                            claims: &claims,
+                            is_owner: false,
+                            deny_only: false,
+                        })
+                        .await,
+                    !member,
+                    "{kind}"
+                );
+            }
+        }
+    }
+
     #[tokio::test]
     #[serial]
     async fn binding_uses_sts_policy_when_regular_mapping_collides() {

rustfs/src/app/multipart_usecase.rs

@@ -69,6 +69,7 @@ use super::storage_api::multipart_usecase::sse::{
 };
 use super::storage_api::multipart_usecase::{
     StorageObjectInfo as ObjectInfo, StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader,
+    validate_object_key,
 };
 use super::trailer_adapter::trailer_source;
 use crate::app::object::{
@@ -584,6 +585,8 @@ impl DefaultMultipartUsecase {
         &self,
         mut req: S3Request<CompleteMultipartUploadInput>,
     ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
+        // Uploads created before the key was reserved must not publish it.
+        validate_object_key(&req.input.key, "CompleteMultipartUpload")?;
         reject_presigned_multipart_max_total_object_size_for_other_operation(
             &req.headers,
             req.uri.query(),
@@ -990,6 +993,7 @@ impl DefaultMultipartUsecase {
         &self,
         req: S3Request<CreateMultipartUploadInput>,
     ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
+        validate_object_key(&req.input.key, "CreateMultipartUpload")?;
         let multipart_max_total_object_size = parse_presigned_multipart_max_total_object_size(
             &req.headers,
             req.uri.query(),
@@ -2706,6 +2710,33 @@ mod tests {
         assert_eq!(s3_op_total(S3Operation::AbortMultipartUpload), before + 1);
     }
 
+    #[tokio::test]
+    async fn multipart_upload_rejects_reserved_directory_markers() {
+        for key in ["__XLDIR__", "prefix__XLDIR__/object", "object__XLDIR__"] {
+            let create = CreateMultipartUploadInput::builder()
+                .bucket("bucket".to_owned())
+                .key(key.to_owned())
+                .build()
+                .expect("valid create request DTO");
+            let error = make_usecase()
+                .execute_create_multipart_upload(build_request(create, Method::POST))
+                .await
+                .expect_err("reserved key must be rejected before accessing storage");
+            assert_eq!(error.code(), &S3ErrorCode::InvalidArgument);
+
+            let complete = CompleteMultipartUploadInput::builder()
+                .bucket("bucket".to_owned())
+                .key(key.to_owned())
+                .upload_id("upload".to_owned())
+                .build()
+                .expect("valid complete request DTO");
+            let error = Box::pin(make_usecase().execute_complete_multipart_upload(build_request(complete, Method::POST)))
+                .await
+                .expect_err("uploads created before the fix must not publish reserved keys");
+            assert_eq!(error.code(), &S3ErrorCode::InvalidArgument);
+        }
+    }
+
     #[tokio::test]
     async fn execute_create_multipart_upload_rejects_invalid_storage_class() {
         let input = CreateMultipartUploadInput::builder()

rustfs/src/app/storage_api.rs

@@ -1331,7 +1331,9 @@ pub(crate) mod multipart_usecase {
     pub(crate) use super::{
         access, bucket, compression, data_usage, error, helper, io, object_utils, options, request_context, s3_api, set_disk, sse,
     };
-    pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader};
+    pub(crate) use crate::storage::storage_api::{
+        ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader, validate_object_key,
+    };
 }
 
 pub(crate) mod select_object {

rustfs/src/auth.rs

@@ -19,7 +19,8 @@ use http::header::{IF_MATCH, IF_NONE_MATCH};
 use rustfs_credentials::Credentials;
 use rustfs_iam::error::Error as IamError;
 use rustfs_iam::sys::{
-    SESSION_POLICY_NAME, get_claims_from_token_with_secret, get_claims_from_token_with_secret_allow_missing_exp,
+    OIDC_ISSUER_CLAIM, SESSION_POLICY_NAME, get_claims_from_token_with_secret,
+    get_claims_from_token_with_secret_allow_missing_exp, is_rustfs_oidc_claims,
 };
 use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive, is_server_derived_condition_key};
 use rustfs_trusted_proxies::ClientInfo;
@@ -585,6 +586,27 @@ fn policy_secure_transport(client_info: Option<&ClientInfo>) -> bool {
         .unwrap_or(false)
 }
 
+/// Canonicalize an `s3:VersionId` condition value.
+///
+/// The object handlers accept several spellings of one version id (upper
+/// case, unhyphenated, braced, `urn:uuid:`), and the storage layer resolves
+/// all of them to the same version. Policy conditions must see the single
+/// spelling the server itself emits, otherwise a `Deny` with `StringEquals`
+/// can be bypassed by re-spelling the id. `null` in any case and the nil
+/// UUID both name the null version. Strings that are not a UUID are passed
+/// through unchanged; the handlers reject them later.
+pub(crate) fn canonical_policy_version_id(version: &str) -> String {
+    let trimmed = version.trim();
+    if trimmed.eq_ignore_ascii_case("null") {
+        return "null".to_string();
+    }
+    match uuid::Uuid::parse_str(trimmed) {
+        Ok(id) if id.is_nil() => "null".to_string(),
+        Ok(id) => id.to_string(),
+        Err(_) => version.to_string(),
+    }
+}
+
 /// Get condition values for policy evaluation
 ///
 /// # Arguments
@@ -679,9 +701,6 @@ pub fn get_condition_values_with_query_and_client_info(
     let curr_time = OffsetDateTime::now_utc();
     let epoch_time = curr_time.unix_timestamp();
 
-    // Use provided version ID or empty string
-    let vid = version_id.unwrap_or("");
-
     // Determine auth type and signature version from headers and query
     let (auth_type, signature_version) = determine_auth_type_and_version_with_query(header, query);
 
@@ -709,9 +728,10 @@ pub fn get_condition_values_with_query_and_client_info(
     args.insert("username".to_owned(), vec![username]);
     args.insert("principaltype".to_owned(), vec![principal_type.to_string()]);
 
-    // Add version ID
-    if !vid.is_empty() {
-        args.insert("versionid".to_owned(), vec![vid.to_string()]);
+    // Add version ID. Policy conditions must identify the same version the
+    // storage layer resolves, whatever spelling the client used.
+    if let Some(vid) = version_id.filter(|vid| !vid.is_empty()) {
+        args.insert("versionid".to_owned(), vec![canonical_policy_version_id(vid)]);
     }
 
     // Add signature version and auth type
@@ -782,6 +802,17 @@ pub fn get_condition_values_with_query_and_client_info(
             }
         }
 
+        if is_rustfs_oidc_claims(claims) {
+            args.remove("iss");
+            if let Some(issuer) = claims
+                .get(OIDC_ISSUER_CLAIM)
+                .and_then(Value::as_str)
+                .filter(|issuer| !issuer.is_empty())
+            {
+                args.insert("iss".to_string(), vec![issuer.to_string()]);
+            }
+        }
+
         let grps = extract_string_list_claim(claims, "groups");
         if !grps.is_empty() {
             args.insert("groups".to_string(), grps);
@@ -793,10 +824,19 @@ pub fn get_condition_values_with_query_and_client_info(
         }
     }
 
-    if let Some(groups) = &cred.groups
-        && !args.contains_key("groups")
-    {
-        args.insert("groups".to_string(), groups.clone());
+    let effective_user = if (cred.is_service_account() || cred.is_temp()) && !cred.parent_user.is_empty() {
+        &cred.parent_user
+    } else {
+        &cred.access_key
+    };
+    let live_groups = current_ready_iam_handle()
+        .map(|iam| iam.user_group_memberships(effective_user))
+        .unwrap_or_default();
+    for group in cred.groups.iter().flatten().chain(live_groups.iter()) {
+        let groups = args.entry("groups".to_string()).or_default();
+        if !groups.contains(group) {
+            groups.push(group.clone());
+        }
     }
 
     // Every remaining header is attacker-controlled. A header must never contribute
@@ -2016,6 +2056,40 @@ mod tests {
         assert_eq!(conditions.get("versionid"), Some(&vec!["server-version".to_string()]));
     }
 
+    #[test]
+    fn version_id_condition_uses_canonical_spelling() {
+        let cred = create_test_credentials();
+        let headers = HeaderMap::new();
+        let canonical = "68f21f07-9bae-4b3a-a120-36f48fb9de1a";
+        for spelling in [
+            canonical,
+            "68F21F07-9BAE-4B3A-A120-36F48FB9DE1A",
+            "68f21f079bae4b3aa12036f48fb9de1a",
+            "{68f21f07-9bae-4b3a-a120-36f48fb9de1a}",
+            "urn:uuid:68f21f07-9bae-4b3a-a120-36f48fb9de1a",
+            " 68f21f07-9bae-4b3a-a120-36f48fb9de1a ",
+        ] {
+            let conditions = get_condition_values(&headers, &cred, Some(spelling), None, None);
+            assert_eq!(conditions.get("versionid"), Some(&vec![canonical.to_string()]), "{spelling}");
+        }
+        for spelling in [
+            "null",
+            "NULL",
+            " null ",
+            "00000000-0000-0000-0000-000000000000",
+            "00000000000000000000000000000000",
+        ] {
+            let conditions = get_condition_values(&headers, &cred, Some(spelling), None, None);
+            assert_eq!(conditions.get("versionid"), Some(&vec!["null".to_string()]), "{spelling}");
+        }
+        // Strings the handlers reject are passed through unchanged.
+        let conditions = get_condition_values(&headers, &cred, Some("not-a-uuid"), None, None);
+        assert_eq!(conditions.get("versionid"), Some(&vec!["not-a-uuid".to_string()]));
+        for absent in [None, Some("")] {
+            assert!(!get_condition_values(&headers, &cred, absent, None, None).contains_key("versionid"));
+        }
+    }
+
     #[test]
     fn ghsa_6r96_claim_condition_keys_ignore_spoofed_headers() {
         // The credential carries no groups/roles claims, so these keys are absent --
@@ -2252,6 +2326,56 @@ mod tests {
         // (we can't directly test this without changing the function signature)
     }
 
+    #[tokio::test]
+    async fn oidc_issuer_condition_uses_verified_provider_not_session_marker_or_header() {
+        use rustfs_policy::policy::{
+            Args, Policy,
+            action::{Action, S3Action},
+        };
+
+        let policy = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[
+                {"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*"},
+                {"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::bucket/*",
+                 "Condition":{"StringEquals":{"jwt:iss":"https://partner.example"}}}
+            ]}"#,
+        )
+        .expect("parse issuer condition");
+        let mut credentials = create_test_credentials();
+        let mut headers = HeaderMap::new();
+        headers.insert("iss", HeaderValue::from_static("https://spoofed.invalid"));
+        headers.insert(OIDC_ISSUER_CLAIM, HeaderValue::from_static("https://spoofed.invalid"));
+        for issuer in [Some("https://partner.example"), Some("https://corporate.example"), None] {
+            let mut claims = HashMap::from([
+                ("iss".to_string(), json!("rustfs-oidc")),
+                ("oidc_provider".to_string(), json!("provider")),
+                ("sub".to_string(), json!("subject")),
+            ]);
+            if let Some(issuer) = issuer {
+                claims.insert(OIDC_ISSUER_CLAIM.to_string(), json!(issuer));
+            }
+            credentials.claims = Some(claims);
+            let conditions = get_condition_values(&headers, &credentials, None, None, None);
+            assert_eq!(conditions.get("iss"), issuer.map(|value| vec![value.to_string()]).as_ref());
+            assert_eq!(
+                policy
+                    .is_allowed(&Args {
+                        account: &credentials.access_key,
+                        groups: &credentials.groups,
+                        action: Action::S3Action(S3Action::GetObjectAction),
+                        bucket: "bucket",
+                        object: "object",
+                        conditions: &conditions,
+                        claims: credentials.claims_or_empty(),
+                        is_owner: false,
+                        deny_only: false,
+                    })
+                    .await,
+                issuer != Some("https://partner.example"),
+            );
+        }
+    }
+
     #[test]
     fn test_get_condition_values_with_claims() {
         let mut cred = create_service_account_credentials();

rustfs/src/storage/access.rs

@@ -155,6 +155,19 @@ struct RequestedObjectLockRetainUntil(OffsetDateTime);
 #[derive(Clone, Debug)]
 struct RequestedObjectLockRetention(ObjectLockRetention);
 
+/// Legal hold status a standalone PutObjectLegalHold applies from its XML body.
+#[derive(Clone, Debug)]
+struct RequestedObjectLegalHold(ObjectLockLegalHold);
+
+/// Tag set a standalone PutObjectTagging applies from its XML body.
+#[derive(Clone, Debug)]
+struct RequestedObjectTags(Tagging);
+
+/// Request headers with the object policy inputs a POST upload applies from its
+/// form fields, so header-derived condition keys describe what is stored.
+#[derive(Clone, Debug)]
+struct PostObjectPolicyHeaders(http::HeaderMap);
+
 /// These operations stage upload state without publishing an object. Keep this
 /// separate from the marker controlling multipart publication fences.
 #[derive(Clone, Copy, Debug)]
@@ -925,8 +938,14 @@ fn authorization_conditions<T>(
     client_info: Option<&ClientInfo>,
     action: Action,
 ) -> S3Result<HashMap<String, Vec<String>>> {
+    // A POST upload applies its form fields, not its headers, so the policy
+    // decision must read the values the handler stores.
+    let headers = req
+        .extensions
+        .get::<PostObjectPolicyHeaders>()
+        .map_or(&req.headers, |headers| &headers.0);
     let mut conditions = get_condition_values_with_query_and_client_info(
-        &req.headers,
+        headers,
         cred,
         version_id,
         None,
@@ -951,7 +970,31 @@ fn authorization_conditions<T>(
         conditions.insert("LocationConstraint".to_string(), vec![location]);
     }
     merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions);
-    merge_request_object_tag_conditions(action, &req.headers, &mut conditions)?;
+    merge_request_object_tag_conditions(action, headers, &mut conditions)?;
+    if let Some(RequestedObjectLegalHold(legal_hold)) = req.extensions.get::<RequestedObjectLegalHold>() {
+        // Standalone legal hold writes apply the XML body, so headers must not
+        // supply a different value to the policy decision.
+        conditions.remove("object-lock-legal-hold");
+        if let Some(status) = &legal_hold.status {
+            conditions.insert("object-lock-legal-hold".to_string(), vec![status.as_str().to_string()]);
+        }
+    }
+    if let Some(RequestedObjectTags(tagging)) = req.extensions.get::<RequestedObjectTags>() {
+        // Standalone tag writes apply the XML body; the x-amz-tagging header is
+        // not stored and must not describe the request to the policy engine.
+        conditions.retain(|key, _| key != "RequestObjectTagKeys" && !key.starts_with("RequestObjectTag/"));
+        let mut keys = Vec::new();
+        for tag in &tagging.tag_set {
+            if let (Some(key), Some(value)) = (&tag.key, &tag.value) {
+                keys.push(key.clone());
+                conditions
+                    .entry(format!("RequestObjectTag/{key}"))
+                    .or_default()
+                    .push(value.clone());
+            }
+        }
+        conditions.insert("RequestObjectTagKeys".to_string(), keys);
+    }
     if let Some(RequestedObjectLockRetention(retention)) = req.extensions.get::<RequestedObjectLockRetention>() {
         // Standalone retention writes apply the XML body, so headers must not
         // supply a different value to the policy decision.
@@ -1007,6 +1050,58 @@ fn set_requested_object_lock_retention(req: &mut S3Request<PutObjectRetentionInp
     req.extensions.insert(RequestedObjectLockRetention(retention));
 }
 
+/// A POST upload stores the object policy inputs from its form fields, while
+/// header-derived condition keys would otherwise read the HTTP headers. Project
+/// the form fields over the headers for the policy decision, removing a header
+/// whose field is absent so a client header cannot stand in for a stored value.
+fn set_post_object_policy_headers(req: &mut S3Request<PutObjectInput>) -> S3Result<()> {
+    if req.method != http::Method::POST {
+        return Ok(());
+    }
+    let input = &req.input;
+    let retain_until = input
+        .object_lock_retain_until_date
+        .as_ref()
+        .map(|date| OffsetDateTime::from(date.clone()).format(&time::format_description::well_known::Rfc3339))
+        .transpose()
+        .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid retain-until date"))?;
+    let mut headers = req.headers.clone();
+    for (key, value) in [
+        ("x-amz-object-lock-mode", input.object_lock_mode.as_ref().map(ObjectLockMode::as_str)),
+        (
+            "x-amz-object-lock-legal-hold",
+            input
+                .object_lock_legal_hold_status
+                .as_ref()
+                .map(ObjectLockLegalHoldStatus::as_str),
+        ),
+        ("x-amz-object-lock-retain-until-date", retain_until.as_deref()),
+        ("x-amz-tagging", input.tagging.as_deref()),
+        ("x-amz-acl", input.acl.as_ref().map(ObjectCannedACL::as_str)),
+        ("x-amz-storage-class", input.storage_class.as_ref().map(StorageClass::as_str)),
+        (
+            "x-amz-server-side-encryption",
+            input.server_side_encryption.as_ref().map(ServerSideEncryption::as_str),
+        ),
+        ("x-amz-server-side-encryption-aws-kms-key-id", input.ssekms_key_id.as_deref()),
+        ("x-amz-server-side-encryption-customer-algorithm", input.sse_customer_algorithm.as_deref()),
+        ("x-amz-grant-full-control", input.grant_full_control.as_deref()),
+        ("x-amz-grant-read", input.grant_read.as_deref()),
+        ("x-amz-grant-read-acp", input.grant_read_acp.as_deref()),
+        ("x-amz-grant-write-acp", input.grant_write_acp.as_deref()),
+    ] {
+        headers.remove(key);
+        if let Some(value) = value {
+            let value = value
+                .parse()
+                .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid form field"))?;
+            headers.insert(key, value);
+        }
+    }
+    req.extensions.insert(PostObjectPolicyHeaders(headers));
+    Ok(())
+}
+
 fn merge_object_lock_remaining_retention_days_condition(
     requested: Option<&RequestedObjectLockRetainUntil>,
     now: OffsetDateTime,
@@ -3465,6 +3560,7 @@ impl S3Access for FS {
             };
             let requested_retain_until = req.input.object_lock_retain_until_date.clone();
             set_requested_object_lock_retain_until(req, requested_retain_until.as_ref());
+            set_post_object_policy_headers(req)?;
             authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?;
             req.extensions.insert(bucket_generation?);
 
@@ -3502,6 +3598,8 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_object_legal_hold(&self, req: &mut S3Request<PutObjectLegalHoldInput>) -> S3Result<()> {
+        req.extensions
+            .insert(RequestedObjectLegalHold(req.input.legal_hold.clone().unwrap_or_default()));
         let req_info = ext_req_info_mut(&mut req.extensions)?;
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
@@ -3549,6 +3647,7 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_object_tagging(&self, req: &mut S3Request<PutObjectTaggingInput>) -> S3Result<()> {
+        req.extensions.insert(RequestedObjectTags(req.input.tagging.clone()));
         let req_info = ext_req_info_mut(&mut req.extensions)?;
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
@@ -3681,8 +3780,8 @@ mod tests {
     };
     use super::{BulkDenialAudit, audit_request_denial};
     use super::{
-        remaining_retention_days, requested_bucket_location_constraint, set_requested_object_lock_retain_until,
-        set_requested_object_lock_retention,
+        RequestedObjectLegalHold, RequestedObjectTags, remaining_retention_days, requested_bucket_location_constraint,
+        set_post_object_policy_headers, set_requested_object_lock_retain_until, set_requested_object_lock_retention,
     };
     use crate::auth::OBJECT_LOCK_REMAINING_RETENTION_DAYS_CONDITION;
     use crate::error::ApiError;
@@ -4812,6 +4911,108 @@ mod tests {
         authorization_conditions(req, &credentials, None, None, None, None, action).expect("conditions should build")
     }
 
+    #[test]
+    fn legal_hold_policy_checks_body_instead_of_conflicting_headers() {
+        let mut req = build_request(
+            PutObjectLegalHoldInput {
+                legal_hold: Some(ObjectLockLegalHold {
+                    status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::OFF)),
+                }),
+                ..Default::default()
+            },
+            Method::PUT,
+        );
+        req.headers
+            .insert("x-amz-object-lock-legal-hold", HeaderValue::from_static("ON"));
+        let action = Action::S3Action(S3Action::PutObjectLegalHoldAction);
+        // Without the hook's extension the header leaks into the decision.
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["ON".to_string()]));
+        req.extensions
+            .insert(RequestedObjectLegalHold(req.input.legal_hold.clone().unwrap_or_default()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["OFF".to_string()]));
+        // A body without a status stores no hold change; the header must not fill it in.
+        req.extensions
+            .insert(RequestedObjectLegalHold(ObjectLockLegalHold::default()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert!(!conditions.contains_key("object-lock-legal-hold"));
+    }
+
+    #[test]
+    fn tagging_policy_checks_body_instead_of_conflicting_headers() {
+        let mut req = build_request(
+            PutObjectTaggingInput::builder()
+                .bucket("bucket".to_string())
+                .key("object".to_string())
+                .tagging(Tagging {
+                    tag_set: vec![Tag {
+                        key: Some("classification".to_string()),
+                        value: Some("confidential".to_string()),
+                    }],
+                })
+                .build()
+                .expect("valid tagging request DTO"),
+            Method::PUT,
+        );
+        req.headers
+            .insert("x-amz-tagging", HeaderValue::from_static("classification=public"));
+        let action = Action::S3Action(S3Action::PutObjectTaggingAction);
+        req.extensions.insert(RequestedObjectTags(req.input.tagging.clone()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("RequestObjectTag/classification"), Some(&vec!["confidential".to_string()]));
+        assert_eq!(conditions.get("RequestObjectTagKeys"), Some(&vec!["classification".to_string()]));
+        req.extensions.insert(RequestedObjectTags(Tagging::default()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert!(!conditions.contains_key("RequestObjectTag/classification"));
+        assert_eq!(conditions.get("RequestObjectTagKeys"), Some(&Vec::new()));
+    }
+
+    #[test]
+    fn post_object_policy_checks_form_fields_instead_of_headers() {
+        let mut req = build_request(
+            PutObjectInput {
+                bucket: "bucket".to_string(),
+                key: "object".to_string(),
+                object_lock_legal_hold_status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::ON)),
+                object_lock_mode: Some(ObjectLockMode::from_static(ObjectLockMode::COMPLIANCE)),
+                tagging: Some("classification=confidential".to_string()),
+                ..Default::default()
+            },
+            Method::POST,
+        );
+        req.headers
+            .insert("x-amz-object-lock-legal-hold", HeaderValue::from_static("OFF"));
+        req.headers
+            .insert("x-amz-object-lock-mode", HeaderValue::from_static("GOVERNANCE"));
+        req.headers
+            .insert("x-amz-object-lock-retain-until-date", HeaderValue::from_static("2029-01-01T00:00:00Z"));
+        req.headers
+            .insert("x-amz-tagging", HeaderValue::from_static("classification=public"));
+        set_post_object_policy_headers(&mut req).expect("form fields project onto policy headers");
+        let action = Action::S3Action(S3Action::PutObjectAction);
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["ON".to_string()]));
+        assert_eq!(conditions.get("object-lock-mode"), Some(&vec!["COMPLIANCE".to_string()]));
+        // The form sets no retain-until date, so the header must not stand in for one.
+        assert!(!conditions.contains_key("object-lock-retain-until-date"));
+        assert_eq!(conditions.get("RequestObjectTag/classification"), Some(&vec!["confidential".to_string()]));
+        // A PUT keeps reading its headers.
+        let mut put = build_request(
+            PutObjectInput {
+                bucket: "bucket".to_string(),
+                key: "object".to_string(),
+                ..Default::default()
+            },
+            Method::PUT,
+        );
+        put.headers
+            .insert("x-amz-object-lock-legal-hold", HeaderValue::from_static("ON"));
+        set_post_object_policy_headers(&mut put).expect("PUT is untouched");
+        let conditions = remaining_retention_days_conditions(&put, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["ON".to_string()]));
+    }
+
     #[tokio::test]
     async fn retention_policy_checks_body_instead_of_conflicting_headers() {
         use rustfs_policy::policy::{Args, Policy};

rustfs/src/storage/ecfs_extend.rs

@@ -339,6 +339,19 @@ pub(crate) fn validate_object_key(key: &str, operation: &str) -> S3Result<()> {
         ));
     }
 
+    // Storage encodes directory objects as `<name>__XLDIR__` after this client-key
+    // boundary, and `scan_dir` decodes every entry with that suffix back into a
+    // marker. A client key carrying the suffix would be mistaken for a marker:
+    // the bare key `__XLDIR__` makes the walk schedule the directory it is
+    // scanning as its own child until the stack overflows, and keys below a
+    // `*__XLDIR__/` segment vanish from listings and quota accounting.
+    if key.contains(rustfs_utils::path::GLOBAL_DIR_SUFFIX) {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidArgument,
+            "Object key contains a reserved directory marker",
+        ));
+    }
+
     if key.is_empty() || !crate::storage::storage_api::ecstore_bucket::utils::is_valid_object_prefix(key) {
         return Err(S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid object key"));
     }
@@ -1139,3 +1152,26 @@ pub(crate) fn parse_part_number_i32_to_usize(part_number: Option<i32>, op: &'sta
         Some(n) => Ok(Some(n as usize)),
     }
 }
+
+#[cfg(test)]
+mod object_key_admission_tests {
+    use super::validate_object_key;
+
+    #[test]
+    fn reserved_directory_marker_keys_are_rejected() {
+        for key in [
+            "__XLDIR__",
+            "__XLDIR__/",
+            "prefix__XLDIR__/object",
+            "object__XLDIR__",
+            "a/__XLDIR__/b",
+        ] {
+            for operation in ["PutObject", "COPY (dest)", "CreateMultipartUpload", "CompleteMultipartUpload"] {
+                assert!(validate_object_key(key, operation).is_err(), "{operation}: {key}");
+            }
+        }
+        for key in ["ordinary/key", "xldir", "__xldir__", "directory/", "unicode-中文"] {
+            assert!(validate_object_key(key, "PutObject").is_ok(), "{key}");
+        }
+    }
+}