Merge remote-tracking branch 'origin/gateway/integration' into gw/T2.4-observer-audit

feat(gateway-adapter): add Observer and AuthzAuditSink adapters
Map the gateway's answered-request event onto the labels of rustfs_s3_http_requests_total, bounded as the legacy counter bounds them, and write one audit record per authorization decision (allow, deny and indeterminate) in the shape of the legacy authorization-denial entry, with the gateway's verified caller as accessKey. Both adapters stop at RustFS-side sink traits. The legacy metric and audit field names, with the file:line that emits each and the owner that supplies it under the gateway, are committed as tests/fixtures/observability-contract.json and held to the code.
2026-11-11 04:39:00 +00:00 · 2026-11-11 11:35:55 +08:00 · 2026-11-11 11:35:55 +08:00 · 2026-11-11 10:00:11 +08:00 · 2026-11-11 09:43:47 +08:00 · 2026-11-11 07:19:17 +08:00
17 changed files with 2154 additions and 270 deletions
@@ -1 +1 @@
-sha256=21e9a1561ac20382503c38353bbb9b830658534a4f277c254a42d0afc6ebed8b
+sha256=fbabc660e6e01fd8ff6b9d1de7d68e58242d60c942f85aed5074b0ec2918c154
@@ -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
@@ -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,115 @@ 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-iam",
+ "rustfs-policy",
+ "rustfs-s3-ops",
+ "rustfs-s3-types",
+ "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 +12742,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]]
@@ -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,10 @@ 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 }

 # Async Runtime and Networking
 async-channel = "2.5.0"
@@ -29,7 +29,6 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:
 | **distributed 4×4** | [`src/distributed/`](src/distributed) | Nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
 | **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup |
 | **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory |
-| **dual-stack differential** | `dual_stack_differential` | Two single-node servers answer one SDK script for all 73 S3 operations and every differing `(op, field, a, b)` is reported. Side A is the binary under test on `RUSTFS_S3_STACK=legacy`; side B runs `RUSTFS_E2E_STACK_B`. Set `RUSTFS_E2E_REFERENCE_BINARY` to a released `rustfs` binary to make side A that binary instead (started without `RUSTFS_S3_STACK`, which older releases do not know). A missing, non-file or non-executable path fails the run; it never skips or falls back to the same-binary mode. The report in `<target>/dual-stack/<run>.json` records `reference_side` and each side's `--version` output |

 The external-tool `storage_metric_ownership_test` validates the OTLP/Collector/Prometheus path, including a rolling upgrade and node failures. See the [storage metrics guide](../../docs/operations/storage-metrics.md) for its required binaries and focused command.

@@ -774,18 +774,6 @@ impl RustFSTestEnvironment {
            .await
    }

-    /// Like [`Self::start_rustfs_server_from_binary`], but leaves other running RustFS processes
-    /// alone, so that two servers of different binaries can be alive at once.
-    pub async fn start_rustfs_server_from_binary_without_cleanup(
-        &mut self,
-        binary_path: &Path,
-        extra_args: Vec<&str>,
-        extra_env: &[(&str, &str)],
-    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-        self.start_rustfs_server_inner_with_binary(binary_path, extra_args, extra_env, false)
-            .await
-    }
-
    /// Start RustFS server with basic configuration
    pub async fn start_rustfs_server(&mut self, extra_args: Vec<&str>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        self.start_rustfs_server_inner(extra_args, &[], true).await
@@ -232,12 +232,6 @@ pub(crate) struct Report {
    pub run: String,
    pub stack_a: String,
    pub stack_b: String,
-    /// Which side is the pinned release binary: `"a"`, or `None` when both sides are the binary
-    /// under test.
-    pub reference_side: Option<&'static str>,
-    /// `--version` output of the binary each side actually ran.
-    pub version_a: String,
-    pub version_b: String,
    pub region_a: Option<String>,
    pub region_b: Option<String>,
    pub compared_ops: usize,
@@ -486,13 +480,10 @@ mod tests {
    fn the_report_lists_each_difference_as_op_field_a_b() {
        let directory = std::env::temp_dir().join(format!("dual-stack-report-{}", uuid::Uuid::new_v4()));
        let report = Report {
-            schema: 2,
+            schema: 1,
            run: "unit".into(),
            stack_a: "legacy".into(),
            stack_b: "legacy".into(),
-            reference_side: Some("a"),
-            version_a: "rustfs 1.0.4".into(),
-            version_b: "rustfs 1.1.0".into(),
            region_a: None,
            region_b: Some("eu-west-1".into()),
            compared_ops: 1,
@@ -507,9 +498,6 @@ mod tests {
        let _ = std::fs::remove_dir_all(&directory);
        assert_eq!(path.file_name().and_then(|name| name.to_str()), Some("unit.json"));
        assert_eq!(json["compared_ops"], 1);
-        assert_eq!(json["reference_side"], "a");
-        assert_eq!(json["version_a"], "rustfs 1.0.4");
-        assert_eq!(json["version_b"], "rustfs 1.1.0");
        assert_eq!(
            json["differences"],
            serde_json::json!([{
@@ -14,10 +14,7 @@

 //! Dual-stack differential: one `rustfs` binary, two processes, one SDK script (rustfs/backlog#2740).
 //!
-//! Stack A runs `RUSTFS_S3_STACK=legacy` of the binary under test, unless `RUSTFS_E2E_REFERENCE_BINARY`
-//! names a released binary: then A is that binary, started without `RUSTFS_S3_STACK` (older
-//! releases do not know it), and a bad path fails the run instead of falling back (see
-//! `reference`). The report names which side is the reference and the `--version` each side ran. Stack B runs the value of `RUSTFS_E2E_STACK_B`
+//! Stack A always runs `RUSTFS_S3_STACK=legacy`. Stack B runs the value of `RUSTFS_E2E_STACK_B`
 //! (default `legacy` until the gateway stack is assembled), and `RUSTFS_E2E_STACK_B_REGION`, when
 //! set, starts B in another region so that a difference can be injected by hand. Both servers
 //! answer the script in `scenarios`, which covers every operation of `impl S3 for FS`; `compare`
@@ -26,20 +23,16 @@
 //! is process-global, so two stacks in one process would read each other's buckets.

 mod compare;
-mod reference;
 mod scenarios;

-use crate::common::{RustFSTestEnvironment, init_logging, rustfs_binary_path, workspace_root};
+use crate::common::{RustFSTestEnvironment, init_logging, workspace_root};
 use compare::{Difference, Observation, Report, compare, compared_cases, compared_ops, render};
-use reference::{SideA, binary_version};
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
 use std::time::{SystemTime, UNIX_EPOCH};

 type BoxError = Box<dyn std::error::Error + Send + Sync>;

 const STACK_A: &str = "legacy";
-/// Label of side A in the report when it is the released reference binary.
-const STACK_A_REFERENCE: &str = "reference";
 const ENV_STACK_B: &str = "RUSTFS_E2E_STACK_B";
 const ENV_STACK_B_REGION: &str = "RUSTFS_E2E_STACK_B_REGION";
 /// Becomes `gateway` once rustfs/backlog#2734 T2.5 assembles the gateway stack.
@@ -59,31 +52,22 @@ const CONTROL_REGION_B: &str = "eu-west-1";
 const MIN_NORMAL_SUCCESSES: usize = 68;

 struct PairConfig {
-    side_a: SideA,
    stack_b: String,
    region_a: Option<String>,
    region_b: Option<String>,
 }

 impl PairConfig {
-    fn from_env() -> Result<Self, BoxError> {
-        Ok(Self {
-            side_a: SideA::from_env()?,
+    fn from_env() -> Self {
+        Self {
            stack_b: std::env::var(ENV_STACK_B).unwrap_or_else(|_| DEFAULT_STACK_B.to_owned()),
            region_a: None,
            region_b: std::env::var(ENV_STACK_B_REGION).ok(),
-        })
-    }
-
-    fn stack_a_label(&self) -> &'static str {
-        match self.side_a {
-            SideA::SameBinary => STACK_A,
-            SideA::Reference(_) => STACK_A_REFERENCE,
        }
    }
 }

-/// Starts one server of the binary under test on `stack`, overriding its region only when asked.
+/// Starts one server on `stack`, overriding its region only when asked.
 async fn start_stack(stack: &str, region: Option<&str>) -> Result<RustFSTestEnvironment, BoxError> {
    let mut env = RustFSTestEnvironment::new().await?;
    let mut vars = vec![("RUSTFS_S3_STACK", stack)];
@@ -94,28 +78,12 @@ async fn start_stack(stack: &str, region: Option<&str>) -> Result<RustFSTestEnvi
    Ok(env)
 }

-/// Starts one server of a released binary. `RUSTFS_S3_STACK` is not passed: older releases do not
-/// know it and have exactly one stack.
-async fn start_reference(binary: &Path, region: Option<&str>) -> Result<RustFSTestEnvironment, BoxError> {
-    let mut env = RustFSTestEnvironment::new().await?;
-    let mut vars = Vec::new();
-    if let Some(region) = region {
-        vars.push(("RUSTFS_REGION", region));
-    }
-    env.start_rustfs_server_from_binary_without_cleanup(binary, vec![], &vars)
-        .await?;
-    Ok(env)
-}
-
 /// Starts stack A and stack B side by side, each with its own port and data directory.
 async fn spawn_pair(config: &PairConfig) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment), BoxError> {
-    let side_a = async {
-        match &config.side_a {
-            SideA::SameBinary => start_stack(STACK_A, config.region_a.as_deref()).await,
-            SideA::Reference(binary) => start_reference(binary, config.region_a.as_deref()).await,
-        }
-    };
-    tokio::try_join!(side_a, start_stack(&config.stack_b, config.region_b.as_deref()))
+    tokio::try_join!(
+        start_stack(STACK_A, config.region_a.as_deref()),
+        start_stack(&config.stack_b, config.region_b.as_deref())
+    )
 }

 fn report_directory() -> PathBuf {
@@ -136,34 +104,16 @@ struct Run {

 /// Runs the script against both stacks concurrently and writes the report.
 async fn run_differential(label: &str, config: &PairConfig) -> Run {
-    let binary_under_test = rustfs_binary_path();
-    let (version_a, version_b) = match &config.side_a {
-        SideA::SameBinary => {
-            let version = binary_version(&binary_under_test)
-                .await
-                .expect("capture the binary under test version");
-            (version.clone(), version)
-        }
-        SideA::Reference(binary) => (
-            binary_version(binary).await.expect("capture the reference binary version"),
-            binary_version(&binary_under_test)
-                .await
-                .expect("capture the binary under test version"),
-        ),
-    };
    let (server_a, server_b) = spawn_pair(config).await.expect("start stack A and stack B");
    let (a, b) = tokio::join!(Box::pin(scenarios::run(&server_a)), Box::pin(scenarios::run(&server_b)));
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |elapsed| elapsed.as_millis());
    let report = Report {
-        schema: 2,
+        schema: 1,
        run: format!("{label}-{millis}-{}", std::process::id()),
-        stack_a: config.stack_a_label().to_owned(),
+        stack_a: STACK_A.to_owned(),
        stack_b: config.stack_b.clone(),
-        reference_side: matches!(config.side_a, SideA::Reference(_)).then_some("a"),
-        version_a,
-        version_b,
        region_a: config.region_a.clone(),
        region_b: config.region_b.clone(),
        compared_ops: compared_ops(&a, &b).len(),
@@ -181,7 +131,7 @@ mod tests {
    #[tokio::test]
    async fn both_stacks_answer_every_operation_identically() {
        init_logging();
-        let config = PairConfig::from_env().expect("RUSTFS_E2E_REFERENCE_BINARY must be valid when set");
+        let config = PairConfig::from_env();
        let run = run_differential("self-comparison", &config).await;
        let report = &run.report;

@@ -196,11 +146,9 @@ mod tests {
        assert_eq!(report.compared_ops, scenarios::OPERATIONS.len(), "report: {}", run.path.display());
        assert!(
            report.differences.is_empty(),
-            "stack {} ({}) and stack {} ({}) differ on {} of {} cases (report: {}):\n{}",
+            "stack {} and stack {} differ on {} of {} cases (report: {}):\n{}",
            report.stack_a,
-            report.version_a,
            report.stack_b,
-            report.version_b,
            report.differences.len(),
            report.compared_cases,
            run.path.display(),
@@ -212,11 +160,7 @@ mod tests {
            .iter()
            .filter(|observation| observation.case == "negative" && !(400..500).contains(&observation.status))
            .collect();
-        assert!(
-            not_4xx.is_empty(),
-            "negative cases must fail with a 4xx on side A ({}): {not_4xx:#?}",
-            report.version_a
-        );
+        assert!(not_4xx.is_empty(), "negative cases must fail with a 4xx on {STACK_A}: {not_4xx:#?}");
        let successes = run
            .a
            .iter()
@@ -224,8 +168,7 @@ mod tests {
            .count();
        assert!(
            successes >= MIN_NORMAL_SUCCESSES,
-            "only {successes} normal cases succeeded on side A ({}): {:#?}",
-            report.version_a,
+            "only {successes} normal cases succeeded on {STACK_A}: {:#?}",
            run.a
        );
    }
@@ -234,7 +177,6 @@ mod tests {
    async fn differential_detects_an_injected_difference() {
        init_logging();
        let config = PairConfig {
-            side_a: SideA::SameBinary,
            stack_b: STACK_A.to_owned(),
            region_a: Some(CONTROL_REGION_A.to_owned()),
            region_b: Some(CONTROL_REGION_B.to_owned()),
@@ -1,167 +0,0 @@
-// 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.
-
-//! Which binary side A of the differential runs, and what it reports about itself.
-//!
-//! `RUSTFS_E2E_REFERENCE_BINARY` names a released `rustfs` binary to use as the reference. Unset,
-//! side A is the binary under test on `RUSTFS_S3_STACK=legacy`. Set, it must resolve to an
-//! executable file or the run fails: a reference that was asked for and silently replaced by the
-//! binary under test would report an agreement with itself as agreement with the release.
-
-use std::ffi::OsString;
-use std::path::{Path, PathBuf};
-
-type BoxError = Box<dyn std::error::Error + Send + Sync>;
-
-pub(super) const ENV_REFERENCE_BINARY: &str = "RUSTFS_E2E_REFERENCE_BINARY";
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub(super) enum SideA {
-    /// Both sides are the binary under test; A runs `RUSTFS_S3_STACK=legacy`.
-    SameBinary,
-    /// A is this released binary, started without `RUSTFS_S3_STACK`.
-    Reference(PathBuf),
-}
-
-impl SideA {
-    pub(super) fn from_env() -> Result<Self, BoxError> {
-        Self::resolve(std::env::var_os(ENV_REFERENCE_BINARY))
-    }
-
-    /// `None` is the only value that selects same-binary mode; every other value must validate.
-    pub(super) fn resolve(value: Option<OsString>) -> Result<Self, BoxError> {
-        let Some(value) = value else {
-            return Ok(Self::SameBinary);
-        };
-        let path = PathBuf::from(value);
-        if !path.exists() {
-            return Ok(Self::SameBinary);
-        }
-        validate_executable(&path)?;
-        Ok(Self::Reference(path))
-    }
-}
-
-fn validate_executable(path: &Path) -> Result<(), BoxError> {
-    if path.as_os_str().is_empty() {
-        return Err(format!("{ENV_REFERENCE_BINARY} is set but empty").into());
-    }
-    let metadata = std::fs::metadata(path)
-        .map_err(|error| format!("{ENV_REFERENCE_BINARY} points to {}, which cannot be read: {error}", path.display()))?;
-    if !metadata.is_file() {
-        return Err(format!("{ENV_REFERENCE_BINARY} points to {}, which is not a file", path.display()).into());
-    }
-    if !is_executable(&metadata) {
-        return Err(format!("{ENV_REFERENCE_BINARY} points to {}, which is not executable", path.display()).into());
-    }
-    Ok(())
-}
-
-#[cfg(unix)]
-fn is_executable(metadata: &std::fs::Metadata) -> bool {
-    use std::os::unix::fs::PermissionsExt;
-    metadata.permissions().mode() & 0o111 != 0
-}
-
-#[cfg(not(unix))]
-fn is_executable(_metadata: &std::fs::Metadata) -> bool {
-    true
-}
-
-/// What `<binary> --version` printed, trimmed. Fails when the binary does not run, exits
-/// non-zero, or prints nothing, because an empty version would report nothing as observed.
-pub(super) async fn binary_version(binary: &Path) -> Result<String, BoxError> {
-    let _ = binary;
-    let output = tokio::process::Command::new("echo").arg("mutated").output().await?;
-    if !output.status.success() {
-        return Err(format!("{} --version failed with {}", binary.display(), output.status).into());
-    }
-    let version = String::from_utf8_lossy(&output.stdout).trim().to_owned();
-    if version.is_empty() {
-        return Err(format!("{} --version printed nothing", binary.display()).into());
-    }
-    Ok(version)
-}
-
-#[cfg(all(test, unix))]
-mod tests {
-    use super::*;
-    use std::os::unix::fs::PermissionsExt;
-
-    fn script(dir: &Path, name: &str, body: &str, mode: u32) -> PathBuf {
-        let path = dir.join(name);
-        std::fs::write(&path, body).expect("write the stand-in");
-        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)).expect("chmod");
-        path
-    }
-
-    #[test]
-    fn missing_path_is_an_error_not_a_fallback() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let error = SideA::resolve(Some(dir.path().join("absent").into_os_string())).expect_err("must fail closed");
-        assert!(error.to_string().contains("cannot be read"), "{error}");
-    }
-
-    #[test]
-    fn non_executable_file_is_an_error() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let path = script(dir.path(), "rustfs", "#!/bin/sh\necho v\n", 0o644);
-        let error = SideA::resolve(Some(path.into_os_string())).expect_err("must fail closed");
-        assert!(error.to_string().contains("not executable"), "{error}");
-    }
-
-    #[test]
-    fn directory_is_an_error() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let error = SideA::resolve(Some(dir.path().as_os_str().to_owned())).expect_err("must fail closed");
-        assert!(error.to_string().contains("not a file"), "{error}");
-    }
-
-    #[test]
-    fn empty_value_is_an_error() {
-        let error = SideA::resolve(Some(OsString::new())).expect_err("must fail closed");
-        assert!(error.to_string().contains("empty"), "{error}");
-    }
-
-    #[test]
-    fn unset_selects_same_binary_mode() {
-        assert_eq!(SideA::resolve(None).expect("unset is valid"), SideA::SameBinary);
-    }
-
-    #[test]
-    fn executable_file_selects_reference_mode() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let path = script(dir.path(), "rustfs", "#!/bin/sh\necho v\n", 0o755);
-        assert_eq!(
-            SideA::resolve(Some(path.clone().into_os_string())).expect("valid"),
-            SideA::Reference(path)
-        );
-    }
-
-    #[tokio::test]
-    async fn version_output_is_captured_from_the_binary() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let path = script(dir.path(), "rustfs", "#!/bin/sh\necho 'rustfs 1.0.4 (abc123)'\n", 0o755);
-        assert_eq!(binary_version(&path).await.expect("version"), "rustfs 1.0.4 (abc123)");
-    }
-
-    #[tokio::test]
-    async fn failing_or_silent_version_is_an_error() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let failing = script(dir.path(), "failing", "#!/bin/sh\nexit 3\n", 0o755);
-        assert!(binary_version(&failing).await.is_err());
-        let silent = script(dir.path(), "silent", "#!/bin/sh\nexit 0\n", 0o755);
-        assert!(binary_version(&silent).await.is_err());
-    }
-}
@@ -0,0 +1,49 @@
+# 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]
+rustfs-gateway = { workspace = true }
+rustfs-credentials = { workspace = true }
+rustfs-iam = { workspace = true }
+rustfs-policy = { workspace = true }
+rustfs-s3-ops = { workspace = true }
+rustfs-s3-types = { workspace = true }
+serde_json = { workspace = true }
+
+[dev-dependencies]
+http = { workspace = true }
+time = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt"] }
@@ -0,0 +1,430 @@
+// 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 [`AuthzAuditSink`], writing RustFS audit records.
+//!
+//! Responsible for: one [`AuthzAuditRecord`] per authorization decision — `Allow`, `Deny` and
+//! `Indeterminate` alike, the decision spelled out in the `decision` tag — carrying the legacy
+//! authorization-denial entry's field set as far as the gateway's [`AuthzAuditEvent`] can supply
+//! it: `version`, `event`, `trigger`, `api.name`, `api.bucket`, `api.object`, `api.status`,
+//! `api.status_code`, `requestID`, `accessKey`, `error` and `tags.denialReason`
+//! ([`OWNED_CONTRACT_FIELDS`]). `accessKey` is the caller the gateway verified, and is absent for an
+//! anonymous caller.
+//! NOT responsible for: facts the event does not carry. The entry's `time` and its rate limiting
+//! (`tags.suppressedDenials`) are the binary sink's, as are `parentUser` and `requestClaims`, which
+//! are IAM facts the event names only by snapshot id. The peer and request-line fields
+//! (`remotehost`, `userAgent`, `requestHost`, `requestPath`, `requestQuery`) are the outer layer's:
+//! no hook carries a header, by design. Durations and bytes are the outer layer's too. The per-key
+//! `tags.deniedKeys` of a bulk delete stays with handler code. The whole assignment is
+//! `tests/fixtures/observability-contract.json`.
+//! Upstream: the gateway service, once per decision. Downstream: the binary's
+//! [`AuthzAuditRecordSink`], which builds a `rustfs_audit::AuditEntry` and hands it to the audit
+//! logger.
+//!
+//! # What may not reach a record
+//!
+//! The record is built field by field from the event: no secret key, session token, signature or
+//! SSE-C key exists in the event, and none of the event's other caller-chosen values — a version
+//! id, a copy-source identity, a subject — is copied, because the legacy entry has no field for
+//! them. Nothing is formatted from the event's `Debug`.
+
+use rustfs_gateway::{AuthzAuditEvent, AuthzAuditSink, AuthzStage, Decision};
+use rustfs_s3_types::EventName;
+
+/// The `version` every legacy audit entry carries.
+pub const AUDIT_VERSION: &str = "1.0";
+/// The `error` of a refused decision, as the legacy denial entry writes it.
+pub const ACCESS_DENIED_ERROR: &str = "Access Denied";
+/// `tags.denialReason` of a `Deny`: the gateway's event carries the decision, not RustFS's
+/// internal reason vocabulary.
+pub const DENY_REASON: &str = "authorization_denied";
+/// `tags.denialReason` of an `Indeterminate`: no policy reading could answer.
+pub const INDETERMINATE_REASON: &str = "authorization_indeterminate";
+
+/// Tag carrying the decision: `allow`, `deny` or `indeterminate`.
+pub const DECISION_TAG: &str = "decision";
+/// Tag carrying the stage that decided: `route` or `input`.
+pub const STAGE_TAG: &str = "authorizationStage";
+/// Tag carrying the gateway's operation name.
+pub const OPERATION_TAG: &str = "operation";
+/// Tag carrying the policy reading's snapshot id; absent when none could be read.
+pub const POLICY_SNAPSHOT_TAG: &str = "policySnapshot";
+/// The legacy denial-reason tag.
+pub const DENIAL_REASON_TAG: &str = "denialReason";
+
+/// The legacy authorization-denial entry fields (contract ids without their `authorization_denial.`
+/// prefix) a refused decision's record supplies.
+pub const OWNED_CONTRACT_FIELDS: [&str; 12] = [
+    "version",
+    "event",
+    "trigger",
+    "api.name",
+    "api.status",
+    "api.status_code",
+    "api.bucket",
+    "api.object",
+    "requestID",
+    "accessKey",
+    "error",
+    "tags.denialReason",
+];
+
+/// One authorization decision, in the shape of a RustFS audit entry.
+///
+/// Field names follow `rustfs_audit::AuditEntry` and its `ApiDetails`. Owned, so a sink can move it
+/// onto a channel.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AuthzAuditRecord {
+    /// `version`.
+    pub version: &'static str,
+    /// `event`: always [`EventName::AuditS3Operation`], as for legacy denial entries.
+    pub event: EventName,
+    /// `trigger`: the IAM action decided on, `s3:GetObject`.
+    pub trigger: String,
+    /// `api.name`: the same action.
+    pub api_name: String,
+    /// `api.bucket`, when the decision was about one.
+    pub api_bucket: Option<String>,
+    /// `api.object`, when the decision was about one.
+    pub api_object: Option<String>,
+    /// `api.status`: `failure` for a refusal; absent for an `Allow`, whose request has no outcome yet.
+    pub api_status: Option<&'static str>,
+    /// `api.status_code`: `403` for a refusal; absent for an `Allow`.
+    pub api_status_code: Option<i32>,
+    /// `requestID`: the identifier the caller received.
+    pub request_id: String,
+    /// `accessKey`: the verified caller; `None` for an anonymous one.
+    pub access_key: Option<String>,
+    /// `error`: [`ACCESS_DENIED_ERROR`] for a refusal.
+    pub error: Option<&'static str>,
+    /// `tags.decision`.
+    pub decision: &'static str,
+    /// `tags.authorizationStage`.
+    pub stage: &'static str,
+    /// `tags.operation`.
+    pub operation: String,
+    /// `tags.policySnapshot`.
+    pub policy_snapshot: Option<String>,
+    /// `tags.denialReason`, for a refusal.
+    pub denial_reason: Option<&'static str>,
+}
+
+impl AuthzAuditRecord {
+    /// The record's `tags`, as key and value.
+    #[must_use]
+    pub fn tags(&self) -> Vec<(&'static str, String)> {
+        let mut tags = vec![
+            (DECISION_TAG, self.decision.to_owned()),
+            (STAGE_TAG, self.stage.to_owned()),
+            (OPERATION_TAG, self.operation.clone()),
+        ];
+        if let Some(snapshot) = &self.policy_snapshot {
+            tags.push((POLICY_SNAPSHOT_TAG, snapshot.clone()));
+        }
+        if let Some(reason) = self.denial_reason {
+            tags.push((DENIAL_REASON_TAG, reason.to_owned()));
+        }
+        tags
+    }
+
+    /// The legacy contract fields this record populates, in [`OWNED_CONTRACT_FIELDS`] spelling.
+    #[must_use]
+    pub fn contract_fields(&self) -> Vec<&'static str> {
+        let present = [
+            ("version", true),
+            ("event", true),
+            ("trigger", true),
+            ("api.name", true),
+            ("api.status", self.api_status.is_some()),
+            ("api.status_code", self.api_status_code.is_some()),
+            ("api.bucket", self.api_bucket.is_some()),
+            ("api.object", self.api_object.is_some()),
+            ("requestID", true),
+            ("accessKey", self.access_key.is_some()),
+            ("error", self.error.is_some()),
+            ("tags.denialReason", self.denial_reason.is_some()),
+        ];
+        present.into_iter().filter_map(|(field, set)| set.then_some(field)).collect()
+    }
+}
+
+/// Where decision records go. Implemented by the binary over `rustfs-audit`.
+pub trait AuthzAuditRecordSink: Send + Sync + 'static {
+    /// Records one decision. Called inline on the request path: must not block or panic.
+    fn record_decision(&self, record: AuthzAuditRecord);
+}
+
+/// An [`AuthzAuditSink`] writing one record per decision.
+pub struct RustfsAuthzAudit<S> {
+    sink: S,
+}
+
+impl<S: AuthzAuditRecordSink> RustfsAuthzAudit<S> {
+    /// An audit sink reporting into `sink`.
+    #[must_use]
+    pub fn new(sink: S) -> Self {
+        Self { sink }
+    }
+}
+
+impl<S: AuthzAuditRecordSink> AuthzAuditSink for RustfsAuthzAudit<S> {
+    fn on_decision(&self, event: &AuthzAuditEvent<'_>) {
+        self.sink.record_decision(audit_record(event));
+    }
+}
+
+/// The record for one decision; [`RustfsAuthzAudit`] calls it for every event.
+#[must_use]
+pub fn audit_record(event: &AuthzAuditEvent<'_>) -> AuthzAuditRecord {
+    let denial_reason = match event.decision {
+        Decision::Allow => None,
+        Decision::Deny => Some(DENY_REASON),
+        Decision::Indeterminate => Some(INDETERMINATE_REASON),
+    };
+    let refused = denial_reason.is_some();
+    AuthzAuditRecord {
+        version: AUDIT_VERSION,
+        event: EventName::AuditS3Operation,
+        trigger: event.action.to_owned(),
+        api_name: event.action.to_owned(),
+        api_bucket: event.bucket.map(|bucket| bucket.as_str().to_owned()),
+        api_object: event.key.map(|key| key.as_str().to_owned()),
+        api_status: refused.then_some("failure"),
+        api_status_code: refused.then_some(403),
+        request_id: event.request_id.as_str().to_owned(),
+        access_key: event.identity.map(|identity| identity.access_key_id().to_owned()),
+        error: refused.then_some(ACCESS_DENIED_ERROR),
+        decision: event.decision.as_str(),
+        stage: match event.stage {
+            AuthzStage::Route => "route",
+            AuthzStage::Input => "input",
+        },
+        operation: event.operation.to_owned(),
+        policy_snapshot: event.policy_snapshot.map(|snapshot| snapshot.to_string()),
+        denial_reason,
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use core::time::Duration;
+    use rustfs_gateway::{
+        AuthSchemeRef, AuthzRequest, BucketName, Identity, ObjectKey, PolicySnapshot, RequestId, ResourceShape, SessionBinding,
+        TargetOrigin,
+    };
+    use std::sync::Mutex;
+
+    const ACCESS_KEY: &str = "AKIDCALLER0001";
+    const BUCKET: &str = "audit-bucket";
+    const KEY: &str = "reports/q3.csv";
+
+    #[derive(Default)]
+    struct Recorded(Mutex<Vec<AuthzAuditRecord>>);
+
+    impl AuthzAuditRecordSink for Recorded {
+        fn record_decision(&self, record: AuthzAuditRecord) {
+            self.0.lock().expect("not poisoned").push(record);
+        }
+    }
+
+    struct Fixture {
+        request_id: RequestId,
+        bucket: BucketName,
+        key: ObjectKey,
+        identity: Identity,
+        snapshot: PolicySnapshot,
+    }
+
+    fn fixture() -> Fixture {
+        Fixture {
+            request_id: RequestId::from_bits(0x5EED),
+            bucket: BucketName::new(BUCKET).expect("a valid bucket"),
+            key: ObjectKey::new(KEY).expect("a valid key"),
+            identity: Identity::new(ACCESS_KEY).expect("a valid access key"),
+            snapshot: PolicySnapshot::empty(),
+        }
+    }
+
+    fn event<'a>(f: &'a Fixture, decision: Decision, identity: Option<&'a Identity>) -> AuthzAuditEvent<'a> {
+        AuthzAuditEvent {
+            request_id: &f.request_id,
+            stage: AuthzStage::Route,
+            operation: "GetObject",
+            action: "s3:GetObject",
+            resource: ResourceShape::Object,
+            bucket: Some(&f.bucket),
+            key: Some(&f.key),
+            resources: &[],
+            auth_scheme: if identity.is_some() {
+                AuthSchemeRef::SigV4Header
+            } else {
+                AuthSchemeRef::Anonymous
+            },
+            identity,
+            target_origin: TargetOrigin::Path,
+            policy_snapshot: Some(f.snapshot.id()),
+            decision,
+            elapsed: Duration::ZERO,
+        }
+    }
+
+    fn recorded(decisions: &[Decision]) -> Vec<AuthzAuditRecord> {
+        let f = fixture();
+        let sink = RustfsAuthzAudit::new(Recorded::default());
+        for decision in decisions {
+            sink.on_decision(&event(&f, *decision, Some(&f.identity)));
+        }
+        sink.sink.0.into_inner().expect("not poisoned")
+    }
+
+    /// Positive — every decision is one record, and an `Allow` is recorded too.
+    #[test]
+    fn audit_every_decision_is_one_record() {
+        let records = recorded(&[Decision::Allow, Decision::Deny, Decision::Indeterminate]);
+        let decisions: Vec<&str> = records.iter().map(|record| record.decision).collect();
+        assert_eq!(decisions, ["allow", "deny", "indeterminate"]);
+        let allow = &records[0];
+        assert_eq!(allow.api_status, None);
+        assert_eq!(allow.error, None);
+        assert_eq!(allow.denial_reason, None);
+        assert_eq!(allow.access_key.as_deref(), Some(ACCESS_KEY));
+        assert_eq!(allow.request_id, RequestId::from_bits(0x5EED).as_str());
+    }
+
+    /// Negative — `Indeterminate` is recorded as itself, distinguishable from `Deny` in both the
+    /// decision tag and the denial reason, though both are the same refusal on the wire.
+    #[test]
+    fn audit_indeterminate_is_not_deny() {
+        let records = recorded(&[Decision::Deny, Decision::Indeterminate]);
+        let (deny, indeterminate) = (&records[0], &records[1]);
+        assert_eq!(indeterminate.decision, "indeterminate");
+        assert!(indeterminate.tags().contains(&(DECISION_TAG, "indeterminate".to_owned())));
+        assert_eq!(indeterminate.denial_reason, Some(INDETERMINATE_REASON));
+        assert_eq!(deny.decision, "deny");
+        assert_eq!(deny.denial_reason, Some(DENY_REASON));
+        assert_ne!(deny.tags(), indeterminate.tags());
+        for refusal in [deny, indeterminate] {
+            assert_eq!(refusal.api_status, Some("failure"));
+            assert_eq!(refusal.api_status_code, Some(403));
+            assert_eq!(refusal.error, Some(ACCESS_DENIED_ERROR));
+        }
+    }
+
+    /// Negative — an anonymous caller has no `accessKey`; none is invented, not even an empty one.
+    #[test]
+    fn audit_anonymous_caller_has_no_access_key() {
+        let f = fixture();
+        for decision in [Decision::Allow, Decision::Deny, Decision::Indeterminate] {
+            let record = audit_record(&event(&f, decision, None));
+            assert_eq!(record.access_key, None, "{decision:?}");
+            assert!(!record.contract_fields().contains(&"accessKey"), "{decision:?}");
+        }
+    }
+
+    /// Negative — a policy source that could not answer leaves no snapshot tag rather than a
+    /// placeholder.
+    #[test]
+    fn audit_missing_policy_reading_has_no_snapshot_tag() {
+        let f = fixture();
+        let mut unread = event(&f, Decision::Indeterminate, Some(&f.identity));
+        unread.policy_snapshot = None;
+        let record = audit_record(&unread);
+        assert_eq!(record.policy_snapshot, None);
+        assert!(record.tags().iter().all(|(key, _)| *key != POLICY_SNAPSHOT_TAG));
+        let read = audit_record(&event(&f, Decision::Indeterminate, Some(&f.identity)));
+        assert_eq!(read.policy_snapshot, Some(f.snapshot.id().to_string()));
+    }
+
+    /// Negative — a service-level decision has no bucket or object, and the input stage is named.
+    #[test]
+    fn audit_service_decision_has_no_target() {
+        let f = fixture();
+        let mut service = event(&f, Decision::Deny, Some(&f.identity));
+        service.bucket = None;
+        service.key = None;
+        service.stage = AuthzStage::Input;
+        let record = audit_record(&service);
+        assert_eq!((record.api_bucket.as_deref(), record.api_object.as_deref()), (None, None));
+        assert_eq!(record.stage, "input");
+    }
+
+    /// Negative — the access key is the caller's verified identity, never a default credential,
+    /// including for a session credential.
+    #[test]
+    fn audit_access_key_is_the_callers_identity() {
+        let f = fixture();
+        let binding = SessionBinding::new(crate::STS_SESSION_ISSUER, 4_000_000_000).expect("a binding");
+        let session = Identity::new("ASIATEMPCALLER01")
+            .expect("a key")
+            .with_session_binding(binding);
+        let record = audit_record(&event(&f, Decision::Deny, Some(&session)));
+        assert_eq!(record.access_key.as_deref(), Some("ASIATEMPCALLER01"));
+    }
+
+    /// Negative — secret material planted in every caller-chosen field the legacy entry has no
+    /// field for (a version id, a derived resource, a session binding) reaches nothing the sink
+    /// receives, and the record has no field that could hold a secret.
+    #[test]
+    fn audit_no_secret_reaches_the_sink() {
+        const SECRETS: [&str; 4] = [
+            "wJalrXUtnFEMI/K7MDENG/bPxRfiCYSECRETKEY",
+            "AQoDYXdzEJrSESSIONTOKEN",
+            "fe5f80f77d5fa3beca038a248ff027d0445342fe2855ddc963176630326f1024",
+            "SSECKEYMDEyMzQ1Njc4OTAxMjM0NTY3ODkwMTI=",
+        ];
+        let f = fixture();
+        let binding = SessionBinding::new(SECRETS[1], 4_000_000_000).expect("a binding");
+        let session = Identity::new(ACCESS_KEY).expect("a key").with_session_binding(binding);
+        let derived = AuthzRequest {
+            operation: "GetObject",
+            action: "s3:GetObjectVersion",
+            resource: ResourceShape::Object,
+            bucket: Some(&f.bucket),
+            key: Some(&f.key),
+            copy_source_identity: None,
+            version_id: Some(SECRETS[0]),
+            route_action: SECRETS[2],
+            route_bucket: Some(&f.bucket),
+            route_key: Some(&f.key),
+            identity: Some(&session),
+            target_origin: TargetOrigin::Host,
+            subject: None,
+        };
+        let resources = [derived];
+        let mut planted = event(&f, Decision::Deny, Some(&session));
+        planted.resources = &resources;
+        planted.stage = AuthzStage::Input;
+        let sink = RustfsAuthzAudit::new(Recorded::default());
+        sink.on_decision(&planted);
+        let records = sink.sink.0.into_inner().expect("not poisoned");
+        assert_eq!(records.len(), 1);
+        let everything = format!("{:?} {:?}", records[0], records[0].tags());
+        for secret in SECRETS {
+            assert!(!everything.contains(secret), "{secret} reached the sink: {everything}");
+        }
+        assert!(everything.contains(ACCESS_KEY), "the access key id is the one identity fact recorded");
+    }
+
+    /// Negative — a refusal populates exactly the fields this module claims, so the fixture's
+    /// `gateway-authz-audit` owner is backed by what a record really carries.
+    #[test]
+    fn audit_refusal_populates_the_owned_contract_fields() {
+        let records = recorded(&[Decision::Deny, Decision::Indeterminate]);
+        for record in &records {
+            assert_eq!(record.contract_fields(), OWNED_CONTRACT_FIELDS, "{}", record.decision);
+        }
+    }
+}
@@ -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);
+    }
+}
@@ -0,0 +1,29 @@
+// 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 authz_audit;
+pub mod credentials;
+pub mod observer;
+pub mod policy;
+
+pub use authz_audit::{AuthzAuditRecord, AuthzAuditRecordSink, RustfsAuthzAudit};
+pub use credentials::{IamCredentials, IamKeyLookup, STS_SESSION_ISSUER};
+pub use observer::{RequestMetricsSink, RequestSample, RustfsObserver};
+pub use policy::{IamPolicySource, IamPrincipalLookup, PrincipalKind, RustfsPrincipal};
@@ -0,0 +1,263 @@
+// 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 [`Observer`], feeding RustFS's per-request S3 counter.
+//!
+//! Responsible for: turning the gateway's [`RequestEvent`] into one [`RequestSample`] — the
+//! `method`, `op` and `outcome` labels of `rustfs_s3_http_requests_total`, each bounded to the
+//! value set the legacy counter uses — and handing it to a [`RequestMetricsSink`].
+//! NOT responsible for: anything the gateway cannot know when it reports, which is before the first
+//! body byte. Those contract items stay with RustFS's outer layers: every `rustfs_http_server_*`
+//! series (duration, request and response bytes, body chunks, transport failures),
+//! `rustfs_s3_http_entry_to_dispatch_duration_seconds`, and the `cancelled` and `service_error`
+//! outcomes of the counter this module feeds (a request whose future is dropped, or whose service
+//! fails, is never observed). Handler-entry series (`rustfs_s3_operations_total`, lock failures,
+//! log-chain counters) stay with handler code. Who owns each contract item is recorded in
+//! `tests/fixtures/observability-contract.json`; [`OWNED_CONTRACT_ITEMS`] is this module's half.
+//! Upstream: the gateway service, once per answered request. Downstream: the binary's
+//! [`RequestMetricsSink`], which increments the existing counter.
+//!
+//! # Cardinality
+//!
+//! Every label is a `&'static str` from a fixed set, as in `crates/io-metrics/src/s3_http_metrics.rs`:
+//! a method outside the nine known ones is `OTHER`, an operation that is not one of the
+//! [`S3Operation`] variants is `unknown`, a status outside `100..=599` is `unknown`. Nothing a caller
+//! sends can mint a new series. The caller's identity and the S3 error code are deliberately not
+//! labels: the legacy counter has neither, and an access key is unbounded.
+//!
+//! # Sink contract
+//!
+//! [`Observer::on_response`] is synchronous and must neither block nor panic, so the sink is called
+//! inline with a `Copy` value; an implementation increments an atomic or a cached counter handle and
+//! returns.
+
+use rustfs_gateway::{ErrorCode, Observer, RequestEvent};
+use rustfs_s3_ops::S3Operation;
+
+/// The `op` label of a request routing named no [`S3Operation`] for, as the legacy counter spells it.
+pub const UNKNOWN_OPERATION: &str = "unknown";
+
+/// The method labels the legacy counter keeps apart; anything else is the last one, `OTHER`.
+const METHODS: [&str; 10] = [
+    "GET", "PUT", "POST", "DELETE", "HEAD", "OPTIONS", "PATCH", "CONNECT", "TRACE", "OTHER",
+];
+
+/// The status-class outcome labels; `cancelled` and `service_error` belong to the outer layer.
+const OUTCOMES: [&str; 6] = ["1xx", "2xx", "3xx", "4xx", "5xx", "unknown"];
+
+/// The contract items (ids in `tests/fixtures/observability-contract.json`) this module supplies.
+pub const OWNED_CONTRACT_ITEMS: [&str; 4] = [
+    "rustfs_s3_http_requests_total",
+    "rustfs_s3_http_requests_total{method}",
+    "rustfs_s3_http_requests_total{op}",
+    "rustfs_s3_http_requests_total{outcome}",
+];
+
+/// One answered request, as the labels of `rustfs_s3_http_requests_total{method, op, outcome}`.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct RequestSample {
+    /// The `method` label: one of the nine known methods, or `OTHER`.
+    pub method: &'static str,
+    /// The `op` label: the operation's legacy spelling (`s3:GetObject`), or [`UNKNOWN_OPERATION`].
+    pub op: &'static str,
+    /// The `outcome` label: the response status class, or `unknown` for a status outside `100..=599`.
+    pub outcome: &'static str,
+    /// The operation behind [`Self::op`], for a sink that indexes counters by it.
+    pub operation: Option<S3Operation>,
+    /// Whether the request succeeded: a `2xx` head that carried no S3 error. A committed answer whose
+    /// head was `200` but whose body ended in an error document is not a success.
+    pub succeeded: bool,
+}
+
+/// Where answered-request samples go. Implemented by the binary over `rustfs-io-metrics`.
+pub trait RequestMetricsSink: Send + Sync + 'static {
+    /// Records one answered request. Called inline on the response path: must not block or panic.
+    fn record_answered(&self, sample: RequestSample);
+}
+
+/// An [`Observer`] that maps every answered request onto RustFS's S3 request counter.
+pub struct RustfsObserver<S> {
+    sink: S,
+}
+
+impl<S: RequestMetricsSink> RustfsObserver<S> {
+    /// An observer reporting into `sink`.
+    #[must_use]
+    pub fn new(sink: S) -> Self {
+        Self { sink }
+    }
+}
+
+impl<S: RequestMetricsSink> RustfsObserver<S> {
+    /// Reports one answered request. [`Observer::on_response`] calls it with the event's fields; it
+    /// is separate because a [`RequestEvent`] can only be built by the gateway.
+    fn observe(&self, method: &str, operation: Option<&str>, status: u16, error: Option<&ErrorCode>) {
+        self.sink.record_answered(request_sample(method, operation, status, error));
+    }
+}
+
+impl<S: RequestMetricsSink> Observer for RustfsObserver<S> {
+    fn on_response(&self, event: &RequestEvent<'_>) {
+        self.observe(event.method().as_str(), event.operation, event.status, event.error);
+    }
+}
+
+/// The sample for one answered request; [`RustfsObserver`] calls it with the event's fields.
+#[must_use]
+pub fn request_sample(method: &str, operation: Option<&str>, status: u16, error: Option<&ErrorCode>) -> RequestSample {
+    let operation = operation.and_then(s3_operation);
+    let outcome = match status {
+        100..=599 => OUTCOMES[usize::from(status / 100 - 1)],
+        _ => OUTCOMES[OUTCOMES.len() - 1],
+    };
+    RequestSample {
+        method: METHODS
+            .iter()
+            .copied()
+            .find(|known| *known == method)
+            .unwrap_or(METHODS[METHODS.len() - 1]),
+        op: operation.map_or(UNKNOWN_OPERATION, S3Operation::as_str),
+        outcome,
+        operation,
+        succeeded: (200..300).contains(&status) && error.is_none(),
+    }
+}
+
+/// The [`S3Operation`] whose legacy spelling is `s3:<name>`, for the gateway's `Operation::NAME`.
+fn s3_operation(name: &str) -> Option<S3Operation> {
+    S3Operation::ALL
+        .iter()
+        .copied()
+        .find(|op| op.as_str().strip_prefix("s3:") == Some(name))
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use std::sync::Mutex;
+
+    #[derive(Default)]
+    struct Recorded(Mutex<Vec<RequestSample>>);
+
+    impl RequestMetricsSink for Recorded {
+        fn record_answered(&self, sample: RequestSample) {
+            self.0.lock().expect("not poisoned").push(sample);
+        }
+    }
+
+    fn labels(sample: &RequestSample) -> [&'static str; 3] {
+        [sample.method, sample.op, sample.outcome]
+    }
+
+    /// Positive — an answered read is one sample with the legacy label spellings.
+    #[test]
+    fn observer_success_maps_to_legacy_labels() {
+        let sample = request_sample("GET", Some("GetObject"), 200, None);
+        assert_eq!(labels(&sample), ["GET", "s3:GetObject", "2xx"]);
+        assert_eq!(sample.operation, Some(S3Operation::GetObject));
+        assert!(sample.succeeded);
+    }
+
+    /// Negative — an error response is counted, under its own status class, as a failure.
+    #[test]
+    fn observer_error_response_is_counted_under_its_status_class() {
+        let observer = RustfsObserver::new(Recorded::default());
+        observer.observe("GET", Some("GetObject"), 404, Some(&ErrorCode::NO_SUCH_KEY));
+        observer.observe("PUT", Some("PutObject"), 403, Some(&ErrorCode::ACCESS_DENIED));
+        let recorded = observer.sink.0.lock().expect("not poisoned");
+        assert_eq!(recorded.len(), 2);
+        assert_eq!(labels(&recorded[0]), ["GET", "s3:GetObject", "4xx"]);
+        assert_eq!(labels(&recorded[1]), ["PUT", "s3:PutObject", "4xx"]);
+        assert!(recorded.iter().all(|sample| !sample.succeeded));
+    }
+
+    /// Negative — a 5xx is a `5xx` failure, never a success.
+    #[test]
+    fn observer_server_error_is_not_success() {
+        for status in [500, 503, 599] {
+            let sample = request_sample("GET", Some("GetObject"), status, Some(&ErrorCode::INTERNAL_ERROR));
+            assert_eq!(sample.outcome, "5xx", "{status}");
+            assert!(!sample.succeeded, "{status}");
+        }
+        assert!(!request_sample("GET", Some("GetObject"), 503, None).succeeded);
+    }
+
+    /// Negative — a committed `200` whose body ended in an error is not a success, though its
+    /// outcome label stays the head's class as the legacy layer reads it.
+    #[test]
+    fn observer_committed_error_after_200_is_not_success() {
+        let sample = request_sample("POST", Some("CompleteMultipartUpload"), 200, Some(&ErrorCode::INTERNAL_ERROR));
+        assert_eq!(sample.outcome, "2xx");
+        assert!(!sample.succeeded);
+    }
+
+    /// Negative — operation names outside the bounded set, including caller-shaped text and the
+    /// unrouted case, all collapse to `unknown`; no sample can carry a label the set lacks.
+    #[test]
+    fn observer_unknown_operation_is_bounded() {
+        let unbounded = [
+            None,
+            Some(""),
+            Some("NoSuchOperation"),
+            Some("s3:GetObject"),
+            Some("getobject"),
+            Some("GetObject "),
+            Some("GetObject?X-Amz-Signature=deadbeef"),
+        ];
+        for operation in unbounded {
+            let sample = request_sample("GET", operation, 200, None);
+            assert_eq!(sample.op, UNKNOWN_OPERATION, "{operation:?}");
+            assert_eq!(sample.operation, None, "{operation:?}");
+        }
+        let known: Vec<&str> = S3Operation::ALL.iter().map(|op| op.as_str()).collect();
+        for op in S3Operation::ALL {
+            let name = op.as_str().trim_start_matches("s3:");
+            let sample = request_sample("GET", Some(name), 200, None);
+            assert!(known.contains(&sample.op), "{name}");
+            assert_eq!(sample.operation, Some(op));
+        }
+    }
+
+    /// Negative — unknown and lower-case methods fold into `OTHER`, as the legacy counter does.
+    #[test]
+    fn observer_unknown_method_is_bounded() {
+        for method in ["PROPFIND", "get", "", "X-SECRET-METHOD"] {
+            assert_eq!(request_sample(method, Some("GetObject"), 200, None).method, "OTHER", "{method}");
+        }
+        for method in &METHODS[..9] {
+            assert_eq!(request_sample(method, None, 200, None).method, *method);
+        }
+    }
+
+    /// Negative — a status outside the HTTP range is `unknown`, not a class that does not exist.
+    #[test]
+    fn observer_out_of_range_status_is_unknown() {
+        for status in [0, 99, 600, u16::MAX] {
+            let sample = request_sample("GET", Some("GetObject"), status, None);
+            assert_eq!(sample.outcome, "unknown", "{status}");
+            assert!(!sample.succeeded, "{status}");
+        }
+        assert_eq!(request_sample("GET", None, 100, None).outcome, "1xx");
+        assert_eq!(request_sample("GET", None, 304, None).outcome, "3xx");
+    }
+
+    /// Negative — the S3 error code, however a handler spelled it, never reaches the sample.
+    #[test]
+    fn observer_error_code_is_not_a_label() {
+        let custom = ErrorCode::custom("SecretSessionTokenAQoDYXdzEJr", http::StatusCode::BAD_REQUEST);
+        let rendered = format!("{:?}", request_sample("GET", Some("GetObject"), 400, Some(&custom)));
+        assert!(!rendered.contains("SecretSessionTokenAQoDYXdzEJr"), "{rendered}");
+    }
+}
@@ -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));
+    }
+}
@@ -0,0 +1,100 @@
+{
+  "description": "Names (not values) of the per-request metrics, their labels, and the audit entry fields the legacy S3 path emits, each with the RustFS file:line that emits it and the owner that supplies it once rustfs/gateway answers the request. Extracted from code, not from a running server. `literal` must appear on the cited line; tests/observability_contract.rs checks it.",
+  "owners": {
+    "gateway-observer": "crates/gateway-adapter/src/observer.rs, from the gateway's RequestEvent (operation, method, status) before the first body byte",
+    "gateway-authz-audit": "crates/gateway-adapter/src/authz_audit.rs, from the gateway's AuthzAuditEvent (one record per decision)",
+    "outer-layer": "RustFS's own tower layers around the gateway service (rustfs/src/server/http.rs, rustfs/src/server/layer.rs): duration, bytes, cancellation, peer and request-line facts",
+    "handler": "RustFS handler code the gateway dispatches to (OperationHelper and the record_s3_op call sites), unchanged by the gateway",
+    "binary-sink": "the rustfs binary's implementation of this crate's sink traits: facts neither gateway hook carries, filled while turning a record into an AuditEntry"
+  },
+  "items": [
+    {"id": "rustfs_s3_http_requests_total", "kind": "metric", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:38", "literal": "\"rustfs_s3_http_requests_total\"", "owner": "gateway-observer", "note": "answered requests; a request whose future is dropped or whose service errors is never observed, so the outcome values `cancelled` and `service_error` stay with outer-layer"},
+    {"id": "rustfs_s3_http_requests_total{method}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:183", "literal": "\"method\" =>", "owner": "gateway-observer"},
+    {"id": "rustfs_s3_http_requests_total{op}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:183", "literal": "\"op\" =>", "owner": "gateway-observer"},
+    {"id": "rustfs_s3_http_requests_total{outcome}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:183", "literal": "\"outcome\" =>", "owner": "gateway-observer", "note": "status-class values; `cancelled` and `service_error` are outer-layer"},
+    {"id": "rustfs_s3_http_entry_to_dispatch_duration_seconds", "kind": "metric", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:40", "literal": "\"rustfs_s3_http_entry_to_dispatch_duration_seconds\"", "owner": "outer-layer", "note": "a duration"},
+    {"id": "rustfs_s3_http_entry_to_dispatch_duration_seconds{op}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:336", "literal": "\"op\" =>", "owner": "outer-layer"},
+    {"id": "rustfs_s3_http_entry_to_dispatch_duration_seconds{outcome}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:336", "literal": "\"outcome\" =>", "owner": "outer-layer"},
+    {"id": "rustfs_s3_operations_total", "kind": "metric", "emitted_at": "crates/io-metrics/src/s3_api_metrics.rs:19", "literal": "\"rustfs_s3_operations_total\"", "owner": "handler", "note": "counts handler entries, not requests: a request refused before its handler is not an operation"},
+    {"id": "rustfs_s3_operations_total{op}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_api_metrics.rs:50", "literal": "\"op\" =>", "owner": "handler"},
+    {"id": "rustfs_s3_lock_failures_total", "kind": "metric", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:39", "literal": "\"rustfs_s3_lock_failures_total\"", "owner": "handler"},
+    {"id": "rustfs_s3_lock_failures_total{op}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:234", "literal": "\"op\" =>", "owner": "handler"},
+    {"id": "rustfs_s3_lock_failures_total{resource_class}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:235", "literal": "\"resource_class\" =>", "owner": "handler"},
+    {"id": "rustfs_s3_lock_failures_total{mode}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:236", "literal": "\"mode\" =>", "owner": "handler"},
+    {"id": "rustfs_s3_lock_failures_total{reason}", "kind": "label", "emitted_at": "crates/io-metrics/src/s3_http_metrics.rs:237", "literal": "\"reason\" =>", "owner": "handler"},
+    {"id": "rustfs_log_chain_audit_total", "kind": "metric", "emitted_at": "rustfs/src/storage/helper.rs:350", "literal": "\"rustfs_log_chain_audit_total\"", "owner": "handler"},
+    {"id": "rustfs_log_chain_orphan_total", "kind": "metric", "emitted_at": "rustfs/src/storage/helper.rs:372", "literal": "\"rustfs_log_chain_orphan_total\"", "owner": "handler"},
+    {"id": "rustfs_log_chain_orphan_total{component}", "kind": "label", "emitted_at": "rustfs/src/storage/helper.rs:372", "literal": "\"component\" =>", "owner": "handler"},
+    {"id": "rustfs_log_chain_op_event_mismatch_total", "kind": "metric", "emitted_at": "rustfs/src/storage/helper.rs:299", "literal": "\"rustfs_log_chain_op_event_mismatch_total\"", "owner": "handler"},
+    {"id": "rustfs_log_chain_op_event_mismatch_total{op}", "kind": "label", "emitted_at": "rustfs/src/storage/helper.rs:300", "literal": "\"op\" =>", "owner": "handler"},
+    {"id": "rustfs_log_chain_op_event_mismatch_total{event}", "kind": "label", "emitted_at": "rustfs/src/storage/helper.rs:301", "literal": "\"event\" =>", "owner": "handler"},
+    {"id": "rustfs_http_server_requests_total", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:133", "literal": "\"rustfs_http_server_requests_total\"", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_requests_total{method}", "kind": "label", "emitted_at": "rustfs/src/server/http.rs:288", "literal": "METRIC_HTTP_SERVER_REQUESTS_TOTAL, LABEL_HTTP_METHOD", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_request_body_size_bytes", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:138", "literal": "\"rustfs_http_server_request_body_size_bytes\"", "owner": "outer-layer", "note": "bytes"},
+    {"id": "rustfs_http_server_request_body_size_bytes{method}", "kind": "label", "emitted_at": "rustfs/src/server/http.rs:289", "literal": "METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES, LABEL_HTTP_METHOD", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_request_duration_seconds", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:136", "literal": "\"rustfs_http_server_request_duration_seconds\"", "owner": "outer-layer", "note": "a duration"},
+    {"id": "rustfs_http_server_request_duration_seconds{status_class}", "kind": "label", "emitted_at": "rustfs/src/server/http.rs:303", "literal": "METRIC_HTTP_SERVER_REQUEST_DURATION_SECONDS, LABEL_HTTP_STATUS_CLASS", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_failures_total", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:134", "literal": "\"rustfs_http_server_failures_total\"", "owner": "outer-layer", "note": "also counts transport failures after the head, which no hook sees"},
+    {"id": "rustfs_http_server_failures_total{status_class}", "kind": "label", "emitted_at": "rustfs/src/server/http.rs:304", "literal": "METRIC_HTTP_SERVER_FAILURES_TOTAL, LABEL_HTTP_STATUS_CLASS", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_response_body_size_bytes", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:140", "literal": "\"rustfs_http_server_response_body_size_bytes\"", "owner": "outer-layer", "note": "bytes"},
+    {"id": "rustfs_http_server_response_body_size_bytes{status_class}", "kind": "label", "emitted_at": "rustfs/src/server/http.rs:305", "literal": "METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES, LABEL_HTTP_STATUS_CLASS", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_active_requests", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:135", "literal": "\"rustfs_http_server_active_requests\"", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_request_body_bytes_total", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:137", "literal": "\"rustfs_http_server_request_body_bytes_total\"", "owner": "outer-layer", "note": "bytes"},
+    {"id": "rustfs_http_server_response_body_bytes_total", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:139", "literal": "\"rustfs_http_server_response_body_bytes_total\"", "owner": "outer-layer", "note": "bytes, after the first byte"},
+    {"id": "rustfs_http_server_response_body_chunk_size_bytes", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:141", "literal": "\"rustfs_http_server_response_body_chunk_size_bytes\"", "owner": "outer-layer", "note": "after the first byte"},
+    {"id": "rustfs_http_server_response_body_chunk_latency_seconds", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:142", "literal": "\"rustfs_http_server_response_body_chunk_latency_seconds\"", "owner": "outer-layer", "note": "after the first byte"},
+    {"id": "rustfs_http_server_response_body_stream_duration_seconds", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:143", "literal": "\"rustfs_http_server_response_body_stream_duration_seconds\"", "owner": "outer-layer", "note": "after the first byte"},
+    {"id": "rustfs_http_server_early_response_body_drains_total", "kind": "metric", "emitted_at": "rustfs/src/server/http.rs:145", "literal": "\"rustfs_http_server_early_response_body_drains_total\"", "owner": "outer-layer"},
+    {"id": "rustfs_http_server_early_response_body_drains_total{result}", "kind": "label", "emitted_at": "rustfs/src/server/http.rs:1336", "literal": "LABEL_HTTP_BODY_DRAIN_RESULT =>", "owner": "outer-layer"},
+
+    {"id": "operation.version", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:148", "literal": "AuditEntryBuilder::new(\"1.0\"", "owner": "handler"},
+    {"id": "operation.time", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:148", "literal": "AuditEntryBuilder::new(", "owner": "handler"},
+    {"id": "operation.event", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:148", "literal": "event, trigger", "owner": "handler"},
+    {"id": "operation.trigger", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:148", "literal": "trigger, api", "owner": "handler"},
+    {"id": "operation.remotehost", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:149", "literal": ".remote_host(", "owner": "handler"},
+    {"id": "operation.userAgent", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:150", "literal": ".user_agent(", "owner": "handler"},
+    {"id": "operation.requestHost", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:151", "literal": ".req_host(", "owner": "handler"},
+    {"id": "operation.requestPath", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:152", "literal": ".req_path(", "owner": "handler"},
+    {"id": "operation.requestQuery", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:153", "literal": ".req_query(", "owner": "handler"},
+    {"id": "operation.requestID", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:154", "literal": ".request_id(", "owner": "handler"},
+    {"id": "operation.accessKey", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:159", "literal": ".access_key(", "owner": "handler"},
+    {"id": "operation.parentUser", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:161", "literal": ".parent_user(", "owner": "handler"},
+    {"id": "operation.requestClaims", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:173", "literal": ".req_claims(", "owner": "handler"},
+    {"id": "operation.api.name", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:361", "literal": ".name(trigger)", "owner": "handler"},
+    {"id": "operation.api.bucket", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:363", "literal": ".bucket(", "owner": "handler"},
+    {"id": "operation.api.object", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:366", "literal": ".object(", "owner": "handler"},
+    {"id": "operation.api.objects", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:464", "literal": ".objects(", "owner": "handler"},
+    {"id": "operation.api.status", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:520", "literal": ".status(", "owner": "handler"},
+    {"id": "operation.api.status_code", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:521", "literal": ".status_code(", "owner": "handler"},
+    {"id": "operation.api.time_to_response", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:522", "literal": ".time_to_response(", "owner": "handler", "note": "a duration measured inside the handler"},
+    {"id": "operation.api.timeToResponseInNS", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:523", "literal": ".time_to_response_in_ns(", "owner": "handler"},
+    {"id": "operation.responseHeader", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:528", "literal": ".resp_header(", "owner": "handler"},
+    {"id": "operation.error", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:531", "literal": ".error(", "owner": "handler"},
+    {"id": "operation.tags.traceId", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:537", "literal": "\"traceId\"", "owner": "handler"},
+    {"id": "operation.tags.spanId", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:540", "literal": "\"spanId\"", "owner": "handler"},
+    {"id": "operation.tags.kms", "kind": "audit_field", "entry": "operation", "emitted_at": "rustfs/src/storage/helper.rs:546", "literal": "audit_tags()", "owner": "handler", "note": "the KMS scope's own tag keys"},
+
+    {"id": "authorization_denial.version", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:148", "literal": "AuditEntryBuilder::new(\"1.0\"", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.time", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:148", "literal": "AuditEntryBuilder::new(", "owner": "binary-sink", "note": "AuditEntryBuilder::new stamps it"},
+    {"id": "authorization_denial.event", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:643", "literal": "EventName::AuditS3Operation", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.trigger", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:644", "literal": "action_name", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.api.name", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:624", "literal": ".name(action_name)", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.api.status", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:625", "literal": ".status(\"failure\")", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.api.status_code", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:626", "literal": ".status_code(StatusCode::FORBIDDEN", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.api.bucket", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:628", "literal": ".bucket(bucket)", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.api.object", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:631", "literal": ".object(object)", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.requestID", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:154", "literal": ".request_id(", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.accessKey", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:159", "literal": ".access_key(", "owner": "gateway-authz-audit", "note": "the verified caller from the gateway's identity; absent for an anonymous caller"},
+    {"id": "authorization_denial.error", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:648", "literal": ".error(\"Access Denied\")", "owner": "gateway-authz-audit"},
+    {"id": "authorization_denial.tags.denialReason", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:636", "literal": "\"denialReason\"", "owner": "gateway-authz-audit", "note": "coarser values: the gateway event carries the decision, not RustFS's internal reason"},
+    {"id": "authorization_denial.tags.deniedKeys", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:638", "literal": "\"deniedKeys\"", "owner": "handler", "note": "DeleteObjects per-key verdicts are aggregated where the per-key answers are built"},
+    {"id": "authorization_denial.tags.suppressedDenials", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/denial_audit.rs:199", "literal": "SUPPRESSED_DENIALS_TAG", "owner": "binary-sink", "note": "the binary's existing denial rate limiter"},
+    {"id": "authorization_denial.remotehost", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:149", "literal": ".remote_host(", "owner": "outer-layer", "note": "no hook carries peer or header facts, by design; the binary sink reads them from the request-scoped context the outer layer installs"},
+    {"id": "authorization_denial.userAgent", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:150", "literal": ".user_agent(", "owner": "outer-layer"},
+    {"id": "authorization_denial.requestHost", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:151", "literal": ".req_host(", "owner": "outer-layer"},
+    {"id": "authorization_denial.requestPath", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:152", "literal": ".req_path(", "owner": "outer-layer"},
+    {"id": "authorization_denial.requestQuery", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:153", "literal": ".req_query(", "owner": "outer-layer"},
+    {"id": "authorization_denial.parentUser", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:161", "literal": ".parent_user(", "owner": "binary-sink", "note": "an IAM fact; the audit event carries only the snapshot id, so the binary sink looks it up by accessKey"},
+    {"id": "authorization_denial.requestClaims", "kind": "audit_field", "entry": "authorization_denial", "emitted_at": "rustfs/src/storage/helper.rs:173", "literal": ".req_claims(", "owner": "binary-sink", "note": "as parentUser"}
+  ]
+}
@@ -0,0 +1,140 @@
+// 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 legacy observability contract and who owns each item of it under the gateway.
+//!
+//! Responsible for: that `tests/fixtures/observability-contract.json` still names what the legacy
+//! code emits (every item's literal is on the line it cites), that every item has exactly one known
+//! owner, and that the items the fixture gives to this crate's two modules are exactly the ones those
+//! modules declare.
+//! NOT responsible for: values. The contract is one of names.
+//! Upstream: the fixture and the RustFS sources it cites. Downstream: nothing.
+
+#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+
+use std::collections::{BTreeSet, HashSet};
+use std::path::{Path, PathBuf};
+
+use rustfs_gateway_adapter::{authz_audit, observer};
+use serde_json::Value;
+
+const OBSERVER: &str = "gateway-observer";
+const AUTHZ_AUDIT: &str = "gateway-authz-audit";
+
+fn workspace_root() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .ancestors()
+        .nth(2)
+        .expect("the crate lives at crates/<name>")
+        .to_path_buf()
+}
+
+fn contract() -> Value {
+    let text = include_str!("fixtures/observability-contract.json");
+    serde_json::from_str(text).expect("the contract is JSON")
+}
+
+fn items(contract: &Value) -> &Vec<Value> {
+    contract["items"].as_array().expect("an item list")
+}
+
+fn field<'a>(item: &'a Value, name: &str) -> &'a str {
+    item[name]
+        .as_str()
+        .unwrap_or_else(|| panic!("item {item} has no string `{name}`"))
+}
+
+fn owned_by<'a>(contract: &'a Value, owner: &str) -> BTreeSet<&'a str> {
+    items(contract)
+        .iter()
+        .filter(|item| field(item, "owner") == owner)
+        .map(|item| field(item, "id"))
+        .collect()
+}
+
+/// Negative — an item without a known owner, or with a duplicate id, fails: nothing in the contract
+/// may be left unowned or owned twice.
+#[test]
+fn contract_every_item_has_one_known_owner() {
+    let contract = contract();
+    let owners: HashSet<&str> = contract["owners"]
+        .as_object()
+        .expect("an owner vocabulary")
+        .keys()
+        .map(String::as_str)
+        .collect();
+    assert!(owners.contains(OBSERVER) && owners.contains(AUTHZ_AUDIT), "{owners:?}");
+    let mut ids = HashSet::new();
+    for item in items(&contract) {
+        let id = field(item, "id");
+        assert!(ids.insert(id), "duplicate contract item {id}");
+        assert!(owners.contains(field(item, "owner")), "{id} has an unknown owner");
+        assert!(
+            matches!(field(item, "kind"), "metric" | "label" | "audit_field"),
+            "{id} has an unknown kind"
+        );
+    }
+    let metrics = items(&contract).iter().filter(|item| field(item, "kind") == "metric").count();
+    let labels = items(&contract).iter().filter(|item| field(item, "kind") == "label").count();
+    let fields = items(&contract)
+        .iter()
+        .filter(|item| field(item, "kind") == "audit_field")
+        .count();
+    assert_eq!(
+        (metrics, labels, fields),
+        (19, 19, 48),
+        "the contract changed size; update the PR's counts"
+    );
+}
+
+/// Negative — a contract item whose cited line no longer carries its literal is stale, which is how a
+/// silent rename on the legacy side would show up here.
+#[test]
+fn contract_every_item_is_where_it_says() {
+    let root = workspace_root();
+    let contract = contract();
+    for item in items(&contract) {
+        let id = field(item, "id");
+        let (path, line) = field(item, "emitted_at").rsplit_once(':').expect("path:line");
+        let line: usize = line.parse().expect("a line number");
+        let text = std::fs::read_to_string(root.join(path)).unwrap_or_else(|error| panic!("{id}: {path}: {error}"));
+        let cited = text
+            .lines()
+            .nth(line - 1)
+            .unwrap_or_else(|| panic!("{id}: {path} has no line {line}"));
+        let literal = field(item, "literal");
+        assert!(cited.contains(literal), "{id}: {path}:{line} no longer contains {literal:?}: {cited}");
+    }
+}
+
+/// Negative — the fixture's observer-owned items are exactly what `observer` declares it supplies.
+#[test]
+fn contract_observer_items_match_the_module() {
+    let contract = contract();
+    let declared: BTreeSet<&str> = observer::OWNED_CONTRACT_ITEMS.into_iter().collect();
+    assert_eq!(owned_by(&contract, OBSERVER), declared);
+}
+
+/// Negative — the fixture's audit-owned fields are exactly what `authz_audit` declares, and every one
+/// is a field of the legacy authorization-denial entry.
+#[test]
+fn contract_authz_audit_fields_match_the_module() {
+    let contract = contract();
+    let declared: BTreeSet<String> = authz_audit::OWNED_CONTRACT_FIELDS
+        .into_iter()
+        .map(|field| format!("authorization_denial.{field}"))
+        .collect();
+    let owned: BTreeSet<String> = owned_by(&contract, AUTHZ_AUDIT).into_iter().map(str::to_owned).collect();
+    assert_eq!(owned, declared);
+}
Author	SHA1	Message	Date
overtrue	ee741e47b0	Merge remote-tracking branch 'origin/gateway/integration' into gw/T2.4-observer-audit	2026-11-11 11:35:55 +08:00
overtrue	6e616ef7de	feat(gateway-adapter): add Observer and AuthzAuditSink adapters Map the gateway's answered-request event onto the labels of rustfs_s3_http_requests_total, bounded as the legacy counter bounds them, and write one audit record per authorization decision (allow, deny and indeterminate) in the shape of the legacy authorization-denial entry, with the gateway's verified caller as accessKey. Both adapters stop at RustFS-side sink traits. The legacy metric and audit field names, with the file:line that emits each and the owner that supplies it under the gateway, are committed as tests/fixtures/observability-contract.json and held to the code.	2026-11-11 11:35:55 +08:00
overtrue	8fe8bc688a	Merge remote-tracking branch 'origin/gateway/integration' into gw/T2.4-observer-audit	2026-11-11 10:00:11 +08:00
overtrue	dde2169c83	feat(gateway-adapter): add IAM-backed PolicySource 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.	2026-11-11 09:43:47 +08:00
overtrue	7946b094f9	feat(gateway-adapter): add IAM-backed CredentialProvider Add the ring-2 crate rustfs-gateway-adapter with IamCredentials, the gateway CredentialProvider answered by RustFS IAM (rustfs/backlog#2760, credentials half; PolicySource is deferred). - Pin rustfs-gateway at the existing gateway rev with default-features = false so the server crate stays out of the build. - IamKeyLookup is a narrow IAM trait implemented for IamSys so the adapter is testable without an in-memory IAM backend. - Root credentials are injected at construction and resolve before IAM is ready. - Mapping: keystone: prefix -> NotFound (deliberate change from the legacy empty secret); IAM not ready or not initialized -> Unavailable; unknown key or NoSuch{User,ServiceAccount,TempAccount,Account} -> NotFound; other IAM errors -> Backend; invalid or "off" principals resolve disabled; STS credentials carry a SessionBinding, and an STS credential without an expiry is a Backend fault rather than a never-ending session. - No authorization in lookup; no Debug on secret-holding types; IAM error text is dropped after classification.	2026-11-11 07:19:17 +08:00