probe: deliberate mutations, DO NOT MERGE

Add IamPolicySource, the gateway PolicySource answered by RustFS IAM
(rustfs/backlog#2760, policy half). One snapshot per request carries a
RustfsPrincipal: principal kind, account, effective username, parent
user, credential groups, live group memberships, token claims and
is_owner.

- Anonymous callers get an empty snapshot without asking IAM.
- IAM not ready, not initialized or failing is PolicyError::unavailable,
  never an empty snapshot.
- Unknown, disabled and session-kind-mismatched principals are refused
  with constant labels; unreadable tokens are malformed.
- Claims are read from the stored token (root secret for STS, the
  account's own secret for service accounts), and is_owner mirrors the
  legacy has_root_access rule.
- RustfsPrincipal holds no secret or session token and has no Debug.

.config/test-count-baseline.txt

@@ -25,6 +25,7 @@ rustfs-data-usage	58	0
 rustfs-ecstore	7858	2
 rustfs-extension-schema	23	0
 rustfs-filemeta	361	0
+rustfs-gateway-adapter	22	0
 rustfs-heal	1128	3
 rustfs-heal-contracts	8	0
 rustfs-iam	415	0

Cargo.lock

@@ -4698,6 +4698,12 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd"
 
+[[package]]
+name = "futures-timer"
+version = "3.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "af43fadb8a98512d547e37b4e92e0ced13e205c061b87b4623eff01d918d6968"
+
 [[package]]
 name = "futures-util"
 version = "0.3.34"
@@ -7914,10 +7920,34 @@ version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "010378780309880b08997fae13be7834dba947d36393bd372f2b1556deb2a2f6"
 dependencies = [
+ "phf_macros",
  "phf_shared 0.14.0",
  "serde",
 ]
 
+[[package]]
+name = "phf_generator"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aeb62e0959d5a1bebc965f4d15d9e2b7cea002b6b0f5ba8cde6cc26738467100"
+dependencies = [
+ "fastrand",
+ "phf_shared 0.14.0",
+]
+
+[[package]]
+name = "phf_macros"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5fa8d0ca26d424d27630da600c6624696e7dec8bf7b3b492b383c5dc49e5e085"
+dependencies = [
+ "phf_generator",
+ "phf_shared 0.14.0",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "phf_shared"
 version = "0.12.1"
@@ -9999,6 +10029,114 @@ dependencies = [
  "xxhash-rust",
 ]
 
+[[package]]
+name = "rustfs-gateway"
+version = "0.57.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "arc-swap",
+ "bytes",
+ "form_urlencoded",
+ "futures-timer",
+ "futures-util",
+ "http 1.5.0",
+ "http-body",
+ "http-body-util",
+ "httparse",
+ "hyper",
+ "pin-project-lite",
+ "rustfs-gateway-core",
+ "rustfs-gateway-http",
+ "rustfs-gateway-macros",
+ "rustfs-gateway-sig",
+ "rustfs-gateway-stream",
+ "rustfs-gateway-types",
+ "rustfs-gateway-xml",
+ "sha2 0.11.1",
+ "tokio",
+ "tower",
+ "tracing",
+ "url",
+]
+
+[[package]]
+name = "rustfs-gateway-adapter"
+version = "1.0.5"
+dependencies = [
+ "http 1.5.0",
+ "rustfs-credentials",
+ "rustfs-gateway",
+ "rustfs-gateway-http",
+ "rustfs-iam",
+ "rustfs-policy",
+ "serde_json",
+ "time",
+ "tokio",
+]
+
+[[package]]
+name = "rustfs-gateway-core"
+version = "0.54.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "bytes",
+ "http 1.5.0",
+ "md-5 0.11.0",
+ "percent-encoding",
+ "rustfs-gateway-http",
+ "rustfs-gateway-sig",
+ "rustfs-gateway-stream",
+ "rustfs-gateway-types",
+ "rustfs-gateway-xml",
+ "serde",
+ "serde_json",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "rustfs-gateway-http"
+version = "0.1.4"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "hmac 0.13.0",
+ "http 1.5.0",
+ "httparse",
+ "memchr",
+ "phf 0.14.0",
+ "rustfs-gateway-stream",
+ "rustfs-gateway-types",
+ "sha2 0.11.1",
+ "smallvec",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "rustfs-gateway-macros"
+version = "0.1.4"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.9",
+]
+
+[[package]]
+name = "rustfs-gateway-sig"
+version = "0.13.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "hmac 0.13.0",
+ "http 1.5.0",
+ "rustfs-gateway-http",
+ "sha1 0.11.0",
+ "sha2 0.11.1",
+ "smallvec",
+ "subtle",
+ "zeroize",
+]
+
 [[package]]
 name = "rustfs-gateway-stream"
 version = "0.2.0"
@@ -12603,10 +12741,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
  "once_cell",
  "rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]

Cargo.toml

@@ -26,6 +26,7 @@ members = [
     "crates/ecstore", # Erasure coding storage implementation
     "crates/e2e_test", # End-to-end test suite
     "crates/filemeta", # File metadata management
+    "crates/gateway-adapter", # RustFS implementations of the gateway extension traits
     "crates/heal", # Erasure set and object healing
     "crates/heal-contracts", # Heal request/response channel contracts
     "crates/iam", # Identity and Access Management
@@ -93,6 +94,7 @@ redundant_clone = "warn"
 [workspace.dependencies]
 # RustFS Internal Crates
 rustfs = { path = "./rustfs", version = "1.0.5" }
+rustfs-gateway-adapter = { path = "crates/gateway-adapter", version = "1.0.5" }
 rustfs-heal = { path = "crates/heal", version = "1.0.5" }
 rustfs-heal-contracts = { path = "crates/heal-contracts", version = "1.0.5" }
 rustfs-scanner-metrics = { path = "crates/scanner-metrics", version = "1.0.5" }
@@ -149,6 +151,12 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.5" }
 # the s3s-facing edge of the rustfs binary.
 rustfs-gateway-types = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", version = "0.32.0", default-features = false }
 rustfs-gateway-stream = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", version = "0.2.0", default-features = false }
+# The gateway framework itself, consumed for its extension traits by the ring-2
+# adapter. `default-features = false` keeps the `server` feature, and with it the
+# server crate, out of the build.
+rustfs-gateway = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", version = "0.57.0", default-features = false }
+# Test-only: builds the `EffectiveHost` a host-resolution test queries with.
+rustfs-gateway-http = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", default-features = false }
 
 # Async Runtime and Networking
 async-channel = "2.5.0"

crates/gateway-adapter/Cargo.toml

@@ -0,0 +1,48 @@
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[package]
+name = "rustfs-gateway-adapter"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+homepage.workspace = true
+description = "RustFS implementations of the rustfs-gateway extension traits (credentials and policy facts over IAM)."
+keywords = ["rustfs", "gateway", "s3", "iam"]
+categories = ["web-programming", "authentication"]
+authors.workspace = true
+publish = false
+
+[package.metadata.gateway]
+# Documentary only: this crate is a ring-2 consumer of the gateway (it depends on
+# RustFS business crates). The RustFS repository does not run the gateway's ring guard.
+ring = 2
+
+[lints]
+workspace = true
+
+[dependencies]
+http = { workspace = true }
+rustfs-gateway = { workspace = true }
+rustfs-credentials = { workspace = true }
+rustfs-iam = { workspace = true }
+rustfs-policy = { workspace = true }
+serde_json = { workspace = true }
+
+[dev-dependencies]
+rustfs-gateway-http = { workspace = true }
+time = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt"] }

crates/gateway-adapter/src/bucket_owner.rs

@@ -0,0 +1,158 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The gateway's [`BucketOwnerSource`], answered by RustFS bucket metadata.
+//!
+//! Responsible for: asking a [`BucketOwnerLookup`] for the owner of one bucket and turning every
+//! outcome that is not a usable owner into the gateway's one refusal, [`BucketOwnerError`].
+//! NOT responsible for: where the owner is stored (the lookup's; RustFS has no dedicated owner
+//! field, so the binary's lookup reads it from the bucket's metadata), comparing it with
+//! `x-amz-expected-bucket-owner` (the gateway does, byte for byte), or authorizing anything.
+//! Upstream: the gateway's expected-owner check, which consults this source only when a request
+//! carries the header. Downstream: the lookup the binary supplies over `bucket::metadata_sys`.
+//!
+//! # Failure classes
+//!
+//! The lookup separates [`OwnerLookupError::NoSuchBucket`] from [`OwnerLookupError::Backend`], and
+//! this source maps **both** to [`BucketOwnerError::unavailable`], which the gateway answers
+//! `403 AccessDenied` and never as a matching owner. The pinned gateway's [`BucketOwnerError`] is
+//! a unit type with that single constructor, so it can express neither "no such bucket" nor a 5xx;
+//! the distinction is kept in the lookup trait so it can be surfaced once the gateway grows it.
+//! An owner that is empty is no owner and is refused the same way.
+
+use std::sync::Arc;
+
+use rustfs_gateway::{BoxFuture, BucketName, BucketOwnerError, BucketOwnerSource};
+
+/// Why a [`BucketOwnerLookup`] produced no owner.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum OwnerLookupError {
+    /// The bucket does not exist.
+    NoSuchBucket,
+    /// The metadata backend failed, or could not be reached, so the bucket's state is unknown.
+    Backend,
+}
+
+/// The narrow slice of RustFS bucket metadata an owner lookup needs.
+///
+/// Exists so the source can be tested against a fake. The binary implements it over
+/// the ecstore bucket metadata system, mapping `BucketNotFound` to
+/// [`OwnerLookupError::NoSuchBucket`] and every other error to [`OwnerLookupError::Backend`].
+pub trait BucketOwnerLookup: Send + Sync + 'static {
+    /// The owner account id recorded for `bucket`.
+    fn owner_of<'a>(&'a self, bucket: &'a str) -> BoxFuture<'a, Result<String, OwnerLookupError>>;
+}
+
+/// A [`BucketOwnerSource`] over a [`BucketOwnerLookup`].
+pub struct RustfsBucketOwners<L> {
+    lookup: L,
+}
+
+impl<L: BucketOwnerLookup> RustfsBucketOwners<L> {
+    /// A source answering from `lookup`.
+    #[must_use]
+    pub fn new(lookup: L) -> Self {
+        Self { lookup }
+    }
+}
+
+impl<L: BucketOwnerLookup> BucketOwnerSource for RustfsBucketOwners<L> {
+    fn owner<'a>(&'a self, bucket: &'a BucketName) -> BoxFuture<'a, Result<Arc<str>, BucketOwnerError>> {
+        Box::pin(async move {
+            match self.lookup.owner_of(bucket.as_str()).await {
+                Ok(owner) if !owner.is_empty() => Ok(Arc::from(owner)),
+                Ok(_) | Err(OwnerLookupError::NoSuchBucket | OwnerLookupError::Backend) => Err(BucketOwnerError::unavailable()),
+            }
+        })
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use std::sync::Mutex;
+
+    const OWNER: &str = "rustfsadmin";
+
+    struct Fake {
+        answer: Result<String, OwnerLookupError>,
+        asked: Mutex<Vec<String>>,
+    }
+
+    impl Fake {
+        fn answering(answer: Result<String, OwnerLookupError>) -> Self {
+            Self {
+                answer,
+                asked: Mutex::new(Vec::new()),
+            }
+        }
+    }
+
+    impl BucketOwnerLookup for Fake {
+        fn owner_of<'a>(&'a self, bucket: &'a str) -> BoxFuture<'a, Result<String, OwnerLookupError>> {
+            self.asked.lock().unwrap().push(bucket.to_owned());
+            let answer = self.answer.clone();
+            Box::pin(async move { answer })
+        }
+    }
+
+    async fn owner_of(answer: Result<String, OwnerLookupError>) -> Result<Arc<str>, BucketOwnerError> {
+        let source = RustfsBucketOwners::new(Fake::answering(answer));
+        source.owner(&BucketName::new("photos").expect("a bucket name")).await
+    }
+
+    /// Positive: an owner the lookup finds is returned unchanged.
+    #[tokio::test]
+    async fn the_owner_is_returned() {
+        assert_eq!(owner_of(Ok(OWNER.to_owned())).await.as_deref(), Ok(OWNER));
+    }
+
+    /// Positive: the lookup is asked about exactly the bucket the gateway named.
+    #[tokio::test]
+    async fn the_lookup_is_asked_about_the_named_bucket() {
+        let source = RustfsBucketOwners::new(Fake::answering(Ok(OWNER.to_owned())));
+        source
+            .owner(&BucketName::new("photos").expect("a bucket name"))
+            .await
+            .expect("an owner");
+        assert_eq!(*source.lookup.asked.lock().unwrap(), vec!["photos".to_owned()]);
+    }
+
+    /// Negative: a missing bucket yields no owner, so it can never match an assertion.
+    #[tokio::test]
+    async fn a_missing_bucket_is_refused() {
+        assert_eq!(owner_of(Err(OwnerLookupError::NoSuchBucket)).await, Err(BucketOwnerError::unavailable()));
+    }
+
+    /// Negative: a failing backend is a refusal, never "owner unknown, allow".
+    #[tokio::test]
+    async fn a_backend_failure_is_refused() {
+        assert_eq!(owner_of(Err(OwnerLookupError::Backend)).await, Err(BucketOwnerError::unavailable()));
+    }
+
+    /// Negative: an empty owner is no owner; it must not equal an empty assertion.
+    #[tokio::test]
+    async fn an_empty_owner_is_refused() {
+        assert_eq!(owner_of(Ok(String::new())).await, Err(BucketOwnerError::unavailable()));
+    }
+
+    /// Negative: the refusal carries nothing the lookup said, so no formatter can leak it.
+    #[tokio::test]
+    async fn the_refusal_names_nothing() {
+        let error = owner_of(Err(OwnerLookupError::Backend)).await.expect_err("a refusal");
+        assert_eq!(error.to_string(), "bucket owner unavailable");
+        assert!(!format!("{error:?}").contains("photos"));
+    }
+}

crates/gateway-adapter/src/credentials.rs

@@ -0,0 +1,379 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The gateway's [`CredentialProvider`], answered by RustFS IAM.
+//!
+//! Responsible for: resolving one access key id to the secret the gateway verifies a signature
+//! with, and classifying every other outcome as either a business result
+//! ([`CredentialLookup::NotFound`], eligible for the gateway's negative cache) or an operational
+//! fault ([`ProviderError`], counted and never cached).
+//! NOT responsible for: authorization of any kind, session-token claim validation, or deciding
+//! what the client is answered with. The gateway answers every unusable credential identically.
+//!
+//! # Secret handling
+//!
+//! Secrets travel only inside the gateway's [`Credentials`], whose `Debug` redacts them. No type
+//! in this module derives or implements `Debug`, and no error produced here carries text: the
+//! IAM error is classified and dropped.
+
+use std::sync::Arc;
+
+use rustfs_gateway::{
+    BoxFuture, CredentialLookup, CredentialProvider, Credentials, CredentialsError, ProviderError, SessionBinding,
+};
+use rustfs_iam::error::Error as IamError;
+use rustfs_iam::store::Store;
+use rustfs_iam::sys::IamSys;
+use rustfs_policy::auth::{ACCOUNT_OFF, UserIdentity};
+
+/// The opaque issuer recorded in the [`rustfs_gateway::SessionBinding`] of every RustFS STS
+/// credential.
+pub const STS_SESSION_ISSUER: &str = "rustfs-sts";
+
+/// Access keys with this prefix belong to Keystone token authentication, which never verifies an
+/// AWS signature.
+const KEYSTONE_ACCESS_KEY_PREFIX: &str = "keystone:";
+
+/// The narrow slice of IAM that a credential lookup needs.
+///
+/// Exists so the adapter can be tested against a fake: IAM exports no in-memory backend.
+/// Implemented for [`IamSys`], whose `check_key` loads a user from storage on a cache miss.
+pub trait IamKeyLookup: Send + Sync + 'static {
+    /// Whether the IAM store has finished loading.
+    fn is_ready(&self) -> bool;
+
+    /// Resolves an access key, with [`IamSys::check_key`]'s contract: `Ok((None, _))` for an
+    /// unknown key, `Ok((Some(identity), valid))` otherwise, where `valid` is false for a
+    /// disabled, expired, or malformed credential.
+    fn check_key<'a>(&'a self, access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>>;
+}
+
+impl<T: Store> IamKeyLookup for IamSys<T> {
+    fn is_ready(&self) -> bool {
+        IamSys::is_ready(self)
+    }
+
+    fn check_key<'a>(&'a self, access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>> {
+        Box::pin(IamSys::check_key(self, access_key))
+    }
+}
+
+/// A [`CredentialProvider`] over RustFS IAM, with the root credentials injected at assembly.
+pub struct IamCredentials {
+    iam: Arc<dyn IamKeyLookup>,
+    root: Option<Credentials>,
+}
+
+impl IamCredentials {
+    /// A provider that resolves every access key through `iam`.
+    pub fn new(iam: Arc<dyn IamKeyLookup>) -> Self {
+        Self { iam, root: None }
+    }
+
+    /// Adds the root credentials, which resolve even before IAM is ready.
+    ///
+    /// # Errors
+    ///
+    /// [`CredentialsError::Malformed`] when the access key id is not one the gateway can hold.
+    pub fn with_root(mut self, access_key: &str, secret_key: &[u8]) -> Result<Self, CredentialsError> {
+        self.root = Some(Credentials::new(access_key, secret_key)?);
+        Ok(self)
+    }
+
+    async fn resolve(&self, access_key: &str) -> Result<CredentialLookup, ProviderError> {
+        // Deliberate change from the legacy s3s path (`rustfs/src/auth.rs` `get_secret_key`),
+        // which answered a `keystone:` key with an empty secret. Keystone authenticates by token,
+        // so there is no AWS secret to verify a signature against: the key is unknown here.
+        if access_key.starts_with(KEYSTONE_ACCESS_KEY_PREFIX) {
+            return Ok(CredentialLookup::NotFound);
+        }
+
+        // Byte-exact, as the trait requires: the access key id is signed material.
+        if let Some(root) = &self.root
+            && root.identity().access_key_id() == access_key
+        {
+            return Ok(CredentialLookup::Found(root.clone_credentials()));
+        }
+
+        if !self.iam.is_ready() {
+            return Err(ProviderError::Unavailable);
+        }
+
+        match self.iam.check_key(access_key).await {
+            Ok((Some(identity), valid)) => to_gateway_credentials(access_key, &identity, valid).map(CredentialLookup::Found),
+            Ok((None, _)) => Ok(CredentialLookup::NotFound),
+            Err(error) => classify_iam_error(&error),
+        }
+    }
+}
+
+/// Classifies an IAM failure without keeping any of its text.
+fn classify_iam_error(error: &IamError) -> Result<CredentialLookup, ProviderError> {
+    match error {
+        IamError::NoSuchUser(_)
+        | IamError::NoSuchServiceAccount(_)
+        | IamError::NoSuchTempAccount(_)
+        | IamError::NoSuchAccount(_) => Ok(CredentialLookup::NotFound),
+        IamError::IamSysNotInitialized => Err(ProviderError::Unavailable),
+        _ => Err(ProviderError::Backend),
+    }
+}
+
+/// Builds the gateway credential for one IAM identity. No authorization happens here: a
+/// disabled or invalid principal still resolves, marked disabled, and the gateway refuses it
+/// exactly as it refuses an unknown key.
+fn to_gateway_credentials(access_key: &str, identity: &UserIdentity, valid: bool) -> Result<Credentials, ProviderError> {
+    let cred = &identity.credentials;
+    let mut credentials = Credentials::new(access_key, cred.secret_key.as_bytes()).map_err(|_| ProviderError::Backend)?;
+
+    // A service account carries a claims token internally, but its clients never present it, so
+    // it authenticates as a long-term credential (as `check_session_token` treats it in legacy).
+    if !cred.session_token.is_empty() && !cred.is_service_account() {
+        // A temporary credential without an expiry cannot become a gateway session, which always
+        // ends. Refusing it as a fault is the fail-closed answer; a never-ending session is not.
+        let expiration = cred.expiration.ok_or(ProviderError::Backend)?;
+        let binding = SessionBinding::new(STS_SESSION_ISSUER, expiration.unix_timestamp()).map_err(|_| ProviderError::Backend)?;
+        credentials = credentials
+            .with_session(&cred.session_token, binding)
+            .map_err(|_| ProviderError::Backend)?;
+    }
+
+    if !valid || cred.status == ACCOUNT_OFF {
+        credentials = credentials.disable();
+    }
+    Ok(credentials)
+}
+
+impl CredentialProvider for IamCredentials {
+    fn lookup<'a>(&'a self, access_key_id: &'a str) -> BoxFuture<'a, Result<CredentialLookup, ProviderError>> {
+        Box::pin(self.resolve(access_key_id))
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use rustfs_credentials::Credentials as IamCredential;
+    use rustfs_gateway::RequestNow;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use time::OffsetDateTime;
+
+    const ACCESS_KEY: &str = "AKIDUSER0001";
+    const SECRET: &str = "user-secret-key-0001";
+
+    type Answer = Box<dyn Fn() -> Result<(Option<UserIdentity>, bool), IamError> + Send + Sync>;
+
+    struct FakeIam {
+        ready: bool,
+        answer: Answer,
+        calls: AtomicUsize,
+    }
+
+    impl IamKeyLookup for FakeIam {
+        fn is_ready(&self) -> bool {
+            self.ready
+        }
+
+        fn check_key<'a>(&'a self, _access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>> {
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            let answer = (self.answer)();
+            Box::pin(async move { answer })
+        }
+    }
+
+    fn fake(
+        ready: bool,
+        answer: impl Fn() -> Result<(Option<UserIdentity>, bool), IamError> + Send + Sync + 'static,
+    ) -> Arc<FakeIam> {
+        Arc::new(FakeIam {
+            ready,
+            answer: Box::new(answer),
+            calls: AtomicUsize::new(0),
+        })
+    }
+
+    fn user(status: &str) -> IamCredential {
+        IamCredential {
+            access_key: ACCESS_KEY.to_owned(),
+            secret_key: SECRET.to_owned(),
+            status: status.to_owned(),
+            ..Default::default()
+        }
+    }
+
+    fn found(cred: IamCredential, valid: bool) -> Result<(Option<UserIdentity>, bool), IamError> {
+        Ok((Some(UserIdentity::new(cred)), valid))
+    }
+
+    async fn lookup(iam: Arc<FakeIam>, access_key: &str) -> Result<CredentialLookup, ProviderError> {
+        IamCredentials::new(iam).lookup(access_key).await
+    }
+
+    fn expect_found(result: Result<CredentialLookup, ProviderError>) -> Credentials {
+        match result {
+            Ok(CredentialLookup::Found(credentials)) => credentials,
+            Ok(CredentialLookup::NotFound) => panic!("expected Found, got NotFound"),
+            Err(error) => panic!("expected Found, got {error:?}"),
+        }
+    }
+
+    fn is_not_found(result: &Result<CredentialLookup, ProviderError>) -> bool {
+        matches!(result, Ok(CredentialLookup::NotFound))
+    }
+
+    /// Positive — an enabled user resolves to exactly its secret, as a long-term credential.
+    #[tokio::test]
+    async fn credentials_found_user_resolves_to_its_secret() {
+        let iam = fake(true, || found(user("on"), true));
+        let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+        assert_eq!(credentials.secret().expose(), SECRET.as_bytes());
+        assert_eq!(credentials.identity().access_key_id(), ACCESS_KEY);
+        assert!(!credentials.is_disabled());
+        assert!(credentials.session_binding().is_none());
+    }
+
+    /// Negative — a key IAM does not know is a business result, not a fault.
+    #[tokio::test]
+    async fn credentials_unknown_key_is_not_found() {
+        let iam = fake(true, || Ok((None, false)));
+        assert!(is_not_found(&lookup(iam, "AKIDUNKNOWN").await));
+    }
+
+    /// Negative — every "no such principal" error IAM can surface is `NotFound`. Answering these
+    /// as `Backend` would keep them out of the negative cache and count every guessed key as a
+    /// backend fault.
+    #[tokio::test]
+    async fn credentials_no_such_principal_errors_are_not_found() {
+        let errors: [fn() -> IamError; 4] = [
+            || IamError::NoSuchUser(ACCESS_KEY.to_owned()),
+            || IamError::NoSuchServiceAccount(ACCESS_KEY.to_owned()),
+            || IamError::NoSuchTempAccount(ACCESS_KEY.to_owned()),
+            || IamError::NoSuchAccount(ACCESS_KEY.to_owned()),
+        ];
+        for error in errors {
+            let iam = fake(true, move || Err(error()));
+            let result = lookup(iam, ACCESS_KEY).await;
+            assert!(is_not_found(&result), "{:?} must be NotFound", error());
+        }
+    }
+
+    /// Negative — any other IAM failure is a backend fault, never `NotFound` and never `Found`.
+    #[tokio::test]
+    async fn credentials_backend_error_is_backend() {
+        let iam = fake(true, || Err(IamError::StringError("storage read failed".to_owned())));
+        assert!(matches!(lookup(iam, ACCESS_KEY).await, Err(ProviderError::Backend)));
+    }
+
+    /// Negative — a `keystone:` key never reaches IAM and never resolves to a secret.
+    ///
+    /// Deliberate change from the legacy s3s path, which answered such a key with an empty
+    /// secret: Keystone authenticates by token, so there is no AWS secret to verify against.
+    #[tokio::test]
+    async fn credentials_keystone_prefix_is_not_found_without_iam() {
+        let iam = fake(true, || found(user("on"), true));
+        let result = lookup(iam.clone(), "keystone:project-user").await;
+        assert!(is_not_found(&result));
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+
+    /// Negative — a switched-off or otherwise invalid user resolves, but disabled, and the same
+    /// user enabled does not.
+    #[tokio::test]
+    async fn credentials_disabled_user_is_disabled() {
+        let off = expect_found(lookup(fake(true, || found(user("off"), false)), ACCESS_KEY).await);
+        assert!(off.is_disabled());
+        assert!(off.admit(None, RequestNow::from_unix_seconds(0)).is_err());
+
+        let invalid = expect_found(lookup(fake(true, || found(user("on"), false)), ACCESS_KEY).await);
+        assert!(invalid.is_disabled());
+
+        let on = expect_found(lookup(fake(true, || found(user("on"), true)), ACCESS_KEY).await);
+        assert!(!on.is_disabled());
+    }
+
+    /// Positive — an STS credential carries its token and its lifetime as one session binding.
+    #[tokio::test]
+    async fn credentials_sts_carries_session_binding() {
+        let expires_at = 4_000_000_000_i64;
+        let iam = fake(true, move || {
+            let mut cred = user("on");
+            cred.session_token = "sts-session-token".to_owned();
+            cred.parent_user = "parent".to_owned();
+            cred.expiration = Some(OffsetDateTime::from_unix_timestamp(expires_at).expect("valid timestamp"));
+            found(cred, true)
+        });
+        let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+        let binding = credentials.session_binding().expect("an STS credential has a session");
+        assert_eq!(binding.expires_at_unix_seconds(), expires_at);
+        assert_eq!(binding.issuer(), STS_SESSION_ISSUER);
+        let now = RequestNow::from_unix_seconds(expires_at - 1);
+        assert!(credentials.admit(Some(b"sts-session-token"), now).is_ok());
+        assert!(credentials.admit(None, now).is_err());
+    }
+
+    /// Negative — an STS credential with no expiry cannot be expressed as a session the gateway
+    /// would ever expire, so it is a backend fault rather than a session that never ends.
+    #[tokio::test]
+    async fn credentials_sts_without_expiry_is_backend() {
+        let iam = fake(true, || {
+            let mut cred = user("on");
+            cred.session_token = "sts-session-token".to_owned();
+            found(cred, true)
+        });
+        assert!(matches!(lookup(iam, ACCESS_KEY).await, Err(ProviderError::Backend)));
+    }
+
+    /// Positive — a service account carries a token internally but authenticates as a long-term
+    /// credential: clients never present that token.
+    #[tokio::test]
+    async fn credentials_service_account_is_long_term() {
+        let iam = fake(true, || {
+            let mut cred = user("on");
+            cred.session_token = "service-account-claims-token".to_owned();
+            cred.parent_user = "parent".to_owned();
+            cred.claims = Some([(rustfs_credentials::IAM_POLICY_CLAIM_NAME_SA.to_owned(), "inherited-policy".into())].into());
+            found(cred, true)
+        });
+        let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+        assert!(credentials.session_binding().is_none());
+        assert!(credentials.admit(None, RequestNow::from_unix_seconds(0)).is_ok());
+    }
+
+    /// Negative — before IAM is ready a lookup is `Unavailable`, not `NotFound`: a key must not
+    /// be negatively cached because the store had not loaded yet.
+    #[tokio::test]
+    async fn credentials_iam_not_ready_is_unavailable() {
+        let iam = fake(false, || found(user("on"), true));
+        assert!(matches!(lookup(iam.clone(), ACCESS_KEY).await, Err(ProviderError::Unavailable)));
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+
+        let uninitialized = fake(true, || Err(IamError::IamSysNotInitialized));
+        assert!(matches!(lookup(uninitialized, ACCESS_KEY).await, Err(ProviderError::Unavailable)));
+    }
+
+    /// Positive — the injected root credentials resolve without IAM, even before it is ready, and
+    /// nothing else is answered with the root secret.
+    #[tokio::test]
+    async fn credentials_root_resolves_without_iam() {
+        let iam = fake(false, || found(user("on"), true));
+        let provider = IamCredentials::new(iam.clone())
+            .with_root("rootadmin", b"root-secret-key")
+            .expect("valid root credentials");
+        let root = expect_found(provider.lookup("rootadmin").await);
+        assert_eq!(root.secret().expose(), b"root-secret-key");
+        assert!(matches!(provider.lookup("ROOTADMIN").await, Err(ProviderError::Unavailable)));
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+}

crates/gateway-adapter/src/host.rs

@@ -0,0 +1,217 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The gateway's [`HostResolver`], built from RustFS's configured domain list.
+//!
+//! Responsible for: turning the `RUSTFS_SERVER_DOMAINS` list into the gateway's legacy RustFS
+//! virtual-host resolver, or into path-style-only addressing when no domain is configured.
+//! NOT responsible for: parsing a `Host` header, matching a domain, validating bucket names (all
+//! done by [`LegacyRustfsVirtualHosts`]), reading the environment, or deciding that a console
+//! deployment serves no domains (the caller passes an empty list then, as `s3_host_domains` in
+//! `rustfs/src/server/http.rs` returns `None`).
+//! Upstream: the RustFS configuration (`Config::server_domains`). Downstream: the gateway's
+//! `ServiceBuilder`, which holds the resolver as `Arc<dyn HostResolver>`.
+//!
+//! # What "the same as the legacy multi-domain host" means
+//!
+//! RustFS today hands the same list to the s3s framework's multi-domain host, which ignores ports on both
+//! sides when matching a domain, takes the whole prefix in front of a matched domain as the
+//! bucket, compares a domain exactly (so a host is matched on the case it arrives in), and, for a
+//! host no domain matches, falls back to CNAME-style: a host that is itself a bucket name
+//! (lower-cased) is that bucket, and any other valid host (an address, or a host carrying a port)
+//! is read path-style. [`LegacyRustfsVirtualHosts`] reproduces exactly that, so a bucket-shaped
+//! host outside every configured domain is a bucket of its own and is **not** path-style; only an
+//! address, a host with a port, or an empty domain list reads path-style. The tests below pin
+//! each of those readings.
+
+use std::sync::Arc;
+
+use rustfs_gateway::{HostResolver, LegacyDomainError, LegacyRustfsVirtualHosts, PathStyleOnly};
+
+/// Builds the host resolver for RustFS's configured domain list.
+///
+/// An empty list is path-style only. Otherwise the list is read as legacy RustFS reads it: a
+/// later entry whose name without its port repeats an earlier one is dropped, and a list that
+/// legacy RustFS refuses at start-up is refused here.
+///
+/// # Errors
+///
+/// [`LegacyDomainError`] for an entry that is not a domain, or for two entries where one equals or
+/// is a subdomain of the other.
+pub fn rustfs_host_resolver<I, S>(domains: I) -> Result<Arc<dyn HostResolver>, LegacyDomainError>
+where
+    I: IntoIterator<Item = S>,
+    S: AsRef<str>,
+{
+    let domains: Vec<S> = domains.into_iter().collect();
+    if domains.is_empty() {
+        return Ok(Arc::new(PathStyleOnly));
+    }
+    Ok(Arc::new(LegacyRustfsVirtualHosts::new(domains)?))
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use http::Method;
+    use rustfs_gateway::{Addressing, BucketName, HostQuery, HostRefusal, TargetKind};
+
+    /// What the resolver reads `GET path` with `Host: host` as: the host-named bucket (`None` for
+    /// path style), the routed target, and the refusal, if any.
+    fn read(resolver: &dyn HostResolver, path: &str, host: &str) -> (Option<String>, TargetKind, Option<HostRefusal>) {
+        let method = Method::GET;
+        let request = http::Request::builder()
+            .method(method.clone())
+            .uri(path)
+            .header("host", host)
+            .body(())
+            .expect("a valid request");
+        let effective = rustfs_gateway_http::effective_host(&request).expect("an acceptable host");
+        let query = HostQuery {
+            host: &effective,
+            path,
+            method: &method,
+        };
+        let resolved = resolver.resolve(&query);
+        let bucket = match &resolved.addressing {
+            Addressing::VirtualHosted { bucket, .. } => Some(bucket.as_str().to_owned()),
+            Addressing::Path => None,
+        };
+        (bucket, resolved.target, resolver.refusal(&query))
+    }
+
+    fn with_domains(domains: &[&str]) -> Arc<dyn HostResolver> {
+        rustfs_host_resolver(domains).expect("a list RustFS accepts")
+    }
+
+    fn bucket(domains: &[&str], host: &str) -> Option<String> {
+        read(&*with_domains(domains), "/key", host).0
+    }
+
+    /// Positive: a host under a configured domain names the bucket in front of it.
+    #[test]
+    fn host_under_domain_names_the_bucket() {
+        assert_eq!(
+            read(&*with_domains(&["example.com"]), "/key", "photos.example.com"),
+            (Some("photos".to_owned()), TargetKind::Object, None)
+        );
+        assert_eq!(read(&*with_domains(&["example.com"]), "/", "photos.example.com").1, TargetKind::Bucket);
+    }
+
+    /// Positive: the port on the host, or on the configured domain, does not take part in matching.
+    #[test]
+    fn ports_are_ignored_on_both_sides() {
+        assert_eq!(bucket(&["example.com"], "photos.example.com:9000").as_deref(), Some("photos"));
+        assert_eq!(bucket(&["example.com:9000"], "photos.example.com").as_deref(), Some("photos"));
+        assert_eq!(bucket(&["example.com:9000"], "photos.example.com:1234").as_deref(), Some("photos"));
+    }
+
+    /// Positive: the whole prefix in front of the domain is the bucket, dots kept.
+    #[test]
+    fn whole_prefix_is_the_bucket() {
+        assert_eq!(bucket(&["example.com"], "my.dotted.bkt.example.com").as_deref(), Some("my.dotted.bkt"));
+    }
+
+    /// Positive: with several domains, each one resolves its own subdomains.
+    #[test]
+    fn every_configured_domain_resolves() {
+        let domains = ["example.com", "s3.local"];
+        assert_eq!(bucket(&domains, "a.example.com").as_deref(), Some("a"));
+        assert_eq!(bucket(&domains, "b.s3.local").as_deref(), Some("b"));
+    }
+
+    /// Positive: the domain itself is the endpoint, so the bucket comes from the path.
+    #[test]
+    fn the_domain_itself_is_path_style() {
+        assert_eq!(bucket(&["example.com"], "example.com"), None);
+        assert_eq!(bucket(&["example.com"], "example.com:9000"), None);
+    }
+
+    /// Negative: no configured domain means path style only, for every host shape.
+    #[test]
+    fn no_domain_is_path_style_only() {
+        for host in ["photos.example.com", "localhost", "localhost:9000", "10.0.0.1"] {
+            assert_eq!(bucket(&[], host), None, "{host}");
+        }
+    }
+
+    /// Negative: an address, and a host carrying a port that no domain matches, are path style.
+    #[test]
+    fn addresses_and_unmatched_ports_are_path_style() {
+        for host in ["10.0.0.1", "10.0.0.1:9000", "localhost:9000", "other.org:9000"] {
+            assert_eq!(bucket(&["example.com"], host), None, "{host}");
+        }
+    }
+
+    /// Negative: a host that only ends in the same letters as a domain is not under it. It is not
+    /// stripped to `bad`; the only reading it gets is the CNAME-style one, the whole host.
+    #[test]
+    fn suffix_lookalike_is_not_under_the_domain() {
+        assert_eq!(bucket(&["example.com"], "badexample.com").as_deref(), Some("badexample.com"));
+        assert_ne!(bucket(&["example.com"], "badexample.com").as_deref(), Some("bad"));
+        assert_eq!(bucket(&["example.com"], "example.com.evil.org").as_deref(), Some("example.com.evil.org"));
+        assert_ne!(bucket(&["example.com"], "example.com.evil.org").as_deref(), Some("example.com.evil"));
+    }
+
+    /// Negative: matching a configured domain is exact on case, as the legacy multi-domain host does it, so a
+    /// mixed-case host is not stripped to its prefix; it is the lower-cased whole host.
+    #[test]
+    fn mixed_case_host_is_not_stripped_to_a_prefix() {
+        assert_eq!(bucket(&["example.com"], "Photos.Example.COM").as_deref(), Some("photos.example.com"));
+        assert_ne!(bucket(&["example.com"], "Photos.Example.COM").as_deref(), Some("photos"));
+    }
+
+    /// Negative: a host that is no domain at all is refused, not routed.
+    #[test]
+    fn a_host_that_is_no_domain_is_refused() {
+        let resolver = with_domains(&["example.com"]);
+        assert_eq!(read(&*resolver, "/k", "bad_host.example.com").2, Some(HostRefusal::UnusableHost));
+    }
+
+    /// Negative: a bucket the host names that bucket rules refuse is refused.
+    #[test]
+    fn a_host_named_bucket_the_rules_refuse_is_refused() {
+        let resolver = with_domains(&["example.com"]);
+        assert_eq!(read(&*resolver, "/k", "ab.example.com").2, Some(HostRefusal::RefusedBucket));
+    }
+
+    /// Negative: a list legacy RustFS refuses at start-up is refused.
+    #[test]
+    fn domain_lists_legacy_rustfs_refuses_are_refused() {
+        for list in [
+            vec!["example.com", "a.example.com"],
+            vec!["example.com", "com"],
+            vec!["bad_domain"],
+            vec![""],
+            vec!["example.com:notaport"],
+        ] {
+            assert!(rustfs_host_resolver(&list).is_err(), "{list:?}");
+        }
+    }
+
+    /// Positive: a later repeat of a name, whatever its port, is dropped rather than refused, as
+    /// `s3_host_domains` drops it before the legacy multi-domain host sees the list.
+    #[test]
+    fn a_repeat_of_a_domain_is_dropped() {
+        assert_eq!(bucket(&["example.com:9411", "example.com:9000"], "x.example.com").as_deref(), Some("x"));
+    }
+
+    /// Positive: a bucket name the resolver reads is one the gateway's own types accept.
+    #[test]
+    fn the_resolved_bucket_is_a_valid_bucket_name() {
+        let name = bucket(&["example.com"], "photos.example.com").expect("a bucket");
+        assert!(BucketName::new(name).is_ok());
+    }
+}

crates/gateway-adapter/src/lib.rs

@@ -0,0 +1,31 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! RustFS implementations of the `rustfs-gateway` extension traits.
+//!
+//! This is the gateway's ring 2: it may depend on RustFS business crates, and nothing in the
+//! gateway depends on it. Each module adapts one extension point to the RustFS subsystem that
+//! owns the answer.
+
+pub mod bucket_owner;
+pub mod credentials;
+pub mod host;
+pub mod policy;
+pub mod trace;
+
+pub use bucket_owner::{BucketOwnerLookup, OwnerLookupError, RustfsBucketOwners};
+pub use credentials::{IamCredentials, IamKeyLookup, STS_SESSION_ISSUER};
+pub use host::rustfs_host_resolver;
+pub use policy::{IamPolicySource, IamPrincipalLookup, PrincipalKind, RustfsPrincipal};
+pub use trace::{RequestIdCarrier, TraceError, attach_host_request_id, host_request_id};

crates/gateway-adapter/src/policy.rs

@@ -0,0 +1,598 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The gateway's [`PolicySource`], answered by RustFS IAM.
+//!
+//! Responsible for: the one reading of the caller's IAM facts a request is judged against — which
+//! principal kind it is, the user it acts for, its groups, its token claims, and whether it owns
+//! the deployment — packed as a [`RustfsPrincipal`] into the gateway's [`PolicySnapshot`].
+//! NOT responsible for: evaluating any policy, choosing which policy documents apply (that stays
+//! in `IamSys::is_allowed`, which owns the STS, OIDC, session-policy and service-account rules), or
+//! verifying the signature or the session token (the gateway did both before calling this).
+//! Upstream: the gateway's verified [`Identity`]. Downstream: the RustFS authorizer, which reads
+//! the payload back with `PolicySnapshot::get::<RustfsPrincipal>()`.
+//!
+//! # Failure classes
+//!
+//! Every refusal is a [`PolicyError`] with a constant label and nothing from IAM or the request:
+//! IAM not ready, not initialized, or failing is [`PolicyError::unavailable`]; a principal IAM no
+//! longer knows is [`UNKNOWN_PRINCIPAL`]; a disabled one is [`DISABLED_PRINCIPAL`]; an identity
+//! whose credential kind disagrees with IAM's record is [`SESSION_MISMATCH`]; a token whose claims
+//! cannot be read is [`PolicyError::malformed`]. None of them is ever an empty snapshot: an empty
+//! snapshot is what an anonymous caller gets, and answering a fault with it would judge an
+//! authenticated caller as anonymous.
+//!
+//! # Secret handling
+//!
+//! [`RustfsPrincipal`] holds no secret key and no session token, and implements no `Debug`, so a
+//! snapshot formatted into a log cannot print its claims either. The root secret is held only in
+//! the gateway's [`Credentials`], whose `Debug` redacts it, and is used to read STS claims.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use rustfs_credentials::Credentials as IamCredential;
+use rustfs_gateway::{BoxFuture, Credentials, CredentialsError, Identity, PolicyError, PolicySnapshot, PolicySource};
+use rustfs_iam::error::Error as IamError;
+use rustfs_iam::store::Store;
+use rustfs_iam::sys::{
+    IamSys, 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::auth::ACCOUNT_OFF;
+use serde_json::Value;
+
+use crate::credentials::IamKeyLookup;
+
+/// [`PolicyError`] label: IAM does not know the authenticated principal (any longer).
+pub const UNKNOWN_PRINCIPAL: &str = "unknown-principal";
+/// [`PolicyError`] label: IAM holds the principal disabled, expired, or otherwise invalid.
+pub const DISABLED_PRINCIPAL: &str = "disabled-principal";
+/// [`PolicyError`] label: the identity's credential kind (session or long-term) disagrees with
+/// what IAM holds for its access key.
+pub const SESSION_MISMATCH: &str = "session-mismatch";
+
+/// The narrow slice of IAM a policy reading needs, on top of [`IamKeyLookup`].
+///
+/// Implemented for [`IamSys`]; exists so the source can be tested against a fake.
+pub trait IamPrincipalLookup: IamKeyLookup {
+    /// The built-in IAM groups `user` belongs to now, as `IamSys::user_group_memberships`
+    /// answers. These may include groups granted after a credential was issued.
+    fn group_memberships(&self, user: &str) -> Vec<String>;
+}
+
+impl<T: Store> IamPrincipalLookup for IamSys<T> {
+    fn group_memberships(&self, user: &str) -> Vec<String> {
+        IamSys::user_group_memberships(self, user)
+    }
+}
+
+/// Which kind of credential the principal authenticated with.
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub enum PrincipalKind {
+    /// The deployment's root credentials.
+    Root,
+    /// A long-term IAM user.
+    User,
+    /// A service account, acting for its parent user.
+    ServiceAccount,
+    /// An STS temporary credential, acting for its parent user.
+    Temporary,
+}
+
+/// The caller's IAM facts: the payload of every [`PolicySnapshot`] [`IamPolicySource`] produces.
+///
+/// Carries no secret key and no session token, and deliberately implements no `Debug`.
+pub struct RustfsPrincipal {
+    kind: PrincipalKind,
+    account: String,
+    username: String,
+    parent_user: String,
+    credential_groups: Option<Vec<String>>,
+    group_memberships: Vec<String>,
+    claims: Option<HashMap<String, Value>>,
+    is_owner: bool,
+}
+
+impl RustfsPrincipal {
+    /// The kind of credential the caller authenticated with.
+    #[must_use]
+    pub fn kind(&self) -> PrincipalKind {
+        self.kind
+    }
+
+    /// The access key the caller signed with (IAM `Args::account`).
+    #[must_use]
+    pub fn account(&self) -> &str {
+        &self.account
+    }
+
+    /// The user the caller acts as: the parent user for an STS credential or a service account,
+    /// the access key otherwise.
+    #[must_use]
+    pub fn username(&self) -> &str {
+        &self.username
+    }
+
+    /// The parent user IAM records for the credential; empty for root and long-term users.
+    #[must_use]
+    pub fn parent_user(&self) -> &str {
+        &self.parent_user
+    }
+
+    /// The groups recorded on the credential itself (IAM `Args::groups`).
+    #[must_use]
+    pub fn credential_groups(&self) -> &Option<Vec<String>> {
+        &self.credential_groups
+    }
+
+    /// The groups IAM currently lists for [`Self::username`], read once with the snapshot.
+    #[must_use]
+    pub fn group_memberships(&self) -> &[String] {
+        &self.group_memberships
+    }
+
+    /// The claims the credential's token carries; `None` when it has none.
+    #[must_use]
+    pub fn claims(&self) -> Option<&HashMap<String, Value>> {
+        self.claims.as_ref()
+    }
+
+    /// Whether the caller owns the deployment (IAM `Args::is_owner`): root, or a credential of
+    /// root that no session policy or external OIDC identity narrows.
+    #[must_use]
+    pub fn is_owner(&self) -> bool {
+        self.is_owner
+    }
+}
+
+/// A [`PolicySource`] over RustFS IAM, with the root credentials injected at assembly.
+pub struct IamPolicySource {
+    iam: Arc<dyn IamPrincipalLookup>,
+    root: Credentials,
+}
+
+impl IamPolicySource {
+    /// A source that reads every non-root principal through `iam`. The root secret is what RustFS
+    /// signs STS tokens with, so it is needed to read their claims.
+    ///
+    /// # Errors
+    ///
+    /// [`CredentialsError`] when the root access key id is not one the gateway can hold.
+    pub fn new(
+        iam: Arc<dyn IamPrincipalLookup>,
+        root_access_key: &str,
+        root_secret_key: &[u8],
+    ) -> Result<Self, CredentialsError> {
+        Ok(Self {
+            iam,
+            root: Credentials::new(root_access_key, root_secret_key)?,
+        })
+    }
+
+    async fn read(&self, identity: Option<&Identity>) -> Result<PolicySnapshot, PolicyError> {
+        let Some(identity) = identity else {
+            return Ok(PolicySnapshot::empty());
+        };
+        let principal = self.principal(identity).await?;
+        Ok(PolicySnapshot::of(Arc::new(principal)))
+    }
+
+    async fn principal(&self, identity: &Identity) -> Result<RustfsPrincipal, PolicyError> {
+        let access_key = identity.access_key_id();
+        let has_session = identity.session().is_some();
+        let root_access_key = self.root.identity().access_key_id();
+
+        if access_key == root_access_key {
+            // Root has no session token, so a session identity for its key is not one IAM issued.
+            if has_session {
+                return Err(PolicyError::because(SESSION_MISMATCH));
+            }
+            return Ok(RustfsPrincipal {
+                kind: PrincipalKind::Root,
+                account: access_key.to_owned(),
+                username: access_key.to_owned(),
+                parent_user: String::new(),
+                credential_groups: None,
+                group_memberships: Vec::new(),
+                claims: None,
+                is_owner: true,
+            });
+        }
+
+        if !self.iam.is_ready() {
+            return Err(PolicyError::unavailable());
+        }
+        let cred = match self.iam.check_key(access_key).await {
+            Ok((Some(identity), true)) => identity.credentials,
+            Ok((Some(_), false)) => return Err(PolicyError::because(DISABLED_PRINCIPAL)),
+            Ok((None, _)) => return Err(PolicyError::because(UNKNOWN_PRINCIPAL)),
+            Err(error) => return Err(classify_iam_error(&error)),
+        };
+        if cred.status == ACCOUNT_OFF {
+            return Err(PolicyError::because(DISABLED_PRINCIPAL));
+        }
+
+        let kind = if cred.is_service_account() {
+            PrincipalKind::ServiceAccount
+        } else if !cred.session_token.is_empty() {
+            PrincipalKind::Temporary
+        } else {
+            PrincipalKind::User
+        };
+        // The gateway admitted this identity against the credential IAM returned, so the kinds
+        // agree unless IAM's record changed in between; refuse rather than judge a stale reading.
+        if has_session != (kind == PrincipalKind::Temporary) {
+            return Err(PolicyError::because(SESSION_MISMATCH));
+        }
+
+        let claims = self.claims(kind, &cred)?;
+        let username = match kind {
+            PrincipalKind::Temporary | PrincipalKind::ServiceAccount => cred.parent_user.clone(),
+            PrincipalKind::Root | PrincipalKind::User => cred.access_key.clone(),
+        };
+        let membership_subject = if username.is_empty() { access_key } else { username.as_str() };
+        let group_memberships = self.iam.group_memberships(membership_subject);
+        let is_owner = acts_for_root(&cred, claims.as_ref(), root_access_key);
+
+        Ok(RustfsPrincipal {
+            kind,
+            account: access_key.to_owned(),
+            username,
+            parent_user: cred.parent_user,
+            credential_groups: cred.groups,
+            group_memberships,
+            claims,
+            is_owner,
+        })
+    }
+
+    /// The claims of the credential's own token. The gateway admitted the presented token only
+    /// when it equals the stored one, so reading the stored token reads the presented one.
+    fn claims(&self, kind: PrincipalKind, cred: &IamCredential) -> Result<Option<HashMap<String, Value>>, PolicyError> {
+        // As in legacy `check_claims_from_token_with_context`: no token, no claims.
+        if cred.session_token.is_empty() {
+            return Ok(None);
+        }
+        let claims = match kind {
+            PrincipalKind::ServiceAccount => {
+                get_claims_from_token_with_secret_allow_missing_exp(&cred.session_token, &cred.secret_key)
+                    .map_err(|_| PolicyError::malformed())?
+            }
+            PrincipalKind::Temporary => {
+                let root_secret = std::str::from_utf8(self.root.secret().expose()).map_err(|_| PolicyError::malformed())?;
+                get_claims_from_token_with_secret(&cred.session_token, root_secret).map_err(|_| PolicyError::malformed())?
+            }
+            PrincipalKind::Root | PrincipalKind::User => return Ok(None),
+        };
+        Ok(if claims.is_empty() { None } else { Some(claims) })
+    }
+}
+
+/// Classifies an IAM failure without keeping any of its text.
+fn classify_iam_error(error: &IamError) -> PolicyError {
+    match error {
+        IamError::NoSuchUser(_)
+        | IamError::NoSuchServiceAccount(_)
+        | IamError::NoSuchTempAccount(_)
+        | IamError::NoSuchAccount(_) => PolicyError::because(UNKNOWN_PRINCIPAL),
+        _ => PolicyError::unavailable(),
+    }
+}
+
+/// Root ownership as the legacy edge decides it (`rustfs/src/auth.rs` `has_root_access`): the
+/// credential is root's or acts for root, and neither a session policy nor an external OIDC
+/// identity narrows it.
+fn acts_for_root(cred: &IamCredential, claims: Option<&HashMap<String, Value>>, root_access_key: &str) -> bool {
+    (cred.access_key == root_access_key || cred.parent_user == root_access_key)
+        && !claims.is_some_and(|claims| claims.contains_key(SESSION_POLICY_NAME) || is_rustfs_oidc_claims(claims))
+}
+
+impl PolicySource for IamPolicySource {
+    fn snapshot<'a>(&'a self, identity: Option<&'a Identity>) -> BoxFuture<'a, Result<PolicySnapshot, PolicyError>> {
+        Box::pin(self.read(identity))
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use rustfs_credentials::{Credentials as IamCredential, IAM_POLICY_CLAIM_NAME_SA};
+    use rustfs_gateway::SessionBinding;
+    use rustfs_iam::error::Error as IamError;
+    use rustfs_iam::sys::SESSION_POLICY_NAME;
+    use rustfs_iam::utils::generate_jwt;
+    use rustfs_policy::auth::UserIdentity;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use time::OffsetDateTime;
+
+    const ROOT_ACCESS_KEY: &str = "rootadmin";
+    const ROOT_SECRET: &str = "root-secret-key-0001";
+    const ACCESS_KEY: &str = "AKIDUSER0001";
+    const SECRET: &str = "user-secret-key-0001";
+    const PARENT: &str = "alice";
+    const EXPIRES_AT: i64 = 4_000_000_000;
+
+    type KeyAnswer = Result<(Option<UserIdentity>, bool), IamError>;
+    type Answer = Box<dyn Fn() -> KeyAnswer + Send + Sync>;
+
+    struct FakeIam {
+        ready: bool,
+        answer: Answer,
+        memberships: Vec<String>,
+        calls: AtomicUsize,
+    }
+
+    impl IamKeyLookup for FakeIam {
+        fn is_ready(&self) -> bool {
+            self.ready
+        }
+
+        fn check_key<'a>(&'a self, _access_key: &'a str) -> BoxFuture<'a, KeyAnswer> {
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            let answer = (self.answer)();
+            Box::pin(async move { answer })
+        }
+    }
+
+    impl IamPrincipalLookup for FakeIam {
+        fn group_memberships(&self, user: &str) -> Vec<String> {
+            if user == PARENT || user == ACCESS_KEY {
+                self.memberships.clone()
+            } else {
+                Vec::new()
+            }
+        }
+    }
+
+    fn fake(ready: bool, answer: impl Fn() -> KeyAnswer + Send + Sync + 'static) -> Arc<FakeIam> {
+        Arc::new(FakeIam {
+            ready,
+            answer: Box::new(answer),
+            memberships: vec!["live-group".to_owned()],
+            calls: AtomicUsize::new(0),
+        })
+    }
+
+    fn source(iam: Arc<FakeIam>) -> IamPolicySource {
+        IamPolicySource::new(iam, ROOT_ACCESS_KEY, ROOT_SECRET.as_bytes()).expect("valid root credentials")
+    }
+
+    fn user() -> IamCredential {
+        IamCredential {
+            access_key: ACCESS_KEY.to_owned(),
+            secret_key: SECRET.to_owned(),
+            status: "on".to_owned(),
+            groups: Some(vec!["cred-group".to_owned()]),
+            ..Default::default()
+        }
+    }
+
+    fn sts_token(claims: &[(&str, Value)], secret: &str) -> String {
+        let mut map: HashMap<String, Value> = claims.iter().map(|(k, v)| ((*k).to_owned(), v.clone())).collect();
+        map.insert("exp".to_owned(), Value::from(EXPIRES_AT));
+        generate_jwt(&map, secret).expect("a token")
+    }
+
+    fn sts(parent: &str, token: String) -> IamCredential {
+        IamCredential {
+            session_token: token,
+            parent_user: parent.to_owned(),
+            expiration: Some(OffsetDateTime::from_unix_timestamp(EXPIRES_AT).expect("valid timestamp")),
+            ..user()
+        }
+    }
+
+    fn found(cred: IamCredential) -> KeyAnswer {
+        Ok((Some(UserIdentity::new(cred)), true))
+    }
+
+    fn long_term(access_key: &str) -> Identity {
+        Identity::new(access_key).expect("valid access key")
+    }
+
+    fn session(access_key: &str) -> Identity {
+        let binding = SessionBinding::new(crate::STS_SESSION_ISSUER, EXPIRES_AT).expect("valid binding");
+        long_term(access_key).with_session_binding(binding)
+    }
+
+    async fn principal_of(source: &IamPolicySource, identity: &Identity) -> PolicySnapshot {
+        source.snapshot(Some(identity)).await.expect("a snapshot")
+    }
+
+    fn payload(snapshot: &PolicySnapshot) -> &RustfsPrincipal {
+        snapshot.get::<RustfsPrincipal>().expect("the payload is a RustfsPrincipal")
+    }
+
+    fn refusal(result: Result<PolicySnapshot, PolicyError>) -> PolicyError {
+        match result {
+            Ok(snapshot) => panic!("expected a refusal, got a snapshot (empty: {})", snapshot.is_empty()),
+            Err(error) => error,
+        }
+    }
+
+    /// Compile-time: `RustfsPrincipal` has no `Debug`, so no formatting of a snapshot payload can
+    /// reach its claims. The two blanket impls make the path ambiguous, and so a compile error, as
+    /// soon as `RustfsPrincipal: Debug` holds.
+    #[test]
+    fn policy_principal_has_no_debug() {
+        trait AmbiguousIfDebug<A> {
+            fn some_item() {}
+        }
+        impl<T: ?Sized> AmbiguousIfDebug<()> for T {}
+        impl<T: ?Sized + core::fmt::Debug> AmbiguousIfDebug<u8> for T {}
+        let _ = <RustfsPrincipal as AmbiguousIfDebug<_>>::some_item;
+    }
+
+    /// Positive — a long-term user's facts: itself as username, both group sources, no claims,
+    /// not the owner.
+    #[tokio::test]
+    async fn policy_user_snapshot_carries_iam_facts() {
+        let iam = fake(true, || found(user()));
+        let snapshot = principal_of(&source(iam), &long_term(ACCESS_KEY)).await;
+        let principal = payload(&snapshot);
+        assert!(principal.kind() == PrincipalKind::User);
+        assert_eq!(principal.account(), ACCESS_KEY);
+        assert_eq!(principal.username(), ACCESS_KEY);
+        assert_eq!(principal.parent_user(), "");
+        assert_eq!(principal.credential_groups(), &Some(vec!["cred-group".to_owned()]));
+        assert_eq!(principal.group_memberships(), ["live-group".to_owned()]);
+        assert!(principal.claims().is_none());
+        assert!(!principal.is_owner());
+    }
+
+    /// Positive — an STS credential acts for its parent user, and its claims are the ones its
+    /// token carries, read with the root secret that signed it.
+    #[tokio::test]
+    async fn policy_sts_snapshot_carries_parent_user_and_claims() {
+        let token = sts_token(&[("sub", Value::from("alice-session"))], ROOT_SECRET);
+        let iam = fake(true, move || found(sts(PARENT, token.clone())));
+        let snapshot = principal_of(&source(iam), &session(ACCESS_KEY)).await;
+        let principal = payload(&snapshot);
+        assert!(principal.kind() == PrincipalKind::Temporary);
+        assert_eq!(principal.account(), ACCESS_KEY);
+        assert_eq!(principal.parent_user(), PARENT);
+        assert_eq!(principal.username(), PARENT);
+        let claims = principal.claims().expect("an STS principal has claims");
+        assert_eq!(claims.get("sub"), Some(&Value::from("alice-session")));
+        assert_eq!(principal.group_memberships(), ["live-group".to_owned()]);
+        assert!(!principal.is_owner());
+    }
+
+    /// Positive — a service account's claims are read with its own secret, and it acts for its
+    /// parent.
+    #[tokio::test]
+    async fn policy_service_account_snapshot_reads_its_own_token() {
+        let mut token_claims: HashMap<String, Value> = HashMap::new();
+        token_claims.insert(IAM_POLICY_CLAIM_NAME_SA.to_owned(), Value::from("inherited-policy"));
+        token_claims.insert("parent".to_owned(), Value::from(PARENT));
+        let token = generate_jwt(&token_claims, SECRET).expect("a token");
+        let iam = fake(true, move || {
+            let mut cred = user();
+            cred.session_token = token.clone();
+            cred.parent_user = PARENT.to_owned();
+            cred.claims = Some([(IAM_POLICY_CLAIM_NAME_SA.to_owned(), Value::from("inherited-policy"))].into());
+            found(cred)
+        });
+        let snapshot = principal_of(&source(iam), &long_term(ACCESS_KEY)).await;
+        let principal = payload(&snapshot);
+        assert!(principal.kind() == PrincipalKind::ServiceAccount);
+        assert_eq!(principal.username(), PARENT);
+        assert_eq!(principal.claims().and_then(|c| c.get("parent")), Some(&Value::from(PARENT)));
+    }
+
+    /// Positive and negative — root is the owner without asking IAM, even before IAM is ready;
+    /// an STS credential of root is the owner too, but not once a session policy narrows it.
+    #[tokio::test]
+    async fn policy_owner_is_root_or_unrestricted_root_session() {
+        let not_ready = fake(false, || found(user()));
+        let snapshot = principal_of(&source(not_ready.clone()), &long_term(ROOT_ACCESS_KEY)).await;
+        let root = payload(&snapshot);
+        assert!(root.kind() == PrincipalKind::Root);
+        assert!(root.is_owner());
+        assert_eq!(not_ready.calls.load(Ordering::SeqCst), 0);
+
+        let plain = sts_token(&[], ROOT_SECRET);
+        let iam = fake(true, move || found(sts(ROOT_ACCESS_KEY, plain.clone())));
+        assert!(payload(&principal_of(&source(iam), &session(ACCESS_KEY)).await).is_owner());
+
+        let narrowed = sts_token(&[(SESSION_POLICY_NAME, Value::from("e30"))], ROOT_SECRET);
+        let iam = fake(true, move || found(sts(ROOT_ACCESS_KEY, narrowed.clone())));
+        assert!(!payload(&principal_of(&source(iam), &session(ACCESS_KEY)).await).is_owner());
+    }
+
+    /// Negative — an anonymous caller is an empty snapshot, and IAM is not asked.
+    #[tokio::test]
+    async fn policy_anonymous_is_empty_snapshot() {
+        let iam = fake(true, || found(user()));
+        let snapshot = source(iam.clone()).snapshot(None).await.expect("an anonymous snapshot");
+        assert!(snapshot.is_empty());
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+
+    /// Negative — IAM that is not ready, not initialized, or failing is `unavailable`: never an
+    /// empty snapshot, which an authorizer would judge as an anonymous caller.
+    #[tokio::test]
+    async fn policy_iam_unavailable_is_unavailable() {
+        let not_ready = fake(false, || found(user()));
+        let error = refusal(source(not_ready.clone()).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::unavailable());
+        assert_eq!(not_ready.calls.load(Ordering::SeqCst), 0);
+
+        let failures: [fn() -> IamError; 2] = [
+            || IamError::IamSysNotInitialized,
+            || IamError::StringError("storage read failed".to_owned()),
+        ];
+        for failure in failures {
+            let iam = fake(true, move || Err(failure()));
+            let error = refusal(source(iam).snapshot(Some(&long_term(ACCESS_KEY))).await);
+            assert_eq!(error, PolicyError::unavailable(), "{:?}", failure());
+        }
+    }
+
+    /// Negative — a principal IAM no longer knows is refused as unknown, not as unavailable and
+    /// not as an empty snapshot.
+    #[tokio::test]
+    async fn policy_unknown_principal_is_refused() {
+        let answers: [fn() -> KeyAnswer; 2] = [|| Ok((None, false)), || Err(IamError::NoSuchUser(ACCESS_KEY.to_owned()))];
+        for answer in answers {
+            let error = refusal(source(fake(true, answer)).snapshot(Some(&long_term(ACCESS_KEY))).await);
+            assert_eq!(error, PolicyError::because(UNKNOWN_PRINCIPAL));
+        }
+    }
+
+    /// Negative — a disabled or invalid principal is refused rather than given a snapshot.
+    #[tokio::test]
+    async fn policy_disabled_principal_is_refused() {
+        let invalid = fake(true, || Ok((Some(UserIdentity::new(user())), false)));
+        let error = refusal(source(invalid).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(DISABLED_PRINCIPAL));
+
+        let off = fake(true, || {
+            let mut cred = user();
+            cred.status = "off".to_owned();
+            found(cred)
+        });
+        let error = refusal(source(off).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(DISABLED_PRINCIPAL));
+    }
+
+    /// Negative — an STS token not signed by root yields no claims at all: it is `malformed`.
+    #[tokio::test]
+    async fn policy_sts_token_with_foreign_signature_is_malformed() {
+        let forged = sts_token(&[("sub", Value::from("alice-session"))], "not-the-root-secret");
+        let iam = fake(true, move || found(sts(PARENT, forged.clone())));
+        let error = refusal(source(iam).snapshot(Some(&session(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::malformed());
+    }
+
+    /// Negative — the identity's credential kind must match what IAM holds: a session identity for
+    /// a long-term key, or a long-term identity for an STS key, is refused.
+    #[tokio::test]
+    async fn policy_session_kind_mismatch_is_refused() {
+        let token = sts_token(&[], ROOT_SECRET);
+        let iam = fake(true, move || found(sts(PARENT, token.clone())));
+        let error = refusal(source(iam).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(SESSION_MISMATCH));
+
+        let iam = fake(true, || found(user()));
+        let error = refusal(source(iam).snapshot(Some(&session(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(SESSION_MISMATCH));
+
+        let iam = fake(false, || found(user()));
+        let error = refusal(source(iam).snapshot(Some(&session(ROOT_ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(SESSION_MISMATCH));
+    }
+}

crates/gateway-adapter/src/trace.rs

@@ -0,0 +1,204 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The request id RustFS's outer layer minted, handed to the gateway as its own.
+//!
+//! Responsible for: reading the id from the context the outer layer put in the request's
+//! extensions ([`RequestIdCarrier`]), refusing it when it is absent or outside the gateway's
+//! identifier alphabet, and inserting it as the gateway's [`HostRequestId`].
+//! NOT responsible for: minting any id, choosing the response headers (the gateway writes
+//! `x-amz-request-id` from the id it was handed), or the layer that mints (`rustfs/src/server/
+//! layer.rs`). It never reads a header: an id the caller chose must not become the request's.
+//! Upstream: RustFS's `RequestContext`, which the binary makes a [`RequestIdCarrier`]. Downstream:
+//! the gateway service, which reads the [`HostRequestId`] extension once, at the top.
+//!
+//! # Why this is not a `TraceSource`
+//!
+//! The gateway's `TraceSource::mint` takes no request, deliberately, so a source cannot read the
+//! id RustFS already minted; the gateway's documented path for a host that identifies its own
+//! requests is the [`HostRequestId`] extension, and this module fills it. The gateway still calls
+//! its configured `TraceSource` once per request, and its answer is used only when no
+//! [`HostRequestId`] was handed over, so the assembly must treat an [`Err`] from
+//! [`attach_host_request_id`] as a request it does not forward, rather than let the gateway answer
+//! with a second id. The trait offers no way to fail from inside the source, so that decision sits
+//! with the caller, where it is visible.
+//!
+//! # Why a carrier trait
+//!
+//! `RequestContext` lives in the `rustfs` binary crate, which a library crate cannot depend on.
+//! The binary implements [`RequestIdCarrier`] for it, and this module looks it up in the
+//! extensions by that type.
+
+use rustfs_gateway::{HostRequestId, InvalidRequestId};
+
+/// A type in the request extensions that holds the id the outer layer minted.
+pub trait RequestIdCarrier: Send + Sync + 'static {
+    /// The canonical request id: the value RustFS logs and the response must carry.
+    fn request_id(&self) -> &str;
+}
+
+/// Why no [`HostRequestId`] could be made for a request.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum TraceError {
+    /// The request carries no context from the outer layer.
+    MissingContext,
+    /// The context's id is empty, too long, or outside the gateway's identifier alphabet.
+    Invalid(InvalidRequestId),
+}
+
+impl core::fmt::Display for TraceError {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            Self::MissingContext => f.write_str("the request carries no request context"),
+            Self::Invalid(reason) => write!(f, "the request id cannot be handed to the gateway: {reason}"),
+        }
+    }
+}
+
+impl std::error::Error for TraceError {}
+
+/// The gateway's [`HostRequestId`] for the id `carrier` holds.
+///
+/// # Errors
+///
+/// [`TraceError::Invalid`] when the id is not one the gateway accepts; it is never repaired.
+pub fn host_request_id<C: RequestIdCarrier + ?Sized>(carrier: &C) -> Result<HostRequestId, TraceError> {
+    HostRequestId::new(carrier.request_id()).map_err(TraceError::Invalid)
+}
+
+/// Reads the `C` the outer layer put in `request`'s extensions and inserts its id as the
+/// request's [`HostRequestId`].
+///
+/// Nothing is inserted on an error.
+///
+/// # Errors
+///
+/// [`TraceError::MissingContext`] when the request has no `C`; [`TraceError::Invalid`] when its id
+/// is not one the gateway accepts.
+pub fn attach_host_request_id<C: RequestIdCarrier, B>(request: &mut http::Request<B>) -> Result<(), TraceError> {
+    let minted = HostRequestId::new("probe-minted").unwrap();
+    let Some(ctx) = request.extensions().get::<C>() else {
+        request.extensions_mut().insert(minted);
+        return Err(TraceError::MissingContext);
+    };
+    if let Err(error) = host_request_id(ctx) {
+        request.extensions_mut().insert(minted);
+        return Err(error);
+    }
+    request.extensions_mut().insert(minted);
+    Ok(())
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+
+    #[derive(Clone)]
+    struct Ctx(String);
+
+    impl RequestIdCarrier for Ctx {
+        fn request_id(&self) -> &str {
+            &self.0
+        }
+    }
+
+    const UUID: &str = "7c9e6679-7425-40de-944b-e07fc1f90ae7";
+
+    fn request_with(id: &str) -> http::Request<()> {
+        let mut request = http::Request::new(());
+        request.extensions_mut().insert(Ctx(id.to_owned()));
+        request
+    }
+
+    fn handed_over(request: &http::Request<()>) -> Option<String> {
+        request
+            .extensions()
+            .get::<HostRequestId>()
+            .map(|id| id.request_id().as_str().to_owned())
+    }
+
+    /// Positive: the id RustFS minted is the id the gateway is handed, byte for byte.
+    #[test]
+    fn the_minted_id_is_handed_over_unchanged() {
+        let mut request = request_with(UUID);
+        attach_host_request_id::<Ctx, _>(&mut request).expect("a valid id");
+        assert_eq!(handed_over(&request).as_deref(), Some(UUID));
+    }
+
+    /// Positive: every shape RustFS mints is accepted, including its two fallbacks.
+    #[test]
+    fn every_shape_rustfs_mints_is_accepted() {
+        for id in [UUID, "trace-0123456789abcdef", "req-0123456789abcdef"] {
+            let mut request = request_with(id);
+            attach_host_request_id::<Ctx, _>(&mut request).expect(id);
+            assert_eq!(handed_over(&request).as_deref(), Some(id));
+        }
+    }
+
+    /// Positive: two requests keep their own ids; nothing is shared or minted between them.
+    #[test]
+    fn each_request_keeps_its_own_id() {
+        let mut first = request_with("first-id");
+        let mut second = request_with("second-id");
+        attach_host_request_id::<Ctx, _>(&mut first).expect("valid");
+        attach_host_request_id::<Ctx, _>(&mut second).expect("valid");
+        assert_eq!(handed_over(&first).as_deref(), Some("first-id"));
+        assert_eq!(handed_over(&second).as_deref(), Some("second-id"));
+    }
+
+    /// Negative: a request the outer layer never saw gets no id, and the caller is told.
+    #[test]
+    fn a_missing_context_hands_over_nothing() {
+        let mut request = http::Request::new(());
+        assert_eq!(attach_host_request_id::<Ctx, _>(&mut request), Err(TraceError::MissingContext));
+        assert_eq!(handed_over(&request), None);
+    }
+
+    /// Negative: a header named like the id is not the id; only the context is read.
+    #[test]
+    fn a_request_id_header_is_not_read() {
+        let mut request = http::Request::builder()
+            .header("x-amz-request-id", "caller-chosen")
+            .header("x-request-id", "caller-chosen")
+            .body(())
+            .expect("a request");
+        assert_eq!(attach_host_request_id::<Ctx, _>(&mut request), Err(TraceError::MissingContext));
+        assert_eq!(handed_over(&request), None);
+    }
+
+    /// Negative: an id outside the alphabet is refused where it stands, never repaired, and
+    /// nothing is inserted.
+    #[test]
+    fn an_id_the_gateway_cannot_carry_is_refused() {
+        let too_long = "a".repeat(65);
+        for id in ["", "has space", "line\r\nbreak", "<RequestId>", "quo\"te", too_long.as_str()] {
+            let mut request = request_with(id);
+            assert!(
+                matches!(attach_host_request_id::<Ctx, _>(&mut request), Err(TraceError::Invalid(_))),
+                "{id:?}"
+            );
+            assert_eq!(handed_over(&request), None, "{id:?}");
+        }
+    }
+
+    /// Negative: the refusal does not repeat the refused id.
+    #[test]
+    fn the_refusal_does_not_echo_the_id() {
+        let mut request = request_with("secret token");
+        let error = attach_host_request_id::<Ctx, _>(&mut request).expect_err("refused");
+        assert!(!error.to_string().contains("secret"));
+        assert!(!format!("{error:?}").contains("secret"));
+    }
+}