test(gateway): assert the store-fault answer names nothing

The 500 answer to a credential-store fault is checked to carry neither the looked-up key, nor a secret, nor a reason for the failed lookup. The test comment records that the earlier pin answered 403.

.config/make/lint-fmt.mak

@@ -70,6 +70,11 @@ s3s-footprint-check: ## Check the s3s dependency footprint ratchet stays frozen
 	@echo "📦 Checking s3s footprint ratchet..."
 	./scripts/check_s3s_footprint.sh
 
+.PHONY: gateway-stack-release-boundary-check
+gateway-stack-release-boundary-check: ## Check no release build can enable the gateway-stack feature
+	@echo "🚧 Checking gateway-stack release boundary..."
+	./scripts/check_gateway_stack_release_boundary.sh
+
 .PHONY: test-count-ratchet-check
 test-count-ratchet-check: ## Check no workspace crate fell below its test floor or rose above its #[ignore] ceiling in the committed baseline
 	@echo "🧪 Checking per-crate test-count ratchet..."

.config/make/pre-commit.mak

@@ -20,11 +20,11 @@ planning-docs-check: ## Check that no planning-type documents are committed
 	./scripts/check_no_planning_docs.sh
 
 .PHONY: pre-commit
-pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check test-count-ratchet-check test-seams-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
+pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check gateway-stack-release-boundary-check test-count-ratchet-check test-seams-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
 	@echo "✅ All pre-commit checks passed!"
 
 .PHONY: pre-pr
-pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check test-count-ratchet-check test-seams-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check log-analyzer-rules-check offline-enrollment-e2e-check clippy-check test ## Run full pre-PR checks with clippy and tests
+pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check gateway-stack-release-boundary-check test-count-ratchet-check test-seams-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check log-analyzer-rules-check offline-enrollment-e2e-check clippy-check test ## Run full pre-PR checks with clippy and tests
 	@echo "✅ All pre-PR checks passed!"
 
 .PHONY: dev-check

.config/make/tests.mak

@@ -61,6 +61,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_check_doc_paths.py
 	./scripts/check_embedded_secrets.sh --self-test
 	./scripts/check_s3s_footprint.sh --self-test
+	./scripts/check_gateway_stack_release_boundary.sh --self-test
 	./scripts/test_check_test_count_ratchet.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/codemods/s3_error_to_s3_types.py --self-test

.github/actions/quick-checks/action.yml

@@ -97,6 +97,10 @@ runs:
       shell: bash
       run: ./scripts/check_s3s_footprint.sh
 
+    - name: Check gateway-stack release boundary
+      shell: bash
+      run: ./scripts/check_gateway_stack_release_boundary.sh
+
     - name: Check per-crate test-count ratchet
       shell: bash
       run: ./scripts/check_test_count_ratchet.sh

.github/workflows/ci.yml

@@ -115,7 +115,7 @@ jobs:
             printf '%s\n' 'mode=full' >> "$GITHUB_OUTPUT"
             echo "Base CI policy unavailable; running the full matrix."
           fi
-          if ! grep -q '^plan=' "$GITHUB_OUTPUT" || ! grep -q '^plan_contract=3$' "$GITHUB_OUTPUT"; then
+          if ! grep -q '^plan=' "$GITHUB_OUTPUT" || ! grep -q '^plan_contract=4$' "$GITHUB_OUTPUT"; then
             printf '%s\n' 'mode=full' 'connect=true' 'full_validation=true' 'rehearse_shards=false' 'packages=' 'plan=' >> "$GITHUB_OUTPUT"
             echo "Base policy does not cover this workflow contract; retaining the full matrix."
           fi
@@ -705,6 +705,53 @@ jobs:
           retention-days: 3
           if-no-files-found: warn
 
+  test-and-lint-gateway-stack:
+    name: Test and Lint (gateway-stack)
+    # The gateway S3 stack (rustfs/backlog#2734) sits behind the rustfs
+    # `gateway-stack` feature, which no other pull-request lane lints or tests.
+    # Unlike the dormant feature lanes above, this one runs for every code
+    # change: each operation slice puts its handler tests, its authorization
+    # census and its end-to-end refusal test behind the feature, and a lane that
+    # ran only on demand would let them rot behind a green pull request.
+    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
+    needs: [ quick-checks, classify-changes ]
+    # Self-hosted for build scale and cache ownership: a workspace-scale Cargo
+    # build that reads the ci-dev cache written on this image.
+    runs-on: sm-standard-4
+    timeout-minutes: 90
+    env:
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Setup Rust environment
+        uses: ./.github/actions/setup
+        with:
+          rust-version: stable
+          cache-shared-key: ci-dev
+          cache-save-if: 'false'
+          install-build-packaging-tools: 'false'
+
+      # A feature that does not compile fails here; nothing below is skipped
+      # into a pass.
+      - name: Run clippy with gateway-stack
+        run: cargo clippy --locked -p rustfs --all-targets --features gateway-stack -- -D warnings
+
+      # The gateway stack's own tests: the assembly, the authorizer table and
+      # its census, the handler template, the migrated use cases, and the
+      # embedded server driven through the whole middleware stack.
+      - name: Run tests with gateway-stack
+        env:
+          # The same bounded link concurrency as the other feature lanes.
+          CARGO_BUILD_JOBS: "2"
+        run: |
+          cargo nextest run --profile ci --locked -p rustfs --features gateway-stack --no-tests fail \
+            --status-level all --final-status-level all \
+            -E 'test(/gateway_stack|gateway_handlers|server::stack::|list_buckets|legacy_stack_is_always_selectable/) | binary(embedded_gateway_stack_test)'
+
   build-rustfs-debug-binary:
     name: Build RustFS Debug Binary
     if: contains(fromJSON('["full", "e2e"]'), needs.classify-changes.outputs.mode) && (github.event_name != 'pull_request' || github.event.action != 'closed')
@@ -1579,6 +1626,7 @@ jobs:
       - fips-provider
       - test-and-lint-rio-v2
       - test-and-lint-protocols
+      - test-and-lint-gateway-stack
       - build-rustfs-debug-binary
       - uring-integration
       - e2e-tests
@@ -1613,6 +1661,7 @@ jobs:
       - core-contract
       - test-and-lint-rio-v2
       - test-and-lint-protocols
+      - test-and-lint-gateway-stack
       - build-rustfs-debug-binary
       - build-rustfs-debug-binary-rio-v2
       - uring-integration

Cargo.lock

@@ -9639,6 +9639,8 @@ dependencies = [
  "rustfs-ecstore",
  "rustfs-extension-schema",
  "rustfs-filemeta",
+ "rustfs-gateway",
+ "rustfs-gateway-adapter",
  "rustfs-gateway-types",
  "rustfs-heal",
  "rustfs-heal-contracts",

docs/testing/ci-gates.md

@@ -54,6 +54,7 @@ The four site-replication state-writer concurrency proofs reserve the available
 | PR, non-doc change | `Typos` | `ci.yml` `typos` | Via aggregate | `typos` |
 | Full selection | `Workspace tests` (serial ILM step) | `ci.yml` `workspace-tests` | Via aggregate | exact `Run ignored ILM integration tests serially` command in the job |
 | Nightly schedule, manual dispatch | `Test and Lint (rio-v2)`, `Test and Lint (swift)`, `Test and Lint (sftp)`, `Test and Lint (http3)` | `ci.yml` `test-and-lint-rio-v2`, `test-and-lint-protocols` | Via aggregate on these events; skipped on PRs, merge groups and main pushes | `cargo nextest run` with the job's `--features` |
+| Full selection, every event | `Test and Lint (gateway-stack)` | `ci.yml` `test-and-lint-gateway-stack` | Via aggregate | `cargo clippy -p rustfs --all-targets --features gateway-stack -- -D warnings`; `cargo nextest run -p rustfs --features gateway-stack` with the job's filter |
 | Nightly schedule, manual dispatch, full selection | `Connect Feature Boundaries` | `ci.yml` `connect-feature-boundaries` | Via aggregate on these events; skipped on PRs, merge groups and main pushes | `cargo test -p rustfs --test connect_registration --features connect-e2e-short-credentials`; `cargo check -p rustfs --release --features connect-e2e-short-credentials` must fail; `scripts/check_offline_enrollment_e2e.sh` |
 | PR, non-doc change | `Build RustFS Debug Binary` | `ci.yml` `build-rustfs-debug-binary` | Via aggregate; prerequisite for black-box jobs | `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks` (binary plus its `rustfs.e2e.json` sidecar) |
 | Nightly schedule, manual dispatch, full selection | `io_uring Integration (real)` | `ci.yml` `uring-integration` | Via aggregate on these events; skipped on PRs, merge groups and main pushes | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |

rustfs/Cargo.toml

@@ -56,6 +56,12 @@ required-features = ["swift"]
 name = "swift_object_integration_test"
 required-features = ["swift"]
 
+# Starts an embedded server on the gateway stack, which only a build with the
+# feature can select.
+[[test]]
+name = "embedded_gateway_stack_test"
+required-features = ["gateway-stack"]
+
 [features]
 default = ["ftps", "webdav", "gcs"]
 fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips", "rustfs-ecstore/fips", "rustfs-protocols/fips", "rustfs-s3-client/fips", "rustfs-targets/fips", "rustfs-utils/fips"]
@@ -69,12 +75,19 @@ license = []
 io-scheduler-debug = []  # Enable debug information in I/O scheduler
 tracing-chunk-debug = []  # Enable per-chunk tracing in data plane (high noise, for debugging only)
 full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp", "pyroscope", "gcs"]
-e2e-test-hooks = ["rustfs-ecstore/e2e-test-hooks", "rustfs-scanner/e2e-test-hooks"]
+# The debug-only E2E binary also carries the gateway S3 stack, so the gateway smoke
+# lane runs the same artifact as the legacy one with RUSTFS_S3_STACK as its only change.
+e2e-test-hooks = ["rustfs-ecstore/e2e-test-hooks", "rustfs-scanner/e2e-test-hooks", "gateway-stack"]
 # Shortens Connect credentials only in debug E2E builds.
 connect-e2e-short-credentials = []
 # Builds the dedicated rustfs-cli-e2e target with a build-time public enrollment root.
 offline-enrollment-e2e-root = []
 rio-v2 = ["rustfs-ecstore/rio-v2"]
+# The rustfs-gateway S3 stack, selected at startup with RUSTFS_S3_STACK=gateway
+# (rustfs/backlog#2734). Under construction: off by default and never enabled in a
+# release build; the stack refuses to compile without debug assertions, and
+# `scripts/check_gateway_stack_release_boundary.sh` keeps it out of release inputs.
+gateway-stack = ["dep:rustfs-gateway", "dep:rustfs-gateway-adapter"]
 pyroscope = ["rustfs-obs/pyroscope", "dep:pyroscope", "dep:object", "dep:symbolic-demangle"]
 # Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`.
 dial9 = ["rustfs-obs/dial9"]
@@ -255,6 +268,8 @@ rustfs-notify = { workspace = true }
 rustfs-obs = { workspace = true }
 rustfs-policy = { workspace = true }
 rustfs-protocols = { workspace = true }
+rustfs-gateway = { workspace = true, optional = true }
+rustfs-gateway-adapter = { workspace = true, optional = true }
 rustfs-gateway-types = { workspace = true, features = ["compat-s3s-rustfs"] }
 rustfs-protos = { workspace = true }
 rustfs-rio = { workspace = true }

rustfs/src/app/bucket_usecase.rs

@@ -19,8 +19,8 @@ use super::storage_api::bucket_usecase::StorageObjectInfo as ObjectInfo;
 #[cfg(test)]
 use super::storage_api::bucket_usecase::access::ReqInfo;
 use super::storage_api::bucket_usecase::access::{
-    TableDataPlaneListAccess, authorize_request, bucket_config_mutation_incarnation, log_list_buckets_iam_implicit_deny,
-    prepare_list_buckets_iam_authorization, prepare_odm_read_generation, req_info_ref,
+    ListBucketsAuthzRequest, TableDataPlaneListAccess, authorize_request, bucket_config_mutation_incarnation,
+    log_list_buckets_iam_implicit_deny, prepare_list_buckets_iam_authorization, prepare_odm_read_generation, req_info_ref,
 };
 #[cfg(test)]
 use super::storage_api::bucket_usecase::bucket::target::BucketTarget;
@@ -47,7 +47,7 @@ use super::storage_api::bucket_usecase::bucket::{
     versioning_sys::BucketVersioningSys,
 };
 use super::storage_api::bucket_usecase::contract::bucket::{
-    BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions,
+    BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions,
 };
 use super::storage_api::bucket_usecase::contract::list::{
     ListObjectVersionsInfo as StorageListObjectVersionsInfo, ListObjectsV2Info as StorageListObjectsV2Info, ListOperations as _,
@@ -56,9 +56,9 @@ use super::storage_api::bucket_usecase::error::StorageError;
 use super::storage_api::bucket_usecase::helper::{OperationHelper, spawn_background_with_context};
 use super::storage_api::bucket_usecase::object_utils::to_s3s_xml_etag;
 use super::storage_api::bucket_usecase::s3_api::bucket::{
-    ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output,
-    build_list_objects_output, build_list_objects_v2_output, encode_list_objects_v2_continuation_token,
-    parse_list_object_versions_params, parse_list_objects_v2_params, rustfs_owner,
+    ListObjectVersionsParams, ListObjectsV2Params, build_list_object_versions_output, build_list_objects_output,
+    build_list_objects_v2_output, encode_list_objects_v2_continuation_token, parse_list_object_versions_params,
+    parse_list_objects_v2_params, rustfs_owner,
 };
 use super::storage_api::bucket_usecase::{
     get_validated_store, process_lambda_configurations, process_queue_configurations, process_topic_configurations,
@@ -66,6 +66,7 @@ use super::storage_api::bucket_usecase::{
 };
 #[cfg(feature = "diagnostic-read-version-caller")]
 use super::storage_api::diagnostic;
+use crate::app::RequestEnvelope;
 use crate::app::bucket_list_through as list_through;
 use crate::app::object::shared::mark_on_demand_migration_list_local_only;
 use crate::app::object_data_cache::invalidate_object_data_cache_bucket_after_delete;
@@ -87,12 +88,14 @@ use http::StatusCode;
 use metrics::counter;
 use rustfs_config::RUSTFS_REGION;
 use rustfs_gateway_types::cors_tagging::{PersistedCorsConfiguration, PersistedTagging};
+use rustfs_gateway_types::dto as gateway_dto;
 use rustfs_gateway_types::persistence::{
     PersistedBucketEncryptionConfiguration, PersistedNotificationConfiguration,
     PersistedObjectLockConfiguration as ObjectLockConfiguration, PersistedPublicAccessBlockConfiguration,
     PersistedVersioningConfiguration,
 };
 use rustfs_gateway_types::persistence::{PersistedLifecycleConfiguration, PersistedReplicationConfiguration};
+use rustfs_gateway_types::{BucketName, Timestamp as GatewayTimestamp};
 use rustfs_io_metrics::record_s3_op;
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
 use rustfs_policy::policy::{
@@ -116,15 +119,15 @@ use s3s::dto::{
     GetBucketCorsInput, GetBucketEncryptionInput, GetBucketLifecycleConfigurationInput, GetBucketLocationInput,
     GetBucketLocationOutput, GetBucketNotificationConfigurationInput, GetBucketPolicyInput, GetBucketPolicyOutput,
     GetBucketPolicyStatusInput, GetBucketPolicyStatusOutput, GetBucketReplicationInput, GetBucketTaggingInput,
-    GetBucketVersioningInput, GetPublicAccessBlockInput, HeadBucketInput, HeadBucketOutput, LifecycleRule, ListBucketsInput,
-    ListBucketsOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsInput, ListObjectsOutput,
-    ListObjectsV2Input, ListObjectsV2Output, MetadataEntry, NotificationConfiguration, NotificationConfigurationFilter, Object,
-    ObjectStorageClass, ObjectVersion, ObjectVersionStorageClass, PolicyStatus, PutBucketCorsInput, PutBucketCorsOutput,
-    PutBucketEncryptionInput, PutBucketEncryptionOutput, PutBucketLifecycleConfigurationInput,
-    PutBucketLifecycleConfigurationOutput, PutBucketNotificationConfigurationInput, PutBucketNotificationConfigurationOutput,
-    PutBucketPolicyInput, PutBucketPolicyOutput, PutBucketReplicationInput, PutBucketReplicationOutput, PutBucketTaggingInput,
-    PutBucketTaggingOutput, PutBucketVersioningInput, PutBucketVersioningOutput, PutPublicAccessBlockInput,
-    PutPublicAccessBlockOutput, ServerSideEncryption, Timestamp, UserMetadata,
+    GetBucketVersioningInput, GetPublicAccessBlockInput, HeadBucketInput, HeadBucketOutput, LifecycleRule,
+    ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsInput, ListObjectsOutput, ListObjectsV2Input,
+    ListObjectsV2Output, MetadataEntry, NotificationConfiguration, NotificationConfigurationFilter, Object, ObjectStorageClass,
+    ObjectVersion, ObjectVersionStorageClass, PolicyStatus, PutBucketCorsInput, PutBucketCorsOutput, PutBucketEncryptionInput,
+    PutBucketEncryptionOutput, PutBucketLifecycleConfigurationInput, PutBucketLifecycleConfigurationOutput,
+    PutBucketNotificationConfigurationInput, PutBucketNotificationConfigurationOutput, PutBucketPolicyInput,
+    PutBucketPolicyOutput, PutBucketReplicationInput, PutBucketReplicationOutput, PutBucketTaggingInput, PutBucketTaggingOutput,
+    PutBucketVersioningInput, PutBucketVersioningOutput, PutPublicAccessBlockInput, PutPublicAccessBlockOutput,
+    ServerSideEncryption, Timestamp, UserMetadata,
 };
 use s3s::region::Region;
 use s3s::xml;
@@ -1403,6 +1406,40 @@ fn lifecycle_has_abort_multipart_rules(config: &PersistedLifecycleConfiguration)
         .any(|rule| rule.status == "Enabled" && rule.abort_incomplete_multipart_upload.is_some())
 }
 
+/// The ListBuckets answer for `bucket_infos`, in the gateway's types.
+///
+/// A stored name the S3 bucket grammar refuses cannot be written into the answer and is left out.
+/// The storage inventory holds such names only for RustFS-internal buckets (the SFTP host-key
+/// store); `rustfs/src/protocols/client.rs` applies the same rule to its listings.
+fn list_buckets_output(bucket_infos: &[BucketInfo]) -> gateway_dto::ListBucketsOutput {
+    let buckets = bucket_infos
+        .iter()
+        .filter_map(|bucket_info| {
+            let name = BucketName::new(bucket_info.name.clone()).ok()?;
+            let created = bucket_info.created.unwrap_or(time::OffsetDateTime::UNIX_EPOCH);
+            let seconds = created.unix_timestamp();
+            let creation_date =
+                GatewayTimestamp::from_secs_nanos(seconds, created.nanosecond()).unwrap_or(GatewayTimestamp::from_secs(seconds));
+            Some(gateway_dto::Bucket {
+                name,
+                creation_date,
+                ..Default::default()
+            })
+        })
+        .collect();
+    let owner = rustfs_owner();
+
+    gateway_dto::ListBucketsOutput {
+        buckets,
+        owner: gateway_dto::Owner {
+            id: owner.id,
+            display_name: owner.display_name,
+        },
+        continuation_token: None,
+        prefix: None,
+    }
+}
+
 #[derive(Clone, Default)]
 pub struct DefaultBucketUsecase {
     context: Option<Arc<AppContext>>,
@@ -1739,21 +1776,61 @@ impl DefaultBucketUsecase {
         Ok(S3Response::new(GetBucketLocationOutput { location_constraint }))
     }
 
-    #[instrument(level = "debug", skip(self, req))]
-    pub async fn execute_list_buckets(&self, req: S3Request<ListBucketsInput>) -> S3Result<S3Response<ListBucketsOutput>> {
+    /// Lists the buckets the caller of `envelope` may see, for either edge.
+    ///
+    /// The access phase must have installed the caller's `ReqInfo` in `envelope`. A caller allowed
+    /// `s3:ListAllMyBuckets` sees every bucket; any other caller sees the buckets it may list or
+    /// locate, and is refused when there is none. `_input` is accepted and not read: RustFS answers
+    /// one unpaginated listing and ignores `prefix`, `max-buckets`, `continuation-token` and
+    /// `bucket-region`, as it always has.
+    ///
+    /// # Errors
+    ///
+    /// `AccessDenied` for an anonymous caller and for one that may see no bucket; `InternalError`
+    /// when the store or IAM is not initialized or the access phase did not run; otherwise the
+    /// storage error of the listing.
+    #[instrument(level = "debug", skip(self, envelope, _input))]
+    pub async fn execute_list_buckets(
+        &self,
+        envelope: RequestEnvelope,
+        _input: gateway_dto::ListBucketsInput,
+    ) -> rustfs_s3_types::S3Result<gateway_dto::ListBucketsOutput> {
         let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+            return Err(rustfs_s3_types::S3Error::with_message(
+                rustfs_s3_types::S3ErrorCode::InternalError,
+                "Not init",
+            ));
         };
 
-        if req.credentials.as_ref().is_none_or(|cred| cred.access_key.is_empty()) {
-            return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied"));
+        if envelope
+            .credentials()
+            .is_none_or(|credentials| credentials.access_key().is_empty())
+        {
+            return Err(rustfs_s3_types::S3Error::with_message(
+                rustfs_s3_types::S3ErrorCode::AccessDenied,
+                "Access Denied",
+            ));
         }
 
-        let iam_authorization = prepare_list_buckets_iam_authorization(&req).await?;
+        let Some(req_info) = envelope.req_info() else {
+            return Err(rustfs_s3_types::S3Error::with_message(
+                rustfs_s3_types::S3ErrorCode::InternalError,
+                "ReqInfo not found in request envelope",
+            ));
+        };
+        let iam_authorization = prepare_list_buckets_iam_authorization(ListBucketsAuthzRequest {
+            req_info,
+            headers: envelope.headers(),
+            raw_query: envelope.uri().query(),
+            remote_addr: envelope.remote_addr(),
+            client_info: envelope.client_info(),
+            server_context: envelope.server_context(),
+        })
+        .await?;
         let bucket_infos = if iam_authorization.is_allowed("", S3Action::ListAllMyBucketsAction).await {
             store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?
         } else {
-            log_list_buckets_iam_implicit_deny(&req)?;
+            log_list_buckets_iam_implicit_deny(req_info);
             let bucket_infos = store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?;
             let mut visible_bucket_infos = Vec::new();
             for info in bucket_infos {
@@ -1772,7 +1849,7 @@ impl DefaultBucketUsecase {
             visible_bucket_infos
         };
 
-        Ok(S3Response::new(build_list_buckets_output(&bucket_infos)))
+        Ok(list_buckets_output(&bucket_infos))
     }
 
     pub async fn execute_delete_bucket_encryption(
@@ -5084,13 +5161,60 @@ mod tests {
 
     #[tokio::test]
     async fn execute_list_buckets_returns_internal_error_when_store_uninitialized() {
-        let input = ListBucketsInput::builder().build().unwrap();
-
-        let req = build_request(input, Method::GET);
+        let req = build_request((), Method::GET);
         let usecase = DefaultBucketUsecase::without_context();
 
-        let err = usecase.execute_list_buckets(req).await.unwrap_err();
-        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        let err = usecase
+            .execute_list_buckets(RequestEnvelope::from_s3s(&req), gateway_dto::ListBucketsInput::default())
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &rustfs_s3_types::S3ErrorCode::InternalError);
+    }
+
+    fn bucket_info(name: &str, created: Option<time::OffsetDateTime>) -> BucketInfo {
+        BucketInfo {
+            name: name.to_owned(),
+            created,
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn list_buckets_output_carries_names_dates_and_the_owner() {
+        let created = time::OffsetDateTime::from_unix_timestamp_nanos(1_700_000_000_123_000_000).expect("valid timestamp");
+
+        let output = list_buckets_output(&[bucket_info("alpha", Some(created)), bucket_info("beta", None)]);
+
+        let listed: Vec<(&str, i64, u32)> = output
+            .buckets
+            .iter()
+            .map(|bucket| (bucket.name.as_str(), bucket.creation_date.secs(), bucket.creation_date.subsec_nanos()))
+            .collect();
+        assert_eq!(listed, [("alpha", 1_700_000_000, 123_000_000), ("beta", 0, 0)]);
+        assert!(output.owner.id.is_some(), "the listing names its owner");
+        assert_eq!(output.continuation_token, None);
+        assert_eq!(output.prefix, None);
+    }
+
+    #[test]
+    fn list_buckets_output_leaves_out_names_the_bucket_grammar_refuses() {
+        let output = list_buckets_output(&[
+            bucket_info("sftp_host_keys", None),
+            bucket_info("UPPER", None),
+            bucket_info("ab", None),
+            bucket_info("kept", None),
+        ]);
+
+        let names: Vec<&str> = output.buckets.iter().map(|bucket| bucket.name.as_str()).collect();
+        assert_eq!(names, ["kept"]);
+    }
+
+    #[test]
+    fn list_buckets_output_of_no_bucket_is_an_empty_listing_with_an_owner() {
+        let output = list_buckets_output(&[]);
+
+        assert!(output.buckets.is_empty());
+        assert!(output.owner.id.is_some());
     }
 
     #[tokio::test]

rustfs/src/app/storage_api.rs

@@ -319,10 +319,11 @@ pub(crate) mod access {
     #[cfg(test)]
     pub(crate) use crate::storage::storage_api::access_consumer::ReqInfo;
     pub(crate) use crate::storage::storage_api::access_consumer::{
-        PostObjectRequestMarker, TABLE_DATA_PLANE_LIST_CURSOR_PREFIX, TableDataPlaneListAccess, TableDataPlaneListCursorPosition,
-        TableDataPlanePublicationGuards, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard,
-        authorize_copy_object_tags, authorize_request, bucket_config_mutation_incarnation, delete_object_authorize_action,
-        has_bypass_governance_header, load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
+        ListBucketsAuthzRequest, PostObjectRequestMarker, TABLE_DATA_PLANE_LIST_CURSOR_PREFIX, TableDataPlaneListAccess,
+        TableDataPlaneListCursorPosition, TableDataPlanePublicationGuards, apply_bucket_generation_guard,
+        apply_copy_source_bucket_generation_guard, authorize_copy_object_tags, authorize_request,
+        bucket_config_mutation_incarnation, delete_object_authorize_action, has_bypass_governance_header,
+        load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
         prepare_list_buckets_iam_authorization, prepare_odm_read_generation, recursive_force_delete_has_authenticated_caller,
         replication_proxy_allowed, replication_request_authorized, req_info_mut, req_info_ref,
         retained_table_data_plane_publication_guards, set_requested_object_lock_retain_until,
@@ -1674,9 +1675,9 @@ pub(crate) mod timeout_wrapper {
 pub(crate) mod s3_api {
     pub(crate) mod bucket {
         pub(crate) use crate::storage::storage_api::s3_api_consumer::bucket::{
-            ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output,
-            build_list_objects_output, build_list_objects_v2_output, encode_list_objects_v2_continuation_token,
-            parse_list_object_versions_params, parse_list_objects_v2_params,
+            ListObjectVersionsParams, ListObjectsV2Params, build_list_object_versions_output, build_list_objects_output,
+            build_list_objects_v2_output, encode_list_objects_v2_continuation_token, parse_list_object_versions_params,
+            parse_list_objects_v2_params,
         };
         pub(crate) use crate::storage::storage_api::s3_api_consumer::common::rustfs_owner;
     }
@@ -1712,7 +1713,7 @@ pub(crate) mod bucket_usecase {
     pub(crate) mod contract {
         pub(crate) mod bucket {
             pub(crate) use super::super::super::storage_contracts::{
-                BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions,
+                BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions,
             };
         }
 

rustfs/src/config/config_struct.rs

@@ -78,7 +78,7 @@ impl FromStr for S3Stack {
 impl S3Stack {
     /// Unset selects the default; a set but invalid value, including an empty
     /// one, is a startup error.
-    fn from_env() -> std::io::Result<Self> {
+    pub(crate) fn from_env() -> std::io::Result<Self> {
         match std::env::var_os(ENV_RUSTFS_S3_STACK) {
             None => Ok(Self::default()),
             Some(value) => value

rustfs/src/server/gateway_handlers/access.rs

@@ -0,0 +1,61 @@
+// 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 access phase of a gateway handler: the legacy access hooks in their owned form.
+//!
+//! Responsible for: [`check`], what the legacy access phase runs once per request before any
+//! operation hook. Per-operation hooks join it here as their operations are registered.
+//! NOT responsible for: the decisions themselves. Every function here calls the function the
+//! legacy access phase calls (`storage/access.rs`), so the two stacks cannot answer differently.
+//! Upstream: each `impl Handler`. Downstream: the storage layer's access functions.
+//!
+//! The caller is resolved here on every request, and again by the gateway's `Authorizer` before
+//! it: the pinned gateway hands a handler no policy snapshot, so one resolution cannot be carried
+//! from the authorization stages to the handler.
+
+use crate::app::RequestEnvelope;
+use crate::runtime_sources::ServerContextSlot;
+use crate::storage_api::server::http::access::{admit_operation, resolve_req_info};
+use rustfs_s3_types::S3Result;
+use std::sync::Arc;
+
+/// Runs the per-request access check for `operation` and installs the caller in `envelope`.
+///
+/// The caller is the credential IAM holds for the access key the gateway verified, checked
+/// against the presented session token; an anonymous request resolves to no caller. The signed
+/// upload-size capabilities are then refused on any operation but their own, and the license is
+/// checked.
+///
+/// # Errors
+///
+/// What the legacy access phase answers: the credential refusals of the key lookup,
+/// `InvalidRequest` for a misplaced capability query, `AccessDenied` for a refused license, and
+/// `InternalError` when IAM or the license cannot be read.
+pub(crate) async fn check(
+    server_ctx: &Arc<ServerContextSlot>,
+    envelope: &mut RequestEnvelope,
+    operation: &'static str,
+) -> S3Result<()> {
+    let credentials = envelope.credentials();
+    let access_key = credentials.map(|credentials| credentials.access_key().to_owned());
+    let session_token = credentials
+        .and_then(|credentials| credentials.session_token())
+        .unwrap_or_default()
+        .to_owned();
+    let request_context = envelope.request_context().cloned();
+    let req_info = resolve_req_info(server_ctx, access_key.as_deref(), &session_token, request_context).await?;
+    envelope.install_req_info(req_info);
+    admit_operation(envelope.headers(), envelope.uri().query(), operation)?;
+    Ok(())
+}

rustfs/src/server/gateway_handlers/bucket/list_buckets.rs

@@ -0,0 +1,48 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! `ListBuckets` on the gateway stack.
+//!
+//! Responsible for: converting one accepted request into the envelope and input the RustFS use
+//! case takes, and its answer back.
+//! NOT responsible for: deciding which buckets the caller sees. The legacy access hook for this
+//! operation is empty and so is this one: `DefaultBucketUsecase::execute_list_buckets` decides
+//! per caller (every bucket with `s3:ListAllMyBuckets`, otherwise the buckets the caller may
+//! list or locate, `AccessDenied` when there is none), for both stacks.
+//! Upstream: the gateway's dispatch. Downstream: `crate::app::bucket_usecase`.
+
+use super::super::GatewayBackend;
+use super::super::access;
+use super::super::error::handler_error;
+use crate::app::RequestEnvelope;
+use crate::app::bucket_usecase::DefaultBucketUsecase;
+use rustfs_gateway::{Handler, HandlerResult, Operation, Req, Resp, dto};
+use rustfs_io_metrics::record_s3_op;
+use rustfs_s3_ops::S3Operation;
+
+impl Handler<dto::ListBuckets> for GatewayBackend {
+    async fn call(&self, request: Req<dto::ListBuckets>) -> HandlerResult<dto::ListBuckets> {
+        record_s3_op(S3Operation::ListBuckets);
+        let mut envelope = RequestEnvelope::from_gateway(request.context(), self.server_ctx()).map_err(handler_error)?;
+        access::check(self.server_ctx(), &mut envelope, <dto::ListBuckets as Operation>::NAME)
+            .await
+            .map_err(handler_error)?;
+        let usecase = DefaultBucketUsecase::with_context(self.server_ctx().app_context());
+        let output = usecase
+            .execute_list_buckets(envelope, request.into_input())
+            .await
+            .map_err(handler_error)?;
+        Ok(Resp::new(output))
+    }
+}

rustfs/src/server/gateway_handlers/bucket/mod.rs

@@ -0,0 +1,17 @@
+// 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.
+
+//! Bucket-level operations of the gateway stack, one file each.
+
+mod list_buckets;

rustfs/src/server/gateway_handlers/envelope.rs

@@ -0,0 +1,373 @@
+// 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.
+//! Builds a [`RequestEnvelope`] from a request the gateway accepted, authenticated and routed.
+//!
+//! Responsible for: `RequestEnvelope::from_gateway`, the gateway edge's counterpart of the
+//! legacy edge's `from_s3s` (`storage/s3_compat/envelope.rs`).
+//! NOT responsible for: the facts the access phase attaches afterwards (`ReqInfo` and what the
+//! operation hooks retain); `super::access` installs those.
+//! Upstream: the gateway's handler request context. Downstream: every gateway handler.
+//!
+//! Every fact is copied from where the gateway or a RustFS layer in front of it put it: the
+//! request line and the accepted header lines from the context, the caller from the verified
+//! principal, and the payloads the connection and request layers attached from the transport
+//! values the gateway carried through unchanged.
+
+use crate::app::object::request_body::BodyReadControl;
+use crate::app::{RequestCredentials, RequestEnvelope, RequestHead};
+use crate::auth::{
+    AuthType, VerifiedPresignedRequest, VerifiedSigV4Request, get_request_auth_type_with_query, get_session_token,
+};
+use crate::runtime_sources::ServerContextSlot;
+use crate::shared_types::{NativeTlsTransport, RemoteAddr};
+use crate::storage_api::server::http::request_context::RequestContext;
+use http::{Extensions, HeaderMap, Uri};
+use rustfs_gateway::{RequestContextView, TransportExtensions};
+use rustfs_s3_types::{S3Error, S3ErrorCode, S3Result};
+use rustfs_trusted_proxies::ClientInfo;
+use std::sync::Arc;
+
+impl RequestEnvelope {
+    /// Copies every request fact of `context` the app layer reads, for the server `server_ctx`.
+    ///
+    /// A payload the request does not carry leaves its field `None`. The caller's secret key is
+    /// copied only when the gateway handed it to this operation; the signature markers are set
+    /// only for a caller the gateway verified.
+    ///
+    /// # Errors
+    ///
+    /// `InternalError` when the accepted request target cannot be read back as a URI, or a
+    /// handed-over secret key is not text. Neither is the caller's fault.
+    pub(crate) fn from_gateway(context: &RequestContextView, server_ctx: &Arc<ServerContextSlot>) -> S3Result<Self> {
+        let uri = request_uri(context)?;
+        let mut headers = HeaderMap::new();
+        for (name, value) in context.headers().iter_raw() {
+            headers.append(name.clone(), value.clone());
+        }
+        let credentials = match context.principal() {
+            Some(principal) => {
+                let secret_key = principal
+                    .secret_key_from_authenticator_lookup()
+                    .map(|secret| secret.expose_secret());
+                Some(request_credentials(
+                    principal.access_key_id(),
+                    secret_key,
+                    get_session_token(&uri, &headers),
+                )?)
+            }
+            None => None,
+        };
+
+        let mut extensions = Extensions::new();
+        let transport = context.transport_extensions();
+        copy_payload::<Option<RemoteAddr>>(transport, &mut extensions);
+        copy_payload::<ClientInfo>(transport, &mut extensions);
+        copy_payload::<NativeTlsTransport>(transport, &mut extensions);
+        copy_payload::<RequestContext>(transport, &mut extensions);
+        copy_payload::<BodyReadControl>(transport, &mut extensions);
+        extensions.insert(Arc::clone(server_ctx));
+        // The legacy access phase marks a request whose SigV4 signature its edge verified; here
+        // the gateway verified it, and only a verified caller has a principal.
+        if credentials.is_some() {
+            match get_request_auth_type_with_query(&headers, uri.query()) {
+                AuthType::Presigned => {
+                    extensions.insert(VerifiedPresignedRequest);
+                    extensions.insert(VerifiedSigV4Request);
+                }
+                AuthType::Signed => {
+                    extensions.insert(VerifiedSigV4Request);
+                }
+                _ => {}
+            }
+        }
+
+        let scope = context.verified_scope();
+        let head = RequestHead {
+            method: context.method().clone(),
+            uri,
+            headers,
+            credentials,
+            region: scope.map(|scope| scope.region().to_owned()),
+            service: scope.map(|scope| scope.service().to_owned()),
+        };
+        Ok(Self::from_parts(head, extensions))
+    }
+}
+
+/// The request target as it arrived: the raw path and query, with the scheme and authority when
+/// the transport carried them (an absolute-form target, or HTTP/2).
+fn request_uri(context: &RequestContextView) -> S3Result<Uri> {
+    let path_and_query = match context.raw_query() {
+        "" => context.raw_path().to_owned(),
+        query => format!("{}?{query}", context.raw_path()),
+    };
+    let builder = match (context.target_scheme(), context.target_authority()) {
+        (Some(scheme), Some(authority)) => Uri::builder().scheme(scheme).authority(authority),
+        _ => Uri::builder(),
+    };
+    builder
+        .path_and_query(path_and_query)
+        .build()
+        .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "the accepted request target is not a URI"))
+}
+
+/// The caller's credentials as the app layer holds them. `secret_key` is what the gateway handed
+/// over, which is nothing unless the operation opted in.
+fn request_credentials(access_key: &str, secret_key: Option<&[u8]>, session_token: Option<&str>) -> S3Result<RequestCredentials> {
+    let session_token = session_token.map(str::to_owned);
+    let Some(secret_key) = secret_key else {
+        return Ok(RequestCredentials::without_secret_key(access_key.to_owned(), session_token));
+    };
+    let secret_key = std::str::from_utf8(secret_key)
+        .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "the caller's secret key is not text"))?;
+    // The gateway refuses a credential whose secret is empty, so a handed-over secret is never
+    // empty. Should one arrive, the request is refused rather than served with a secret that
+    // every stored secret would have to be compared against.
+    RequestCredentials::new(access_key.to_owned(), secret_key.to_owned(), session_token)
+        .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "the caller's secret key is empty"))
+}
+
+/// Copies the payload of type `T` a layer in front of the gateway attached, when one did.
+fn copy_payload<T: Clone + Send + Sync + 'static>(transport: &TransportExtensions, extensions: &mut Extensions) {
+    if let Some(payload) = transport.get::<T>() {
+        extensions.insert(payload.clone());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::server::gateway_handlers::error::handler_error;
+    use bytes::Bytes;
+    use http::{HeaderValue, Method, Request, StatusCode};
+    use http_body_util::{BodyExt, Empty};
+    use rustfs_gateway::sig::{AmzDate, PayloadMode, SigService, SigV4Signer, SigningCredentials, SigningRequest, SigningScope};
+    use rustfs_gateway::{
+        Credentials, Handler, HandlerResult, Limits, RegionSet, Req, Resp, ServiceBuilder, SigV4Authenticator, StaticCredentials,
+        WireRequest, allow_when, dto,
+    };
+    use std::net::{IpAddr, SocketAddr};
+    use std::sync::Mutex;
+    use std::time::Instant;
+
+    const PEER: ([u8; 4], u16) = ([10, 0, 0, 7], 41000);
+    const CLIENT: [u8; 4] = [203, 0, 113, 9];
+    const REQUEST_ID: &str = "envelope-request-id";
+    const SIGNER_ACCESS_KEY: &str = "AKIAENVELOPESIGNER";
+    const SIGNER_SECRET_KEY: &str = "envelope-signer-secret-key-0123456789";
+
+    /// Records the envelope the gateway edge builds for the request it is handed.
+    struct Probe {
+        server_ctx: Arc<ServerContextSlot>,
+        seen: Mutex<Option<RequestEnvelope>>,
+    }
+
+    impl Handler<dto::ListBuckets> for Probe {
+        async fn call(&self, request: Req<dto::ListBuckets>) -> HandlerResult<dto::ListBuckets> {
+            let envelope = RequestEnvelope::from_gateway(request.context(), &self.server_ctx).map_err(handler_error)?;
+            *self.seen.lock().expect("probe lock") = Some(envelope);
+            Ok(Resp::new(dto::ListBucketsOutput::default()))
+        }
+    }
+
+    /// Sends `request` through a gateway whose only handler is the probe, and returns what the
+    /// probe built. The test authorizer admits the one probed operation and nothing else.
+    async fn envelope_of(request: Request<Empty<Bytes>>, server_ctx: &Arc<ServerContextSlot>) -> RequestEnvelope {
+        let probe = Arc::new(Probe {
+            server_ctx: Arc::clone(server_ctx),
+            seen: Mutex::new(None),
+        });
+        let regions = RegionSet::new(["us-east-1"]).expect("a region");
+        let service = ServiceBuilder::new()
+            .rustfs_profile()
+            .authenticator(SigV4Authenticator::new(Arc::new(signer_credentials()), regions).rustfs_profile())
+            .authorizer(allow_when(|request| request.operation == "ListBuckets"))
+            .register::<dto::ListBuckets, _>(Arc::clone(&probe))
+            .build()
+            .expect("the probe service assembles");
+
+        let response = service.call(request).await;
+        let status = response.status();
+        if status != StatusCode::OK {
+            let body = response
+                .into_body()
+                .collect()
+                .await
+                .map(|body| body.to_bytes())
+                .unwrap_or_default();
+            panic!("the probe must be reached: {status} {}", String::from_utf8_lossy(&body));
+        }
+        probe
+            .seen
+            .lock()
+            .expect("probe lock")
+            .take()
+            .expect("the probe saw the request")
+    }
+
+    /// The one caller the probe service knows.
+    fn signer_credentials() -> StaticCredentials {
+        StaticCredentials::new().with(Credentials::new(SIGNER_ACCESS_KEY, SIGNER_SECRET_KEY.as_bytes()).expect("a valid key"))
+    }
+
+    /// A ListBuckets request header-signed now by the probe service's one caller.
+    fn signed_list_buckets() -> Request<Empty<Bytes>> {
+        let mut headers = HeaderMap::new();
+        headers.insert(http::header::HOST, HeaderValue::from_static("localhost"));
+        let probe = list_buckets_request("/").body(Bytes::new()).expect("static request");
+        let accepted = WireRequest::accept(probe, &Limits::default()).expect("an acceptable host");
+        let stamp = AmzDate::parse(&chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string()).expect("a stamp");
+        let scope = SigningScope::new(stamp.day(), "us-east-1", SigService::S3).expect("a scope");
+        let credentials = SigningCredentials::new(SIGNER_ACCESS_KEY, SIGNER_SECRET_KEY.as_bytes()).expect("credentials");
+        let mut signer = SigV4Signer::new(credentials, scope);
+        let method = Method::GET;
+        let signing = SigningRequest::new(
+            &method,
+            "/",
+            "",
+            &headers,
+            accepted.host().raw_for_signing(),
+            PayloadMode::Unsigned,
+            stamp,
+        );
+        let signed = signer.sign_headers(&signing).expect("a signable request");
+        let mut builder = Request::builder().method(method).uri("/");
+        for (name, value) in signed.headers() {
+            builder = builder.header(name, value);
+        }
+        builder.body(Empty::new()).expect("a request")
+    }
+
+    fn list_buckets_request(target: &str) -> http::request::Builder {
+        Request::builder().method(Method::GET).uri(target).header("host", "localhost")
+    }
+
+    fn request_context() -> RequestContext {
+        RequestContext {
+            request_id: REQUEST_ID.to_owned(),
+            x_amz_request_id: REQUEST_ID.to_owned(),
+            trace_id: None,
+            span_id: None,
+            start_time: Instant::now(),
+        }
+    }
+
+    #[tokio::test]
+    async fn from_gateway_copies_every_fact_of_the_accepted_request() {
+        let server_ctx = ServerContextSlot::new();
+        let request = list_buckets_request("/?prefix=probe")
+            .header("x-envelope-probe", "probe")
+            .extension(Some(RemoteAddr(SocketAddr::from(PEER))))
+            .extension(ClientInfo::direct(SocketAddr::new(IpAddr::from(CLIENT), 443)))
+            .extension(NativeTlsTransport(true))
+            .extension(request_context())
+            .extension(BodyReadControl::default())
+            .body(Empty::new())
+            .expect("static request");
+
+        let envelope = envelope_of(request, &server_ctx).await;
+
+        assert_eq!(envelope.method(), Method::GET);
+        assert_eq!(envelope.uri().path(), "/");
+        assert_eq!(envelope.uri().query(), Some("prefix=probe"));
+        assert_eq!(envelope.headers().get("x-envelope-probe"), Some(&HeaderValue::from_static("probe")));
+        assert_eq!(envelope.remote_addr(), Some(SocketAddr::from(PEER)));
+        assert_eq!(envelope.client_info().map(|info| info.real_ip), Some(IpAddr::from(CLIENT)));
+        assert_eq!(envelope.native_tls_transport(), Some(true));
+        assert_eq!(envelope.request_context().map(|context| context.request_id.as_str()), Some(REQUEST_ID));
+        assert!(envelope.body_read_control().is_some(), "body_read_control");
+        assert!(
+            envelope.server_context().is_some_and(|slot| Arc::ptr_eq(slot, &server_ctx)),
+            "server_context"
+        );
+        assert!(envelope.residual_extensions().is_empty(), "every copied payload is a named field");
+    }
+
+    #[tokio::test]
+    async fn from_gateway_leaves_absent_payloads_none_and_an_anonymous_caller_unverified() {
+        let server_ctx = ServerContextSlot::new();
+        let request = list_buckets_request("/").body(Empty::new()).expect("static request");
+
+        let envelope = envelope_of(request, &server_ctx).await;
+
+        assert!(envelope.credentials().is_none(), "credentials");
+        assert_eq!(envelope.region(), None, "region");
+        assert_eq!(envelope.service(), None, "service");
+        assert_eq!(envelope.remote_addr(), None, "remote_addr");
+        assert!(envelope.client_info().is_none(), "client_info");
+        assert_eq!(envelope.native_tls_transport(), None, "native_tls_transport");
+        assert!(envelope.request_context().is_none(), "request_context");
+        assert!(envelope.body_read_control().is_none(), "body_read_control");
+        assert!(envelope.verified_sigv4().is_none(), "verified_sigv4");
+        assert!(envelope.verified_presigned().is_none(), "verified_presigned");
+        assert!(envelope.req_info().is_none(), "the access phase installs req_info, not the edge");
+    }
+
+    #[tokio::test]
+    async fn from_gateway_treats_a_peerless_remote_addr_as_absent() {
+        let server_ctx = ServerContextSlot::new();
+        let request = list_buckets_request("/")
+            .extension(None::<RemoteAddr>)
+            .body(Empty::new())
+            .expect("static request");
+
+        let envelope = envelope_of(request, &server_ctx).await;
+
+        assert_eq!(envelope.remote_addr(), None);
+        assert!(envelope.residual_extensions().is_empty(), "the peerless marker is consumed, not kept");
+    }
+
+    #[test]
+    fn credentials_carry_the_secret_only_when_the_gateway_handed_it_over() {
+        let kept = request_credentials("AKIAENVELOPE", None, Some("token")).expect("credentials");
+        assert_eq!(kept.access_key(), "AKIAENVELOPE");
+        assert!(kept.secret_key().is_none());
+        assert_eq!(kept.session_token(), Some("token"));
+
+        let handed = request_credentials("AKIAENVELOPE", Some(b"envelope-secret"), None).expect("credentials");
+        assert_eq!(handed.secret_key(), Some("envelope-secret"));
+        assert_eq!(handed.session_token(), None);
+    }
+
+    #[test]
+    fn an_empty_handed_over_secret_is_refused_instead_of_kept() {
+        let error = request_credentials("AKIAENVELOPE", Some(b""), None).expect_err("empty");
+
+        assert_eq!(error.code(), &S3ErrorCode::InternalError);
+    }
+
+    /// ListBuckets does not opt in to the caller's secret, so the gateway keeps it: the handler
+    /// sees who signed and no secret, and a reader that needs one is refused instead of being
+    /// handed an empty string to compare with.
+    #[tokio::test]
+    async fn an_operation_that_did_not_opt_in_reaches_the_handler_without_the_caller_s_secret() {
+        let server_ctx = ServerContextSlot::new();
+
+        let envelope = envelope_of(signed_list_buckets(), &server_ctx).await;
+
+        let credentials = envelope.credentials().expect("a signed request has a caller");
+        assert_eq!(credentials.access_key(), SIGNER_ACCESS_KEY);
+        assert!(credentials.secret_key().is_none(), "the secret stays with the gateway");
+        assert!(credentials.require_secret_key().is_err(), "a reader of the secret fails closed");
+        assert!(!credentials.secret_key_matches(""), "an absent secret equals no stored secret");
+        assert!(!credentials.secret_key_matches(SIGNER_SECRET_KEY));
+    }
+
+    #[test]
+    fn a_secret_that_is_not_text_is_refused_instead_of_dropped() {
+        let error = request_credentials("AKIAENVELOPE", Some(&[0xff, 0xfe]), None).expect_err("not text");
+
+        assert_eq!(error.code(), &S3ErrorCode::InternalError);
+    }
+}

rustfs/src/server/gateway_handlers/error.rs

@@ -0,0 +1,174 @@
+// 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 one conversion of a RustFS error into the gateway's handler error.
+//!
+//! Responsible for: [`handler_error`], called by every gateway handler and by nothing else.
+//! NOT responsible for: choosing an error (the use case's), or writing it (the gateway renders the
+//! document, the status and the request identifiers).
+//! Upstream: `rustfs_s3_types::S3Error`. Downstream: `rustfs_gateway::HandlerError`.
+//!
+//! The gateway's error carries a code with its status, a message, and response headers from a
+//! closed set. An error this function cannot carry whole is answered `500 InternalError` rather
+//! than sent with a member missing: a `416` without its `Content-Range`, or a throttle without
+//! its `Retry-After`, tells a client something RustFS did not say. Each operation slice that
+//! raises a header not listed in [`error_header`] adds it there, with its test.
+
+use http::{HeaderName, HeaderValue, StatusCode};
+use rustfs_gateway::{ErrorHeader, HandlerError};
+use rustfs_gateway_types::ErrorCode;
+use rustfs_s3_types::S3Error;
+use std::borrow::Cow;
+
+/// Answered when an error names a code without a status: the raiser of a custom code must set one.
+const MISSING_STATUS: &str = "the error carries a code without a status";
+/// Answered when an error carries a response header this edge cannot write.
+const UNWRITABLE_HEADER: &str = "the error carries a response header the gateway cannot write";
+
+/// Converts a use-case error into the error the gateway answers with: the same code, status and
+/// message, and every response header it carries.
+pub(crate) fn handler_error(error: S3Error) -> HandlerError {
+    let parts = error.into_parts();
+    let Some(status) = parts.status.or_else(|| parts.code.status_code()) else {
+        return HandlerError::internal_error(MISSING_STATUS);
+    };
+    let mut mapped = HandlerError::new(error_code(parts.code.as_str(), status), parts.message.unwrap_or(Cow::Borrowed("")));
+    for (name, value) in parts.headers.iter().flatten() {
+        let Some(header) = error_header(name, value) else {
+            return HandlerError::internal_error(UNWRITABLE_HEADER);
+        };
+        mapped = mapped.with_header(header);
+    }
+    mapped
+}
+
+/// The gateway's code for `name` answered with `status`: its own declared code when both agree,
+/// otherwise a code with exactly the spelling and status RustFS chose.
+fn error_code(name: &str, status: StatusCode) -> ErrorCode {
+    match ErrorCode::known(name) {
+        Some(declared) if declared.default_status() == status => declared,
+        _ => ErrorCode::custom(name.to_owned(), status),
+    }
+}
+
+/// The gateway's spelling of one error response header, or `None` for a header it cannot write.
+fn error_header(name: &HeaderName, value: &HeaderValue) -> Option<ErrorHeader> {
+    if name == http::header::RETRY_AFTER {
+        let seconds = value.to_str().ok()?.parse().ok()?;
+        return Some(ErrorHeader::RetryAfter { seconds });
+    }
+    None
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use http::HeaderMap;
+    use rustfs_s3_types::S3ErrorCode;
+
+    #[test]
+    fn a_declared_code_keeps_its_spelling_status_and_message() {
+        let mapped = handler_error(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied"));
+
+        assert_eq!(mapped.code().as_str(), "AccessDenied");
+        assert_eq!(mapped.code().default_status(), StatusCode::FORBIDDEN);
+        assert_eq!(mapped.message(), "Access Denied");
+        assert!(mapped.headers().is_empty());
+    }
+
+    #[test]
+    fn an_internal_error_stays_a_server_fault() {
+        let mapped = handler_error(S3Error::with_message(S3ErrorCode::InternalError, "Not init"));
+
+        assert_eq!(mapped.code().as_str(), "InternalError");
+        assert_eq!(mapped.code().default_status(), StatusCode::INTERNAL_SERVER_ERROR);
+        assert_eq!(mapped.message(), "Not init");
+    }
+
+    #[test]
+    fn an_explicit_status_overrides_the_declared_one() {
+        let mut error = S3Error::with_message(S3ErrorCode::AccessDenied, "locked");
+        error.set_status_code(StatusCode::CONFLICT);
+
+        let mapped = handler_error(error);
+
+        assert_eq!(mapped.code().as_str(), "AccessDenied");
+        assert_eq!(mapped.code().default_status(), StatusCode::CONFLICT);
+    }
+
+    #[test]
+    fn a_custom_code_keeps_the_status_its_raiser_set() {
+        let mut error = S3Error::with_message(S3ErrorCode::Custom("RustFsOnly".into()), "custom");
+        error.set_status_code(StatusCode::SERVICE_UNAVAILABLE);
+
+        let mapped = handler_error(error);
+
+        assert_eq!(mapped.code().as_str(), "RustFsOnly");
+        assert_eq!(mapped.code().default_status(), StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(mapped.message(), "custom");
+    }
+
+    #[test]
+    fn a_custom_code_without_a_status_is_a_server_fault_not_a_guess() {
+        let mapped = handler_error(S3Error::with_message(S3ErrorCode::Custom("RustFsOnly".into()), "custom"));
+
+        assert_eq!(mapped.code().as_str(), "InternalError");
+        assert_eq!(mapped.code().default_status(), StatusCode::INTERNAL_SERVER_ERROR);
+        assert_eq!(mapped.message(), MISSING_STATUS);
+    }
+
+    #[test]
+    fn a_missing_message_is_not_invented() {
+        let mapped = handler_error(S3Error::new(S3ErrorCode::AccessDenied));
+
+        assert_eq!(mapped.code().as_str(), "AccessDenied");
+        assert_eq!(mapped.message(), "");
+    }
+
+    #[test]
+    fn a_retry_after_header_is_carried() {
+        let mut headers = HeaderMap::new();
+        headers.insert(http::header::RETRY_AFTER, HeaderValue::from_static("7"));
+        let mut error = S3Error::with_message(S3ErrorCode::SlowDown, "busy");
+        error.set_headers(headers);
+
+        let mapped = handler_error(error);
+
+        assert_eq!(mapped.code().as_str(), "SlowDown");
+        assert_eq!(mapped.headers(), [ErrorHeader::RetryAfter { seconds: 7 }]);
+    }
+
+    #[test]
+    fn a_header_the_gateway_cannot_write_fails_the_whole_error() {
+        let mut headers = HeaderMap::new();
+        headers.insert("x-rustfs-unlisted", HeaderValue::from_static("value"));
+        let mut error = S3Error::with_message(S3ErrorCode::AccessDenied, "denied");
+        error.set_headers(headers);
+
+        let mapped = handler_error(error);
+
+        assert_eq!(mapped.code().as_str(), "InternalError");
+        assert_eq!(mapped.message(), UNWRITABLE_HEADER);
+        assert!(mapped.headers().is_empty());
+    }
+
+    #[test]
+    fn an_unreadable_retry_after_fails_the_whole_error() {
+        let mut headers = HeaderMap::new();
+        headers.insert(http::header::RETRY_AFTER, HeaderValue::from_static("soon"));
+        let mut error = S3Error::with_message(S3ErrorCode::SlowDown, "busy");
+        error.set_headers(headers);
+
+        assert_eq!(handler_error(error).message(), UNWRITABLE_HEADER);
+    }
+}

rustfs/src/server/gateway_handlers/mod.rs

@@ -0,0 +1,68 @@
+// 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 S3 operations the gateway stack serves, one file per operation (rustfs/backlog#2734).
+//!
+//! Responsible for: [`GatewayBackend`], the one backend every operation is implemented on, and
+//! [`register`], the explicit list of what is registered. An operation that is not in that list
+//! has no handler, and the gateway answers it `501 NotImplemented`.
+//! NOT responsible for: assembling the service (`server/gateway_stack.rs`), authenticating or
+//! routing a request (the gateway), or any S3 behaviour: a handler converts and delegates, and
+//! the RustFS use case it calls is the one the legacy stack calls.
+//! Upstream: the gateway's dispatch. Downstream: `crate::app` use cases.
+//!
+//! # The shape every handler has
+//!
+//! `<family>/<snake_op>.rs` holds exactly one `impl Handler<Op> for GatewayBackend`, in four
+//! steps and nothing else:
+//!
+//! 1. build the [`RequestEnvelope`](crate::app::RequestEnvelope) with `from_gateway`;
+//! 2. run the access phase: [`access::check`], then the operation's own access hook, which is
+//!    the authoritative authorization and runs exactly where the legacy stack runs it;
+//! 3. call the use case with the envelope and the gateway's own input type;
+//! 4. map a refusal with [`error::handler_error`], the only place an error is converted.
+//!
+//! The gateway's `Authorizer` (`server/gateway_stack/authorizer.rs`) is an earlier gate in front
+//! of step 2 and never a replacement for it.
+
+pub(crate) mod access;
+mod bucket;
+mod envelope;
+pub(crate) mod error;
+
+use crate::runtime_sources::ServerContextSlot;
+use rustfs_gateway::{ServiceBuilder, dto};
+use std::sync::Arc;
+
+/// The backend every gateway operation is implemented on: the server whose use cases answer.
+pub(crate) struct GatewayBackend {
+    server_ctx: Arc<ServerContextSlot>,
+}
+
+impl GatewayBackend {
+    pub(crate) fn new(server_ctx: Arc<ServerContextSlot>) -> Self {
+        Self { server_ctx }
+    }
+
+    pub(crate) fn server_ctx(&self) -> &Arc<ServerContextSlot> {
+        &self.server_ctx
+    }
+}
+
+/// Registers every operation the gateway stack serves. Each slice adds one line here and one row
+/// to the authorization table; the census in `server/gateway_stack/authorizer.rs` holds the two
+/// lists equal.
+pub(crate) fn register(builder: ServiceBuilder, backend: &Arc<GatewayBackend>) -> ServiceBuilder {
+    builder.register::<dto::ListBuckets, _>(Arc::clone(backend))
+}

rustfs/src/server/gateway_stack.rs

@@ -0,0 +1,323 @@
+// 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 S3 stack of one server: its assembly, and the facts the host hands every request
+//! (rustfs/backlog#2734 T2.5).
+//!
+//! Responsible for: [`GatewayStack::assemble`], the one place the `rustfs-gateway` service is
+//! built for RustFS — the RustFS profile, signature verification against RustFS IAM, the RustFS
+//! authorizer, explicit deadlines, and the registered handlers — and [`GatewayStack::call`],
+//! which attaches the host's facts to a request before the gateway accepts it.
+//! NOT responsible for: choosing the stack (`server/stack.rs`), the middleware around it
+//! (`server/http.rs`), or any S3 behaviour (`server/gateway_handlers` and the use cases).
+//! Upstream: `server/http.rs`. Downstream: `rustfs-gateway`, `rustfs-gateway-adapter`.
+//!
+//! # What the host keeps
+//!
+//! RustFS's own layers stay in front of the gateway and the gateway's counterparts stay off
+//! (rustfs/backlog#2771): CORS is answered by `ConditionalCorsLayer`, so no CORS source is
+//! installed and the gateway writes no CORS header; the SSE-C transport layer runs first and the
+//! gateway's own strict gate stays behind it; and the request identifier is the one
+//! `ExternalRequestContextLayer` assigned, handed over as the gateway's `HostRequestId`, so an
+//! answer and RustFS's logs name a request identically.
+
+// The stack is under construction, so a build without debug assertions must not carry it. The
+// message is checked by `scripts/check_gateway_stack_release_boundary.sh`.
+#[cfg(not(debug_assertions))]
+compile_error!("gateway-stack is restricted to debug builds");
+
+mod authorizer;
+
+use self::authorizer::{AUTHORIZATION_TABLE, OwnedDecisionCore, RequestLine, RustfsAuthorizer};
+use crate::runtime_sources::{ServerContextSlot, current_action_credentials};
+use crate::server::gateway_handlers::{GatewayBackend, register};
+use crate::server::ssec_transport::is_secure_transport;
+use crate::shared_types::{NativeTlsTransport, RemoteAddr};
+use crate::storage_api::server::http::request_context::RequestContext;
+use http::{Extensions, Request, Response};
+use rustfs_config::RUSTFS_REGION;
+use rustfs_gateway::{
+    AssemblyError, Authorizer, BoxFuture, ClientAddr, DEFAULT_MAX_BUFFERED_BODY_BYTES, HandlerDeadlineClass,
+    HandlerDeadlineConfig, MintedTraces, RegionSet, RequestBodyDeadlineConfig, ServiceBuilder, ServiceConfig, SigV4Authenticator,
+    SseConfig, TransportSecurity,
+};
+use rustfs_gateway_adapter::{
+    IamAdapters, IamKeyLookup, IamPrincipalLookup, RequestIdCarrier, RootCredentials, RootCredentialsError, RootSource,
+    host_request_id, rustfs_host_resolver,
+};
+use rustfs_iam::error::Error as IamError;
+use rustfs_iam::manager::get_token_signing_key;
+use rustfs_iam::store::object::ObjectStore;
+use rustfs_iam::sys::IamSys;
+use rustfs_policy::auth::UserIdentity;
+use rustfs_trusted_proxies::ClientInfo;
+use std::future::Future;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// What the assembly takes from the server it is built for.
+pub(crate) struct GatewayAssembly {
+    pub(crate) server_ctx: Arc<ServerContextSlot>,
+    /// The configured root credentials, checked once here. Requests read the live ones.
+    pub(crate) root_access_key: String,
+    pub(crate) root_secret_key: String,
+    /// The configured region; the default region signs when none is configured.
+    pub(crate) region: Option<String>,
+    /// The virtual-hosted-style base domains of the S3 listener, as the legacy stack reads them.
+    pub(crate) server_domains: Vec<String>,
+    /// `RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT`: the longest silence between two body reads. Zero
+    /// disables the bound, as it does on the legacy stack.
+    pub(crate) body_read_timeout: Duration,
+    /// Whether Keystone authentication is enabled in configuration (`RUSTFS_KEYSTONE_ENABLE`).
+    pub(crate) keystone_enabled: bool,
+}
+
+/// Whether Keystone authentication is switched on in configuration. A Keystone configuration
+/// that cannot be read counts as switched on: only a readable configuration can say it is off.
+pub(crate) fn keystone_enabled_in_configuration() -> bool {
+    rustfs_keystone::KeystoneConfig::from_env().map_or(true, |config| config.enable)
+}
+
+/// Why the gateway stack could not be assembled. Startup fails with it; the legacy stack is never
+/// started in its place.
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum GatewayAssemblyError {
+    #[error("the root credentials cannot verify signatures on the gateway stack: {0}")]
+    RootCredentials(RootCredentialsError),
+    #[error("region {region:?} cannot be a signing region on the gateway stack")]
+    Region { region: String },
+    #[error("the server domains cannot be served by the gateway stack: {0:?}")]
+    ServerDomains(rustfs_gateway::LegacyDomainError),
+    #[error("the gateway service refused its assembly: {0:?}")]
+    Service(AssemblyError),
+    #[error(
+        "RUSTFS_S3_STACK=gateway cannot be served while RUSTFS_KEYSTONE_ENABLE is on: the gateway admits a request that \
+         carries only X-Auth-Token as anonymous while RustFS reads the Keystone identity behind it; Keystone support on the \
+         gateway stack is rustfs/gateway#1586"
+    )]
+    KeystoneEnabled,
+}
+
+/// The assembled gateway service.
+#[derive(Clone)]
+pub(crate) struct GatewayStack {
+    service: rustfs_gateway::S3Service,
+}
+
+impl GatewayStack {
+    /// Assembles the gateway stack for one server.
+    ///
+    /// # Errors
+    ///
+    /// [`GatewayAssemblyError`] naming the input the gateway refused. The secret key is never
+    /// part of it.
+    pub(crate) fn assemble(assembly: GatewayAssembly) -> Result<Self, GatewayAssemblyError> {
+        let iam = Arc::new(ServerIam {
+            server_ctx: Arc::clone(&assembly.server_ctx),
+        });
+        let root = Arc::new(live_root(Arc::clone(&assembly.server_ctx)));
+        let core = OwnedDecisionCore::new(Arc::clone(&assembly.server_ctx));
+        let authorizer = RustfsAuthorizer::new(AUTHORIZATION_TABLE, Arc::new(core));
+        Self::assemble_with(assembly, iam, root, authorizer)
+    }
+
+    fn assemble_with(
+        assembly: GatewayAssembly,
+        iam: Arc<dyn IamPrincipalLookup>,
+        root: Arc<dyn RootSource>,
+        authorizer: impl Authorizer,
+    ) -> Result<Self, GatewayAssemblyError> {
+        // The pinned gateway admits a request that carries only `X-Auth-Token` as anonymous and
+        // runs no authenticator for it, while the RustFS code behind the gateway reads the
+        // Keystone identity: the request would be authorized as one caller and served as
+        // another. Refused until the gateway authenticates the token (rustfs/gateway#1586).
+        if assembly.keystone_enabled {
+            return Err(GatewayAssemblyError::KeystoneEnabled);
+        }
+        // A root the gateway could not verify a signature with is a startup error, never a
+        // server that silently has no root. Requests read the live root from `root`.
+        RootCredentials::new(&assembly.root_access_key, &assembly.root_secret_key, &assembly.root_secret_key)
+            .map_err(GatewayAssemblyError::RootCredentials)?;
+        let region = assembly.region.as_deref().unwrap_or(RUSTFS_REGION);
+        let regions = RegionSet::new([region]).map_err(|_| GatewayAssemblyError::Region {
+            region: region.to_owned(),
+        })?;
+        let hosts = rustfs_host_resolver(&assembly.server_domains).map_err(GatewayAssemblyError::ServerDomains)?;
+        let backend = Arc::new(GatewayBackend::new(Arc::clone(&assembly.server_ctx)));
+
+        // Every legacy RustFS reading of a request, a credential scope and an answer, in the
+        // gateway's one reviewable preset; its other half is on the authenticator below.
+        let builder = ServiceBuilder::new().rustfs_profile();
+        // The credential provider and the policy source go in together, over one IAM and one
+        // live root, so a key cannot be root to one and an ordinary user to the other.
+        let builder = IamAdapters::new(iam, root)
+            .install(builder, |credentials| SigV4Authenticator::new(credentials, regions).rustfs_profile());
+        let builder = builder
+            .authorizer(authorizer)
+            // The outer SSE-C transport layer answers first; the gateway's own gate stays strict
+            // behind it.
+            .sse_config(SseConfig::strict())
+            .host_resolver(hosts)
+            // The fallback only: `attach_host_facts` hands RustFS's own identifier over.
+            .trace_source(MintedTraces::with_uuid_request_ids());
+        let (builder, _settings) = builder.config(service_config(assembly.body_read_timeout));
+        let service = register(builder, &backend).build().map_err(GatewayAssemblyError::Service)?;
+        Ok(Self { service })
+    }
+
+    /// The operations the assembled service answers, read from its registry. Everything else is
+    /// answered `501 NotImplemented`.
+    pub(crate) fn registered_operations(&self) -> Vec<&'static str> {
+        self.service.operations().collect()
+    }
+
+    /// Answers one request on the gateway stack, with the host's facts attached.
+    pub(crate) fn call<B>(&self, mut request: Request<B>) -> impl Future<Output = Response<rustfs_gateway::Body>> + Send + 'static
+    where
+        B: http_body::Body + Send + 'static,
+        B::Data: Send,
+        B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+    {
+        // The authorizer is handed the headers and the query but not the request line, and the
+        // shared decision core is asked about the request as it arrived.
+        let line = RequestLine {
+            method: request.method().clone(),
+            uri: request.uri().clone(),
+        };
+        request.extensions_mut().insert(line);
+        attach_host_facts(request.extensions_mut());
+        let mut service = self.service.clone();
+        let answer = tower::Service::call(&mut service, request);
+        async move {
+            match answer.await {
+                Ok(response) => response,
+                Err(never) => match never {},
+            }
+        }
+    }
+}
+
+/// The gateway's request settings for RustFS, every deadline stated instead of inherited.
+///
+/// No handler runs under a total deadline, object data included: the legacy stack has none, and a
+/// large transfer must not be cut off for its duration. A committed answer is not bounded either.
+/// The request body keeps RustFS's own idle bound and no throughput floor.
+fn service_config(body_read_timeout: Duration) -> ServiceConfig {
+    let handlers = HandlerDeadlineConfig::default()
+        .without_deadline(HandlerDeadlineClass::Standard)
+        .without_deadline(HandlerDeadlineClass::Extended)
+        .without_commit_progress_deadline();
+    ServiceConfig::new(DEFAULT_MAX_BUFFERED_BODY_BYTES)
+        .with_handler_deadlines(handlers)
+        .with_request_body_deadlines(body_deadlines(body_read_timeout))
+}
+
+/// RustFS's body read timeout as the gateway's idle deadlines: the same silence is allowed before
+/// the first byte and between reads, and zero lifts both.
+fn body_deadlines(read_timeout: Duration) -> RequestBodyDeadlineConfig {
+    RequestBodyDeadlineConfig::new(read_timeout, read_timeout)
+        .unwrap_or(RequestBodyDeadlineConfig::S3.without_idle_deadlines())
+        .without_throughput_floor()
+}
+
+/// Attaches what the gateway reads about a request from its host: whether the transport was
+/// secure, the client address after RustFS's trusted-proxy correction, and RustFS's request
+/// identifier. Each is derived from what RustFS's own layers already attached; no header is read.
+fn attach_host_facts(extensions: &mut Extensions) {
+    let client_info = extensions.get::<ClientInfo>();
+    let native_tls = extensions.get::<NativeTlsTransport>().is_some_and(|transport| transport.0);
+    let security = if is_secure_transport(native_tls, client_info) {
+        TransportSecurity::Encrypted
+    } else {
+        TransportSecurity::Plaintext
+    };
+    let client_ip = client_info.map(|info| info.real_ip).or_else(|| {
+        extensions
+            .get::<Option<RemoteAddr>>()
+            .and_then(|peer| peer.as_ref().map(|peer| peer.0.ip()))
+    });
+    // An identifier outside the gateway's alphabet is not handed over; the gateway mints its own.
+    let request_id = extensions.get::<RequestContext>().and_then(host_request_id);
+
+    extensions.insert(security);
+    if let Some(client_ip) = client_ip {
+        extensions.insert(ClientAddr::from_peer(client_ip));
+    }
+    if let Some(request_id) = request_id {
+        extensions.insert(request_id);
+    }
+}
+
+/// The identifier the outer layer assigned is the one the gateway is handed.
+impl RequestIdCarrier for RequestContext {
+    fn request_id(&self) -> &str {
+        &self.request_id
+    }
+}
+
+/// The root credentials of one server as they are now, read on every call from where the legacy
+/// stack reads them, so a root change reaches both stacks at once. `None` while they are not
+/// initialized or cannot verify a signature, which the adapters answer as unavailable.
+fn live_root(server_ctx: Arc<ServerContextSlot>) -> impl Fn() -> Option<Arc<RootCredentials>> + Send + Sync + 'static {
+    move || {
+        let root = match server_ctx.app_context() {
+            Some(context) => context.action_credentials().get(),
+            None => current_action_credentials(),
+        }?;
+        // IAM signs STS session tokens with its token-signing key, which is the root secret
+        // until IAM has published one of its own.
+        let signing_key = get_token_signing_key().unwrap_or_else(|| root.secret_key.clone());
+        RootCredentials::new(&root.access_key, &root.secret_key, &signing_key)
+            .ok()
+            .map(Arc::new)
+    }
+}
+
+/// The IAM system of one server, looked up when a credential is: IAM is installed after the
+/// listener starts, so the assembly cannot hold a handle.
+struct ServerIam {
+    server_ctx: Arc<ServerContextSlot>,
+}
+
+impl ServerIam {
+    fn ready(&self) -> Option<Arc<IamSys<ObjectStore>>> {
+        self.server_ctx
+            .app_context()
+            .filter(|context| context.iam().is_ready())
+            .map(|context| context.iam().handle())
+    }
+}
+
+impl IamKeyLookup for ServerIam {
+    fn is_ready(&self) -> bool {
+        self.ready().is_some()
+    }
+
+    fn check_key<'a>(&'a self, access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>> {
+        Box::pin(async move {
+            let Some(iam) = self.ready() else {
+                return Err(IamError::IamSysNotInitialized);
+            };
+            iam.check_key(access_key).await
+        })
+    }
+}
+
+impl IamPrincipalLookup for ServerIam {
+    fn group_memberships(&self, user: &str) -> Vec<String> {
+        self.ready().map(|iam| iam.user_group_memberships(user)).unwrap_or_default()
+    }
+}
+
+#[cfg(test)]
+mod tests;

rustfs/src/server/gateway_stack/authorizer.rs

@@ -0,0 +1,687 @@
+// 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 `Authorizer` of the RustFS assembly: an early gate, never the authorization.
+//!
+//! Responsible for: [`AUTHORIZATION_TABLE`], one row per registered operation, and
+//! [`RustfsAuthorizer`], which answers both gateway authorization stages from it.
+//! NOT responsible for: the authoritative decision. That is the legacy access hook in its owned
+//! form, run by the operation's handler immediately before its use case
+//! (`server/gateway_handlers`), where the legacy stack runs it; decisions, error codes and their
+//! order are therefore the same code on both stacks (rustfs/backlog#2761).
+//! Upstream: the gateway's route and input authorization stages. Downstream: the shared
+//! authorization core, through [`DecisionCore`].
+//!
+//! # What this gate may do
+//!
+//! It may refuse a request the hook would certainly refuse too, and nothing else. It is never
+//! the only check, and it never allows by default:
+//!
+//! - an operation without a row is denied;
+//! - a `RouteDecidable` row asks the shared decision core the hook asks, with the same facts;
+//! - a `DeferToHandlerHook` row evaluates only its [`RouteGate`], which names exactly what the
+//!   route stage can know, and leaves the rest to the hook;
+//! - a `LegacyNoCheck` row is an operation whose legacy hook is empty, said so explicitly;
+//! - a core that cannot answer is `Indeterminate`, which the gateway refuses.
+//!
+//! Plain rows therefore evaluate policy twice, once here and once in the hook. That is accepted
+//! until it is measured.
+//!
+//! A row may be `RouteDecidable` only when its denial concerns the one routed resource and the
+//! one routed action. An operation whose denial can concern a second resource (a copy source) or
+//! a version-specific action (a versioned read) is a `DeferToHandlerHook` row: the hook writes
+//! the denial record with the resource and action it actually refused, and an early refusal
+//! here would record the wrong ones.
+//!
+//! # Deferred rows
+//!
+//! Every row that is not `RouteDecidable` is listed here, and the census below holds the table
+//! to these two lists, so a deferral cannot be added without this section changing.
+//!
+//! `DeferToHandlerHook`:
+//!
+//! - `ListBuckets`: the legacy hook is empty and the use case decides per caller. It lists every
+//!   bucket for a caller allowed `s3:ListAllMyBuckets`, otherwise the buckets the caller may list
+//!   or locate, and refuses when there is none; that needs the bucket inventory. The route stage
+//!   refuses an anonymous caller, which the use case refuses with the same answer.
+//!
+//! `LegacyNoCheck`: none.
+
+use crate::app::{RequestCredentials, RequestEnvelope, RequestHead, authorize_owned_request};
+use crate::auth::get_session_token;
+use crate::runtime_sources::ServerContextSlot;
+use crate::shared_types::RemoteAddr;
+use crate::storage_api::server::http::access::resolve_req_info;
+use crate::storage_api::server::http::request_context::RequestContext;
+use http::{Extensions, HeaderMap, Method, Uri};
+use rustfs_gateway::{
+    Authorizer, AuthzRequest, BoxFuture, Decision, InputAuthzRequest, InputDecisions, RequestContext as GatewayRequestContext,
+};
+use rustfs_policy::policy::action::{Action, S3Action};
+use rustfs_s3_types::{S3Error, S3ErrorCode, S3Result};
+use rustfs_trusted_proxies::ClientInfo;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+/// The resource an operation's action is asked about.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum ResourceForm {
+    /// The service itself: no bucket and no key.
+    Service,
+    /// One bucket.
+    #[cfg_attr(not(test), expect(dead_code, reason = "constructed by the first bucket-level row"))]
+    Bucket,
+    /// One object in one bucket.
+    #[cfg_attr(not(test), expect(dead_code, reason = "constructed by the first object-level row"))]
+    Object,
+}
+
+/// What a deferred row's route stage evaluates: only facts that do not depend on the body.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum RouteGate {
+    /// Refuses a request that presented no credentials. A request with a verified caller passes
+    /// to the hook, which decides.
+    AuthenticatedCaller,
+}
+
+/// How much of an operation's decision the route stage can make.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum RouteStage {
+    /// The route stage asks the shared decision core the same question, with the same facts, as
+    /// the hook.
+    #[cfg_attr(not(test), expect(dead_code, reason = "constructed by the first plain operation row"))]
+    RouteDecidable,
+    /// The route stage evaluates `gate` and the hook in the handler decides the rest.
+    DeferToHandlerHook {
+        /// Exactly what the route stage evaluates.
+        gate: RouteGate,
+        /// Why the route stage cannot decide the whole of it.
+        reason: &'static str,
+    },
+    /// The legacy hook for this operation is empty, so there is nothing to evaluate early.
+    #[cfg_attr(
+        not(test),
+        expect(dead_code, reason = "constructed by the row of an operation whose legacy hook is empty")
+    )]
+    LegacyNoCheck {
+        /// Why legacy checks nothing.
+        reason: &'static str,
+    },
+}
+
+/// The checks an operation's legacy hook performs beyond its one action, which the shared
+/// decision core does not perform. There is no default: every row states this column.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum ExtraChecks {
+    /// The hook asks its one action and nothing else.
+    #[cfg_attr(not(test), expect(dead_code, reason = "constructed by the first plain operation row"))]
+    None,
+    /// The hook, or the use case it defers to, also performs these.
+    Performed(&'static [&'static str]),
+}
+
+/// One operation's authorization, as the legacy access hook performs it.
+///
+/// A struct literal names every column, so a row cannot leave one unspecified.
+#[cfg_attr(
+    not(test),
+    expect(
+        dead_code,
+        reason = "the body-input and extra-check columns record what the handler hook does and are read by the census"
+    )
+)]
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct AuthorizationRow {
+    /// The operation, by the name it is registered under.
+    pub(crate) operation: &'static str,
+    /// The action the legacy hook asks, mapped from the operation and not from the gateway's
+    /// spelling of its action: the two differ for 22 operations, and legacy's is the baseline.
+    pub(crate) action: Action,
+    pub(crate) resource: ResourceForm,
+    pub(crate) route_stage: RouteStage,
+    /// The body-derived inputs the hook needs, which no route stage has.
+    pub(crate) body_inputs: &'static [&'static str],
+    pub(crate) extra_checks: ExtraChecks,
+}
+
+/// Every operation the gateway stack registers. Kept equal to the registered set by the census
+/// in `server/gateway_stack.rs`.
+pub(crate) static AUTHORIZATION_TABLE: &[AuthorizationRow] = &[AuthorizationRow {
+    operation: "ListBuckets",
+    action: Action::S3Action(S3Action::ListAllMyBucketsAction),
+    resource: ResourceForm::Service,
+    route_stage: RouteStage::DeferToHandlerHook {
+        gate: RouteGate::AuthenticatedCaller,
+        reason: "the use case decides per caller from the bucket inventory",
+    },
+    body_inputs: &[],
+    extra_checks: ExtraChecks::Performed(&[
+        "a caller without s3:ListAllMyBuckets sees the buckets it may list (s3:ListBucket) or locate (s3:GetBucketLocation), \
+         judged by IAM alone, and is refused when there is none",
+    ]),
+}];
+
+/// The request line as it reached RustFS, attached by the host before the gateway accepts the
+/// request. The gateway hands an authorizer the headers and the query, but neither the method
+/// nor the target, and the shared core is asked about the request as it arrived.
+#[derive(Clone, Debug)]
+pub(crate) struct RequestLine {
+    pub(crate) method: Method,
+    pub(crate) uri: Uri,
+}
+
+/// The request facts the shared decision core is asked with, each copied from where the gateway
+/// or a RustFS layer in front of it put it.
+pub(crate) struct RouteFacts<'a> {
+    /// The access key the gateway verified; `None` for an anonymous request.
+    pub(crate) access_key: Option<&'a str>,
+    /// The bucket, key and version the gateway routed, when the request names them.
+    pub(crate) bucket: Option<&'a str>,
+    pub(crate) key: Option<&'a str>,
+    pub(crate) version_id: Option<&'a str>,
+    /// The accepted header lines; `None` when the gateway handed over none.
+    pub(crate) headers: Option<HeaderMap>,
+    pub(crate) request_line: Option<&'a RequestLine>,
+    pub(crate) client_info: Option<&'a ClientInfo>,
+    pub(crate) remote_addr: Option<SocketAddr>,
+    pub(crate) request_context: Option<&'a RequestContext>,
+}
+
+/// The shared authorization decision core, as the route stage of a `RouteDecidable` row asks it.
+pub(crate) trait DecisionCore: Send + Sync + 'static {
+    /// Decides `row`'s action for the caller and resource of `facts`: `Ok` to allow,
+    /// `AccessDenied` to deny, and any other error when the core could not answer.
+    fn decide<'a>(&'a self, row: &'a AuthorizationRow, facts: &'a RouteFacts<'a>) -> BoxFuture<'a, S3Result<()>>;
+}
+
+/// The decision core both stacks share: the owned authorization entry point, asked about the
+/// request the route stage sees.
+///
+/// The caller is resolved by the function the legacy access check resolves it with, and the
+/// decision is `authorize_owned_request`, which the handler's access hook calls too. It is only
+/// the policy decision: a row that needs a body-derived input or a composite check is not
+/// `RouteDecidable`.
+pub(crate) struct OwnedDecisionCore {
+    server_ctx: Arc<ServerContextSlot>,
+}
+
+impl OwnedDecisionCore {
+    pub(crate) fn new(server_ctx: Arc<ServerContextSlot>) -> Self {
+        Self { server_ctx }
+    }
+
+    async fn authorize(&self, row: &AuthorizationRow, facts: &RouteFacts<'_>) -> S3Result<()> {
+        let (Some(line), Some(headers)) = (facts.request_line, facts.headers.as_ref()) else {
+            return Err(unanswerable("the route stage was handed no request line or headers"));
+        };
+        let session_token = get_session_token(&line.uri, headers).unwrap_or_default().to_owned();
+        let mut req_info =
+            resolve_req_info(&self.server_ctx, facts.access_key, &session_token, facts.request_context.cloned()).await?;
+        match row.resource {
+            ResourceForm::Service => {}
+            ResourceForm::Bucket => req_info.bucket = Some(routed(facts.bucket)?),
+            ResourceForm::Object => {
+                req_info.bucket = Some(routed(facts.bucket)?);
+                req_info.object = Some(routed(facts.key)?);
+                req_info.version_id = facts.version_id.map(str::to_owned);
+            }
+        }
+
+        let credentials = facts.access_key.map(|access_key| {
+            let session_token = (!session_token.is_empty()).then(|| session_token.clone());
+            RequestCredentials::without_secret_key(access_key.to_owned(), session_token)
+        });
+        let mut extensions = Extensions::new();
+        extensions.insert(facts.remote_addr.map(RemoteAddr));
+        if let Some(client_info) = facts.client_info {
+            extensions.insert(client_info.clone());
+        }
+        if let Some(request_context) = facts.request_context {
+            extensions.insert(request_context.clone());
+        }
+        extensions.insert(Arc::clone(&self.server_ctx));
+        let head = RequestHead {
+            method: line.method.clone(),
+            uri: line.uri.clone(),
+            headers: headers.clone(),
+            credentials,
+            region: None,
+            service: None,
+        };
+        let mut envelope = RequestEnvelope::from_parts(head, extensions);
+        envelope.install_req_info(req_info);
+        authorize_owned_request(&mut envelope, row.action, None).await
+    }
+}
+
+impl DecisionCore for OwnedDecisionCore {
+    fn decide<'a>(&'a self, row: &'a AuthorizationRow, facts: &'a RouteFacts<'a>) -> BoxFuture<'a, S3Result<()>> {
+        Box::pin(self.authorize(row, facts))
+    }
+}
+
+/// The routed name a row's resource form requires. Its absence is a question the core cannot
+/// answer, never a decision about some other resource.
+fn routed(name: Option<&str>) -> S3Result<String> {
+    name.map(str::to_owned)
+        .ok_or_else(|| unanswerable("the route stage was handed no name for the row's resource"))
+}
+
+fn unanswerable(message: &'static str) -> S3Error {
+    S3Error::with_message(S3ErrorCode::InternalError, message)
+}
+
+/// What the table decides about one request by itself, and what it leaves to the core.
+enum Question<'a> {
+    Decided(Decision),
+    AskCore(&'static AuthorizationRow, Box<RouteFacts<'a>>),
+}
+
+/// The RustFS implementation of the gateway's two authorization stages.
+pub(crate) struct RustfsAuthorizer {
+    table: &'static [AuthorizationRow],
+    core: Arc<dyn DecisionCore>,
+}
+
+impl RustfsAuthorizer {
+    /// The authorizer over `table`, asking `core` for `RouteDecidable` rows.
+    pub(crate) fn new(table: &'static [AuthorizationRow], core: Arc<dyn DecisionCore>) -> Self {
+        Self { table, core }
+    }
+
+    /// Reads `operation`'s row. `facts` is built only for a row that asks the core.
+    fn question<'f>(&self, operation: &str, anonymous: bool, facts: impl FnOnce() -> RouteFacts<'f>) -> Question<'f> {
+        let table: &'static [AuthorizationRow] = self.table;
+        let Some(row) = table.iter().find(|row| row.operation == operation) else {
+            // No row, no answer to give: refused, never allowed.
+            return Question::Decided(Decision::Deny);
+        };
+        match row.route_stage {
+            RouteStage::RouteDecidable => Question::AskCore(row, Box::new(facts())),
+            RouteStage::DeferToHandlerHook { gate, .. } => Question::Decided(match gate {
+                RouteGate::AuthenticatedCaller if anonymous => Decision::Deny,
+                RouteGate::AuthenticatedCaller => Decision::Allow,
+            }),
+            RouteStage::LegacyNoCheck { .. } => Question::Decided(Decision::Allow),
+        }
+    }
+
+    async fn answer(&self, question: Question<'_>) -> Decision {
+        match question {
+            Question::Decided(decision) => decision,
+            Question::AskCore(row, facts) => core_decision(self.core.decide(row, &facts).await),
+        }
+    }
+
+    fn ask<'a>(&self, context: &GatewayRequestContext<'a>, request: &AuthzRequest<'a>) -> Question<'a> {
+        self.question(request.operation, request.identity.is_none(), || route_facts(context, request))
+    }
+}
+
+/// Reads the core's answer as a decision. Only an explicit denial is a `Deny`; a failure to
+/// answer is `Indeterminate` and never an allowance.
+fn core_decision(answer: S3Result<()>) -> Decision {
+    match answer {
+        Ok(()) => Decision::Allow,
+        Err(error) if error.code() == &S3ErrorCode::AccessDenied => Decision::Deny,
+        Err(_) => Decision::Indeterminate,
+    }
+}
+
+fn route_facts<'a>(context: &GatewayRequestContext<'a>, request: &AuthzRequest<'a>) -> RouteFacts<'a> {
+    let host = context.server_extensions();
+    RouteFacts {
+        access_key: request.identity.map(|identity| identity.access_key_id()),
+        bucket: request.bucket.map(|bucket| bucket.as_str()),
+        key: request.key.map(|key| key.as_str()),
+        version_id: request.version_id,
+        headers: context.headers().map(|accepted| {
+            let mut headers = HeaderMap::new();
+            for (name, value) in accepted.iter_raw() {
+                headers.append(name.clone(), value.clone());
+            }
+            headers
+        }),
+        request_line: host.get::<RequestLine>(),
+        client_info: host.get::<ClientInfo>(),
+        remote_addr: host
+            .get::<Option<RemoteAddr>>()
+            .and_then(|peer| peer.as_ref().map(|peer| peer.0)),
+        request_context: host.get::<RequestContext>(),
+    }
+}
+
+impl Authorizer for RustfsAuthorizer {
+    fn authorize_route<'a>(
+        &'a self,
+        context: &'a GatewayRequestContext<'a>,
+        request: &'a AuthzRequest<'a>,
+    ) -> BoxFuture<'a, Decision> {
+        let question = self.ask(context, request);
+        Box::pin(self.answer(question))
+    }
+
+    fn authorize_input<'a>(
+        &'a self,
+        context: &'a GatewayRequestContext<'a>,
+        request: &'a InputAuthzRequest<'a>,
+    ) -> BoxFuture<'a, InputDecisions> {
+        // `decide_all` visits every derived resource and then the visibility question, in that
+        // order; each is read from the table now and answered below, where the core is awaited.
+        let stage = self.ask(context, request.route());
+        let mut derived: Vec<Question<'a>> = request
+            .resources()
+            .iter()
+            .map(|resource| self.ask(context, resource))
+            .collect();
+        if let Some(visibility) = request.visibility() {
+            derived.push(self.ask(context, visibility));
+        }
+        Box::pin(async move {
+            let stage = self.answer(stage).await;
+            let mut answers = Vec::with_capacity(derived.len());
+            for question in derived {
+                answers.push(self.answer(question).await);
+            }
+            let mut answers = answers.into_iter();
+            request.decide_all(stage, |_| answers.next().unwrap_or(Decision::Deny))
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::BTreeSet;
+    use std::sync::Mutex;
+
+    /// The rows documented as `DeferToHandlerHook` in this file's header.
+    const DOCUMENTED_DEFERRED_ROWS: &[&str] = &["ListBuckets"];
+    /// The rows documented as `LegacyNoCheck` in this file's header.
+    const DOCUMENTED_NO_CHECK_ROWS: &[&str] = &[];
+
+    static PLAIN_TABLE: &[AuthorizationRow] = &[
+        AuthorizationRow {
+            operation: "HeadBucket",
+            action: Action::S3Action(S3Action::ListBucketAction),
+            resource: ResourceForm::Bucket,
+            route_stage: RouteStage::RouteDecidable,
+            body_inputs: &[],
+            extra_checks: ExtraChecks::None,
+        },
+        AuthorizationRow {
+            operation: "GetObjectTorrent",
+            action: Action::S3Action(S3Action::GetObjectAction),
+            resource: ResourceForm::Object,
+            route_stage: RouteStage::LegacyNoCheck {
+                reason: "the legacy hook is empty",
+            },
+            body_inputs: &[],
+            extra_checks: ExtraChecks::None,
+        },
+    ];
+
+    /// A core that gives one scripted answer and records what it was asked.
+    struct ScriptedCore {
+        answer: fn() -> S3Result<()>,
+        asked: Mutex<Vec<(&'static str, Action, Option<String>)>>,
+    }
+
+    impl ScriptedCore {
+        fn answering(answer: fn() -> S3Result<()>) -> Arc<Self> {
+            Arc::new(Self {
+                answer,
+                asked: Mutex::new(Vec::new()),
+            })
+        }
+    }
+
+    impl DecisionCore for ScriptedCore {
+        fn decide<'a>(&'a self, row: &'a AuthorizationRow, facts: &'a RouteFacts<'a>) -> BoxFuture<'a, S3Result<()>> {
+            self.asked
+                .lock()
+                .expect("core lock")
+                .push((row.operation, row.action, facts.access_key.map(str::to_owned)));
+            let answer = (self.answer)();
+            Box::pin(async move { answer })
+        }
+    }
+
+    fn denied() -> S3Result<()> {
+        Err(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied"))
+    }
+
+    fn unavailable() -> S3Result<()> {
+        Err(S3Error::with_message(S3ErrorCode::InternalError, "the policy store is unavailable"))
+    }
+
+    /// One request as a test asks about it.
+    struct Asked<'a> {
+        operation: &'a str,
+        facts: RouteFacts<'a>,
+    }
+
+    fn facts<'a>(operation: &'a str, access_key: Option<&'a str>) -> Asked<'a> {
+        Asked {
+            operation,
+            facts: RouteFacts {
+                access_key,
+                bucket: None,
+                key: None,
+                version_id: None,
+                headers: None,
+                request_line: None,
+                client_info: None,
+                remote_addr: None,
+                request_context: None,
+            },
+        }
+    }
+
+    async fn decision(table: &'static [AuthorizationRow], core: Arc<dyn DecisionCore>, asked: Asked<'_>) -> Decision {
+        let authorizer = RustfsAuthorizer::new(table, core);
+        let anonymous = asked.facts.access_key.is_none();
+        let question = authorizer.question(asked.operation, anonymous, || asked.facts);
+        authorizer.answer(question).await
+    }
+
+    /// The real decision core of a server that has no object store yet.
+    fn owned_core() -> Arc<dyn DecisionCore> {
+        Arc::new(OwnedDecisionCore::new(ServerContextSlot::new()))
+    }
+
+    fn head_bucket() -> RequestLine {
+        RequestLine {
+            method: Method::HEAD,
+            uri: Uri::from_static("/bucket"),
+        }
+    }
+
+    fn rows_where(predicate: fn(&RouteStage) -> bool) -> BTreeSet<&'static str> {
+        AUTHORIZATION_TABLE
+            .iter()
+            .filter(|row| predicate(&row.route_stage))
+            .map(|row| row.operation)
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn an_operation_without_a_row_is_denied() {
+        let core = ScriptedCore::answering(|| Ok(()));
+
+        for operation in ["GetObject", "PutObject", "", "listbuckets"] {
+            let decided = decision(AUTHORIZATION_TABLE, core.clone(), facts(operation, Some("AKIDUSER0001"))).await;
+            assert_eq!(decided, Decision::Deny, "{operation:?}");
+        }
+        assert!(core.asked.lock().expect("core lock").is_empty(), "an unmapped operation asks nothing");
+    }
+
+    #[tokio::test]
+    async fn a_core_that_cannot_answer_is_indeterminate_never_an_allowance() {
+        let decided = decision(
+            PLAIN_TABLE,
+            ScriptedCore::answering(unavailable),
+            facts("HeadBucket", Some("AKIDUSER0001")),
+        )
+        .await;
+
+        assert_eq!(decided, Decision::Indeterminate);
+    }
+
+    #[tokio::test]
+    async fn the_owned_core_without_a_store_cannot_answer_and_never_allows() {
+        let line = head_bucket();
+        let mut asked = facts("HeadBucket", None);
+        asked.facts.bucket = Some("bucket");
+        asked.facts.headers = Some(HeaderMap::new());
+        asked.facts.request_line = Some(&line);
+
+        assert_eq!(decision(PLAIN_TABLE, owned_core(), asked).await, Decision::Indeterminate);
+    }
+
+    #[tokio::test]
+    async fn the_owned_core_refuses_a_request_the_host_did_not_describe() {
+        let line = head_bucket();
+
+        let mut without_a_line = facts("HeadBucket", None);
+        without_a_line.facts.bucket = Some("bucket");
+        without_a_line.facts.headers = Some(HeaderMap::new());
+        assert_eq!(decision(PLAIN_TABLE, owned_core(), without_a_line).await, Decision::Indeterminate);
+
+        let mut without_headers = facts("HeadBucket", None);
+        without_headers.facts.bucket = Some("bucket");
+        without_headers.facts.request_line = Some(&line);
+        assert_eq!(decision(PLAIN_TABLE, owned_core(), without_headers).await, Decision::Indeterminate);
+    }
+
+    #[tokio::test]
+    async fn the_owned_core_refuses_a_bucket_row_whose_request_names_no_bucket() {
+        let line = head_bucket();
+        let mut asked = facts("HeadBucket", None);
+        asked.facts.headers = Some(HeaderMap::new());
+        asked.facts.request_line = Some(&line);
+
+        assert_eq!(decision(PLAIN_TABLE, owned_core(), asked).await, Decision::Indeterminate);
+    }
+
+    #[tokio::test]
+    async fn a_core_denial_is_a_denial() {
+        let decided = decision(PLAIN_TABLE, ScriptedCore::answering(denied), facts("HeadBucket", Some("AKIDUSER0001"))).await;
+
+        assert_eq!(decided, Decision::Deny);
+    }
+
+    #[tokio::test]
+    async fn a_route_decidable_row_asks_the_core_its_legacy_action_for_the_verified_caller() {
+        let core = ScriptedCore::answering(|| Ok(()));
+
+        let decided = decision(PLAIN_TABLE, core.clone(), facts("HeadBucket", Some("AKIDUSER0001"))).await;
+
+        assert_eq!(decided, Decision::Allow);
+        assert_eq!(
+            *core.asked.lock().expect("core lock"),
+            [(
+                "HeadBucket",
+                Action::S3Action(S3Action::ListBucketAction),
+                Some("AKIDUSER0001".to_owned())
+            )]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_deferred_row_refuses_an_anonymous_caller_without_asking_the_core() {
+        let core = ScriptedCore::answering(|| Ok(()));
+
+        assert_eq!(
+            decision(AUTHORIZATION_TABLE, core.clone(), facts("ListBuckets", None)).await,
+            Decision::Deny
+        );
+        assert_eq!(
+            decision(AUTHORIZATION_TABLE, core.clone(), facts("ListBuckets", Some("AKIDUSER0001"))).await,
+            Decision::Allow,
+            "a verified caller passes to the hook"
+        );
+        assert!(
+            core.asked.lock().expect("core lock").is_empty(),
+            "a deferred row decides nothing through the core"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_no_check_row_is_allowed_only_because_its_row_says_so() {
+        let core = ScriptedCore::answering(denied);
+
+        assert_eq!(
+            decision(PLAIN_TABLE, core.clone(), facts("GetObjectTorrent", None)).await,
+            Decision::Allow
+        );
+        assert_eq!(decision(AUTHORIZATION_TABLE, core, facts("GetObjectTorrent", None)).await, Decision::Deny);
+    }
+
+    #[test]
+    fn the_deferred_rows_are_exactly_the_documented_ones() {
+        let deferred = rows_where(|stage| matches!(stage, RouteStage::DeferToHandlerHook { .. }));
+
+        assert_eq!(deferred, DOCUMENTED_DEFERRED_ROWS.iter().copied().collect::<BTreeSet<_>>());
+    }
+
+    #[test]
+    fn the_no_check_rows_are_exactly_the_documented_ones() {
+        let unchecked = rows_where(|stage| matches!(stage, RouteStage::LegacyNoCheck { .. }));
+
+        assert_eq!(unchecked, DOCUMENTED_NO_CHECK_ROWS.iter().copied().collect::<BTreeSet<_>>());
+    }
+
+    #[test]
+    fn every_row_states_every_column_and_names_one_operation() {
+        let mut seen = BTreeSet::new();
+        for row in AUTHORIZATION_TABLE {
+            assert!(seen.insert(row.operation), "{} has two rows", row.operation);
+            assert!(!row.operation.is_empty());
+            match row.route_stage {
+                RouteStage::DeferToHandlerHook { reason, .. } | RouteStage::LegacyNoCheck { reason } => {
+                    assert!(!reason.is_empty(), "{} defers without a reason", row.operation);
+                }
+                RouteStage::RouteDecidable => {}
+            }
+            match row.extra_checks {
+                ExtraChecks::Performed(checks) => {
+                    assert!(!checks.is_empty(), "{} lists no extra check; say `ExtraChecks::None`", row.operation);
+                    assert!(checks.iter().all(|check| !check.is_empty()), "{}", row.operation);
+                }
+                ExtraChecks::None => {}
+            }
+            assert!(row.body_inputs.iter().all(|input| !input.is_empty()), "{}", row.operation);
+        }
+    }
+
+    #[test]
+    fn list_buckets_asks_the_legacy_action_on_the_service() {
+        let row = AUTHORIZATION_TABLE
+            .iter()
+            .find(|row| row.operation == "ListBuckets")
+            .expect("ListBuckets is registered");
+
+        assert_eq!(row.action, Action::S3Action(S3Action::ListAllMyBucketsAction));
+        assert_eq!(row.resource, ResourceForm::Service);
+        assert_eq!(
+            row.route_stage,
+            RouteStage::DeferToHandlerHook {
+                gate: RouteGate::AuthenticatedCaller,
+                reason: "the use case decides per caller from the bucket inventory",
+            }
+        );
+        assert!(row.body_inputs.is_empty());
+        assert!(matches!(row.extra_checks, ExtraChecks::Performed(_)));
+    }
+}

rustfs/src/server/gateway_stack/tests.rs

@@ -0,0 +1,577 @@
+// 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 assembled gateway stack, driven through [`GatewayStack::call`] exactly as `server/http.rs`
+//! drives it. Requests that need object storage are in `rustfs/tests/embedded_gateway_stack_test.rs`.
+
+use super::authorizer::AUTHORIZATION_TABLE;
+use super::*;
+use bytes::Bytes;
+use http::{HeaderMap, Method, StatusCode};
+use http_body_util::{BodyExt, Empty};
+use rustfs_gateway::{AuthzRequest, ClientFacts, Decision, HostRequestId, InputAuthzRequest, InputDecisions, NO_DEADLINE};
+use std::collections::BTreeSet;
+use std::net::{IpAddr, SocketAddr};
+use std::sync::Mutex;
+use std::time::Instant;
+
+const ROOT_ACCESS_KEY: &str = "rustfsgatewayroot";
+const ROOT_SECRET_KEY: &str = "rustfs-gateway-root-secret";
+const REQUEST_ID: &str = "7c9e6679-7425-40de-944b-e07fc1f90ae7";
+const PEER: ([u8; 4], u16) = ([10, 0, 0, 7], 41000);
+const CLIENT: [u8; 4] = [203, 0, 113, 9];
+const EMPTY_SHA256: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+fn assembly() -> GatewayAssembly {
+    GatewayAssembly {
+        server_ctx: ServerContextSlot::new(),
+        root_access_key: ROOT_ACCESS_KEY.to_owned(),
+        root_secret_key: ROOT_SECRET_KEY.to_owned(),
+        region: Some("us-east-1".to_owned()),
+        server_domains: Vec::new(),
+        body_read_timeout: Duration::from_secs(300),
+        keystone_enabled: false,
+    }
+}
+
+/// An IAM whose readiness and answer are scripted.
+struct ScriptedIam {
+    ready: bool,
+}
+
+impl IamKeyLookup for ScriptedIam {
+    fn is_ready(&self) -> bool {
+        self.ready
+    }
+
+    fn check_key<'a>(&'a self, _access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>> {
+        Box::pin(async { Ok((None, false)) })
+    }
+}
+
+impl IamPrincipalLookup for ScriptedIam {
+    fn group_memberships(&self, _user: &str) -> Vec<String> {
+        Vec::new()
+    }
+}
+
+/// A settable live root, standing in for the server's runtime credentials.
+#[derive(Clone)]
+struct LiveRoot(Arc<Mutex<Option<Arc<RootCredentials>>>>);
+
+impl LiveRoot {
+    fn set(&self, root: RootCredentials) {
+        *self.0.lock().expect("root lock") = Some(Arc::new(root));
+    }
+
+    fn clear(&self) {
+        *self.0.lock().expect("root lock") = None;
+    }
+
+    fn source(&self) -> Arc<dyn RootSource> {
+        let cell = Arc::clone(&self.0);
+        Arc::new(move || cell.lock().expect("root lock").clone())
+    }
+}
+
+/// A live root holding the test root.
+fn root() -> LiveRoot {
+    let root = RootCredentials::new(ROOT_ACCESS_KEY, ROOT_SECRET_KEY, ROOT_SECRET_KEY).expect("a usable root");
+    LiveRoot(Arc::new(Mutex::new(Some(Arc::new(root)))))
+}
+
+/// The authorizer production installs: the registered table over the shared decision core.
+fn production_authorizer() -> RustfsAuthorizer {
+    RustfsAuthorizer::new(AUTHORIZATION_TABLE, Arc::new(OwnedDecisionCore::new(ServerContextSlot::new())))
+}
+
+/// The stack as production assembles it, over an IAM that knows no key but the root's.
+fn stack_over(iam: ScriptedIam) -> GatewayStack {
+    let authorizer = production_authorizer();
+    GatewayStack::assemble_with(assembly(), Arc::new(iam), root().source(), authorizer).expect("the stack assembles")
+}
+
+fn request(method: Method, target: &str) -> http::request::Builder {
+    Request::builder().method(method).uri(target).header("host", "localhost")
+}
+
+/// A well-formed SigV4 header signature for `access_key` whose signature is not the right one.
+fn signed_with_a_wrong_signature(builder: http::request::Builder, access_key: &str) -> http::request::Builder {
+    let now = jiff::Timestamp::now().strftime("%Y%m%dT%H%M%SZ").to_string();
+    let date = &now[..8];
+    builder
+        .header("x-amz-date", now.as_str())
+        .header("x-amz-content-sha256", EMPTY_SHA256)
+        .header(
+            "authorization",
+            format!(
+                "AWS4-HMAC-SHA256 Credential={access_key}/{date}/us-east-1/s3/aws4_request, \
+                 SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature={}",
+                "0".repeat(64)
+            ),
+        )
+}
+
+async fn answer(stack: &GatewayStack, builder: http::request::Builder) -> (StatusCode, HeaderMap, String) {
+    let response = stack.call(builder.body(Empty::<Bytes>::new()).expect("static request")).await;
+    let (parts, body) = response.into_parts();
+    let body = body.collect().await.expect("the answer is read").to_bytes();
+    (parts.status, parts.headers, String::from_utf8_lossy(&body).into_owned())
+}
+
+fn rustfs_request_context(request_id: &str) -> RequestContext {
+    RequestContext {
+        request_id: request_id.to_owned(),
+        x_amz_request_id: request_id.to_owned(),
+        trace_id: None,
+        span_id: None,
+        start_time: Instant::now(),
+    }
+}
+
+#[test]
+fn every_registered_operation_has_an_authorization_row_and_every_row_an_operation() {
+    let stack = stack_over(ScriptedIam { ready: true });
+
+    let registered: BTreeSet<&str> = stack.registered_operations().into_iter().collect();
+    let mapped: BTreeSet<&str> = AUTHORIZATION_TABLE.iter().map(|row| row.operation).collect();
+
+    assert_eq!(registered, mapped);
+    assert_eq!(registered, BTreeSet::from(["ListBuckets"]), "the walking skeleton serves one operation");
+}
+
+#[test]
+fn production_assembly_registers_what_the_test_assembly_does() {
+    let stack = GatewayStack::assemble(assembly()).expect("the stack assembles");
+
+    assert_eq!(stack.registered_operations(), ["ListBuckets"]);
+}
+
+#[test]
+fn an_unusable_region_fails_the_assembly() {
+    let refused = GatewayStack::assemble(GatewayAssembly {
+        region: Some("Not A Region".to_owned()),
+        ..assembly()
+    });
+
+    assert!(matches!(refused, Err(GatewayAssemblyError::Region { .. })), "{:?}", refused.err());
+}
+
+#[test]
+fn an_unusable_root_access_key_fails_the_assembly_without_naming_the_secret() {
+    let refused = GatewayStack::assemble(GatewayAssembly {
+        root_access_key: String::new(),
+        ..assembly()
+    });
+
+    let error = refused.err().expect("an empty access key verifies nothing");
+    assert!(
+        matches!(error, GatewayAssemblyError::RootCredentials(RootCredentialsError::InvalidAccessKey)),
+        "{error:?}"
+    );
+    for rendered in [error.to_string(), format!("{error:?}")] {
+        assert!(!rendered.contains(ROOT_SECRET_KEY), "{rendered}");
+    }
+}
+
+#[test]
+fn a_missing_or_short_root_secret_fails_the_assembly() {
+    for weak in ["", "1234567"] {
+        let refused = GatewayStack::assemble(GatewayAssembly {
+            root_secret_key: weak.to_owned(),
+            ..assembly()
+        });
+
+        assert!(
+            matches!(
+                refused.as_ref().err(),
+                Some(GatewayAssemblyError::RootCredentials(RootCredentialsError::WeakSecretKey))
+            ),
+            "{} bytes",
+            weak.len()
+        );
+    }
+}
+
+#[tokio::test]
+async fn a_root_change_reaches_the_next_request_and_a_withdrawn_root_signs_nothing() {
+    const ROTATED_ACCESS_KEY: &str = "rotatedgatewayroot";
+    let live = root();
+    let authorizer = production_authorizer();
+    let stack = GatewayStack::assemble_with(assembly(), Arc::new(ScriptedIam { ready: true }), live.source(), authorizer)
+        .expect("the stack assembles");
+    let signed_as = |access_key: &str| signed_with_a_wrong_signature(request(Method::GET, "/"), access_key);
+
+    let (_, _, known) = answer(&stack, signed_as(ROOT_ACCESS_KEY)).await;
+    assert!(known.contains("<Code>SignatureDoesNotMatch</Code>"), "the root key is known: {known}");
+
+    // The root access key changes: the old one is now just an unknown key, and the new one is
+    // known to the very next request.
+    live.set(RootCredentials::new(ROTATED_ACCESS_KEY, ROOT_SECRET_KEY, ROOT_SECRET_KEY).expect("a usable root"));
+    let (status, _, retired) = answer(&stack, signed_as(ROOT_ACCESS_KEY)).await;
+    assert_eq!(status, StatusCode::FORBIDDEN, "{retired}");
+    assert!(retired.contains("<Code>InvalidAccessKeyId</Code>"), "{retired}");
+    let (_, _, rotated) = answer(&stack, signed_as(ROTATED_ACCESS_KEY)).await;
+    assert!(
+        rotated.contains("<Code>SignatureDoesNotMatch</Code>"),
+        "the new root key is known: {rotated}"
+    );
+
+    // Injected: a root source that returns nothing, so the provider returns an error for the
+    // lookup. The key that was root a moment ago verifies nothing, and the fault is answered as
+    // one, `500`, as the legacy stack answers a failed lookup. Ledger entry R-068
+    // (rustfs/backlog#2684): the gateway's RustFS authenticator preset sets the switch of
+    // rustfs/gateway#1585; the assembly sets nothing of its own.
+    live.clear();
+    let (status, _, withdrawn) = answer(&stack, signed_as(ROTATED_ACCESS_KEY)).await;
+    assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{withdrawn}");
+    assert!(withdrawn.contains("<Code>InternalError</Code>"), "{withdrawn}");
+    assert!(!withdrawn.contains("ListAllMyBucketsResult"), "{withdrawn}");
+    assert_fault_answer_names_nothing(&withdrawn, ROTATED_ACCESS_KEY);
+}
+
+#[test]
+fn keystone_enabled_fails_the_assembly_and_names_both_settings() {
+    let refused = GatewayStack::assemble(GatewayAssembly {
+        keystone_enabled: true,
+        ..assembly()
+    });
+
+    let error = refused.err().expect("the gateway stack is not served with Keystone enabled");
+    assert!(matches!(error, GatewayAssemblyError::KeystoneEnabled), "{error:?}");
+    let message = error.to_string();
+    for named in ["RUSTFS_S3_STACK=gateway", "RUSTFS_KEYSTONE_ENABLE", "rustfs/gateway#1586"] {
+        assert!(message.contains(named), "{named} is missing from: {message}");
+    }
+}
+
+#[test]
+fn keystone_disabled_assembles() {
+    GatewayStack::assemble(GatewayAssembly {
+        keystone_enabled: false,
+        ..assembly()
+    })
+    .expect("the gateway stack assembles without Keystone");
+}
+
+#[test]
+fn keystone_is_read_from_configuration_and_an_unreadable_one_counts_as_enabled() {
+    const ENABLE: &str = "RUSTFS_KEYSTONE_ENABLE";
+    const AUTH_URL: &str = "RUSTFS_KEYSTONE_AUTH_URL";
+
+    temp_env::with_vars([(ENABLE, None::<&str>), (AUTH_URL, None)], || {
+        assert!(!keystone_enabled_in_configuration(), "unset is off");
+    });
+    temp_env::with_vars([(ENABLE, Some("false")), (AUTH_URL, None)], || {
+        assert!(!keystone_enabled_in_configuration(), "false is off");
+    });
+    temp_env::with_vars([(ENABLE, Some("true")), (AUTH_URL, Some("http://keystone.example.test:5000"))], || {
+        assert!(keystone_enabled_in_configuration(), "true is on");
+    });
+    temp_env::with_vars([(ENABLE, Some("true")), (AUTH_URL, None)], || {
+        assert!(keystone_enabled_in_configuration(), "on but unreadable is still on");
+    });
+}
+
+#[test]
+fn no_handler_runs_under_a_total_deadline_and_the_body_keeps_the_rustfs_idle_bound() {
+    let config = service_config(Duration::from_secs(300));
+
+    assert_eq!(config.handler_deadline(HandlerDeadlineClass::Standard), NO_DEADLINE);
+    assert_eq!(config.handler_deadline(HandlerDeadlineClass::Extended), NO_DEADLINE);
+    assert_eq!(config.commit_progress_deadline(), NO_DEADLINE);
+    assert_eq!(config.request_body_deadlines().read_idle(), Duration::from_secs(300));
+    assert_eq!(config.request_body_deadlines().first_byte(), Duration::from_secs(300));
+    assert_eq!(config.request_body_deadlines().throughput_window(), NO_DEADLINE);
+}
+
+#[test]
+fn a_disabled_body_read_timeout_lifts_the_idle_bound_instead_of_arming_zero() {
+    let deadlines = body_deadlines(Duration::ZERO);
+
+    assert_eq!(deadlines.read_idle(), NO_DEADLINE);
+    assert_eq!(deadlines.first_byte(), NO_DEADLINE);
+}
+
+#[test]
+fn host_facts_come_from_what_rustfs_layers_attached() {
+    let peer = SocketAddr::from(PEER);
+    let mut extensions = Extensions::new();
+    extensions.insert(Some(RemoteAddr(peer)));
+    extensions.insert(ClientInfo::direct(SocketAddr::new(IpAddr::from(CLIENT), 443)));
+    extensions.insert(NativeTlsTransport(true));
+    extensions.insert(rustfs_request_context(REQUEST_ID));
+
+    attach_host_facts(&mut extensions);
+
+    assert_eq!(extensions.get::<TransportSecurity>(), Some(&TransportSecurity::Encrypted));
+    assert_eq!(
+        extensions.get::<ClientAddr>().map(|address| address.ip()),
+        Some(IpAddr::from(CLIENT)),
+        "the trusted-proxy-corrected address wins over the socket peer"
+    );
+    assert_eq!(
+        extensions
+            .get::<HostRequestId>()
+            .map(|id| id.request_id().as_str().to_owned()),
+        Some(REQUEST_ID.to_owned())
+    );
+}
+
+#[test]
+fn host_facts_never_lean_towards_secure_or_invent_an_address() {
+    let mut bare = Extensions::new();
+    attach_host_facts(&mut bare);
+    assert_eq!(bare.get::<TransportSecurity>(), Some(&TransportSecurity::Plaintext));
+    assert!(bare.get::<ClientAddr>().is_none(), "no address is known");
+    assert!(bare.get::<HostRequestId>().is_none(), "no identifier is known");
+
+    let mut plaintext = Extensions::new();
+    plaintext.insert(Some(RemoteAddr(SocketAddr::from(PEER))));
+    plaintext.insert(NativeTlsTransport(false));
+    plaintext.insert(rustfs_request_context("not a request id"));
+    attach_host_facts(&mut plaintext);
+    assert_eq!(plaintext.get::<TransportSecurity>(), Some(&TransportSecurity::Plaintext));
+    assert_eq!(
+        plaintext.get::<ClientAddr>().map(|address| address.ip()),
+        Some(IpAddr::from(PEER.0)),
+        "without a proxy the socket peer is the client"
+    );
+    assert!(
+        plaintext.get::<HostRequestId>().is_none(),
+        "an identifier outside the alphabet is not handed over"
+    );
+
+    let mut peerless = Extensions::new();
+    peerless.insert(None::<RemoteAddr>);
+    attach_host_facts(&mut peerless);
+    assert!(peerless.get::<ClientAddr>().is_none());
+}
+
+/// Records what the gateway hands its authorizer about the client and the request line, and
+/// refuses.
+struct RecordingAuthorizer {
+    seen: Arc<Mutex<Option<Option<ClientFacts>>>>,
+    line: Arc<Mutex<Option<RequestLine>>>,
+}
+
+impl Authorizer for RecordingAuthorizer {
+    fn authorize_route<'a>(
+        &'a self,
+        context: &'a rustfs_gateway::RequestContext<'a>,
+        _request: &'a AuthzRequest<'a>,
+    ) -> BoxFuture<'a, Decision> {
+        *self.seen.lock().expect("recording lock") = Some(context.client().copied());
+        *self.line.lock().expect("recording lock") = context.server_extensions().get::<RequestLine>().cloned();
+        Box::pin(async { Decision::Deny })
+    }
+
+    fn authorize_input<'a>(
+        &'a self,
+        _context: &'a rustfs_gateway::RequestContext<'a>,
+        request: &'a InputAuthzRequest<'a>,
+    ) -> BoxFuture<'a, InputDecisions> {
+        let decisions = request.decide_all(Decision::Deny, |_| Decision::Deny);
+        Box::pin(async move { decisions })
+    }
+}
+
+#[tokio::test]
+async fn the_authorizer_sees_the_client_facts_the_host_attached() {
+    let seen = Arc::new(Mutex::new(None));
+    let line = Arc::new(Mutex::new(None));
+    let authorizer = RecordingAuthorizer {
+        seen: Arc::clone(&seen),
+        line: Arc::clone(&line),
+    };
+    let stack = GatewayStack::assemble_with(assembly(), Arc::new(ScriptedIam { ready: true }), root().source(), authorizer)
+        .expect("assembles");
+    let proxied = request(Method::GET, "/?prefix=probe")
+        .extension(Some(RemoteAddr(SocketAddr::from(PEER))))
+        .extension(ClientInfo::direct(SocketAddr::new(IpAddr::from(CLIENT), 443)))
+        .extension(NativeTlsTransport(true));
+
+    let (status, _, _) = answer(&stack, proxied).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    let facts = seen
+        .lock()
+        .expect("recording lock")
+        .take()
+        .expect("the authorizer was asked")
+        .expect("the host attached client facts");
+    assert!(facts.transport_secure, "the listener terminated TLS");
+    assert_eq!(facts.client_ip, Some(IpAddr::from(CLIENT)));
+    assert_eq!(facts.source_ip(), Some(IpAddr::from(CLIENT)));
+    let line = line
+        .lock()
+        .expect("recording lock")
+        .take()
+        .expect("the host attached the request line");
+    assert_eq!(line.method, Method::GET);
+    assert_eq!(line.uri.path(), "/");
+    assert_eq!(line.uri.query(), Some("prefix=probe"));
+}
+
+#[tokio::test]
+async fn the_authorizer_sees_a_cleartext_transport_as_cleartext() {
+    let seen = Arc::new(Mutex::new(None));
+    let line = Arc::new(Mutex::new(None));
+    let authorizer = RecordingAuthorizer {
+        seen: Arc::clone(&seen),
+        line: Arc::clone(&line),
+    };
+    let stack = GatewayStack::assemble_with(assembly(), Arc::new(ScriptedIam { ready: true }), root().source(), authorizer)
+        .expect("assembles");
+
+    let (status, _, _) = answer(&stack, request(Method::GET, "/").extension(NativeTlsTransport(false))).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    let facts = seen
+        .lock()
+        .expect("recording lock")
+        .take()
+        .expect("the authorizer was asked")
+        .expect("the transport declaration alone makes the facts present");
+    assert!(!facts.transport_secure);
+    assert_eq!(facts.client_ip, None);
+}
+
+#[tokio::test]
+async fn an_anonymous_list_buckets_is_denied() {
+    let stack = stack_over(ScriptedIam { ready: true });
+
+    let (status, _, body) = answer(&stack, request(Method::GET, "/")).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert!(body.contains("<Code>AccessDenied</Code>"), "{body}");
+}
+
+#[tokio::test]
+async fn an_unregistered_operation_is_not_implemented() {
+    let stack = stack_over(ScriptedIam { ready: true });
+
+    for (method, target) in [
+        (Method::GET, "/bucket/key"),
+        (Method::PUT, "/bucket"),
+        (Method::DELETE, "/bucket/key"),
+    ] {
+        let (status, _, body) = answer(&stack, request(method.clone(), target)).await;
+
+        assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{method} {target}: {body}");
+        assert!(body.contains("<Code>NotImplemented</Code>"), "{method} {target}: {body}");
+    }
+}
+
+#[tokio::test]
+async fn a_wrong_signature_is_a_signature_mismatch() {
+    let stack = stack_over(ScriptedIam { ready: true });
+
+    let (status, _, body) = answer(&stack, signed_with_a_wrong_signature(request(Method::GET, "/"), ROOT_ACCESS_KEY)).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert!(body.contains("<Code>SignatureDoesNotMatch</Code>"), "{body}");
+    assert!(!body.contains(ROOT_SECRET_KEY), "{body}");
+}
+
+#[tokio::test]
+async fn an_unknown_access_key_is_refused_as_the_caller_s_fault() {
+    let stack = stack_over(ScriptedIam { ready: true });
+
+    let (status, _, body) = answer(&stack, signed_with_a_wrong_signature(request(Method::GET, "/"), "AKIDUNKNOWN0001")).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN, "{body}");
+    assert!(body.contains("<Code>InvalidAccessKeyId</Code>"), "{body}");
+}
+
+/// The answer to a credential-store fault names neither the key that was looked up, nor a secret,
+/// nor why the lookup failed.
+fn assert_fault_answer_names_nothing(body: &str, access_key: &str) {
+    assert!(!body.contains(access_key), "the looked-up key is not echoed: {body}");
+    assert!(!body.contains(ROOT_SECRET_KEY), "{body}");
+    let lowered = body.to_ascii_lowercase();
+    for detail in ["iam", "credential", "lookup", "store", "not ready", "unavailable", "root"] {
+        assert!(!lowered.contains(detail), "the answer says `{detail}`: {body}");
+    }
+}
+
+/// Injected: a credential provider that returns an error for the lookup of a non-root key.
+///
+/// The fault is answered as one, `500 InternalError`, as the legacy stack answers a failed lookup,
+/// and not as an unknown key. Up to gateway pin 21283e87 this test observed `403
+/// InvalidAccessKeyId`; pin 7915d069 changed the answer. Ledger entry R-068 (rustfs/backlog#2684): the gateway's RustFS
+/// authenticator preset sets the switch of rustfs/gateway#1585; the assembly sets nothing of its
+/// own. Nothing is served either way.
+#[tokio::test]
+async fn a_credential_lookup_that_fails_is_refused_and_never_served() {
+    let stack = stack_over(ScriptedIam { ready: false });
+
+    let (status, _, body) = answer(&stack, signed_with_a_wrong_signature(request(Method::GET, "/"), "AKIDUSER0001")).await;
+
+    assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{body}");
+    assert!(body.contains("<Code>InternalError</Code>"), "{body}");
+    assert!(!body.contains("InvalidAccessKeyId"), "a store fault is not an unknown key: {body}");
+    assert_fault_answer_names_nothing(&body, "AKIDUSER0001");
+    assert!(!body.contains("ListAllMyBucketsResult"), "{body}");
+}
+
+#[tokio::test]
+async fn the_root_key_verifies_while_iam_is_unavailable() {
+    let stack = stack_over(ScriptedIam { ready: false });
+
+    let (status, _, body) = answer(&stack, signed_with_a_wrong_signature(request(Method::GET, "/"), ROOT_ACCESS_KEY)).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN, "the root secret is known without IAM: {body}");
+    assert!(body.contains("<Code>SignatureDoesNotMatch</Code>"), "{body}");
+}
+
+#[tokio::test]
+async fn the_answer_carries_the_request_identifier_rustfs_assigned() {
+    let stack = stack_over(ScriptedIam { ready: true });
+
+    let (_, headers, _) = answer(&stack, request(Method::GET, "/").extension(rustfs_request_context(REQUEST_ID))).await;
+
+    assert_eq!(headers.get("x-amz-request-id").and_then(|value| value.to_str().ok()), Some(REQUEST_ID));
+    assert_eq!(headers.get("x-request-id").and_then(|value| value.to_str().ok()), Some(REQUEST_ID));
+}
+
+#[tokio::test]
+async fn without_a_rustfs_identifier_the_gateway_mints_its_own() {
+    let stack = stack_over(ScriptedIam { ready: true });
+
+    let (_, headers, _) = answer(&stack, request(Method::GET, "/")).await;
+
+    let minted = headers
+        .get("x-amz-request-id")
+        .and_then(|value| value.to_str().ok())
+        .expect("every answer names its request");
+    assert!(!minted.is_empty());
+    assert_ne!(minted, REQUEST_ID);
+}
+
+#[tokio::test]
+async fn the_gateway_writes_no_cors_header() {
+    let stack = stack_over(ScriptedIam { ready: true });
+    let browser = |builder: http::request::Builder| builder.header("origin", "https://app.example.test");
+
+    for (method, target) in [(Method::GET, "/"), (Method::GET, "/bucket/key"), (Method::PUT, "/bucket/key")] {
+        let (_, headers, _) = answer(&stack, browser(request(method.clone(), target))).await;
+
+        let cors: Vec<&str> = headers
+            .keys()
+            .map(http::HeaderName::as_str)
+            .filter(|name| name.starts_with("access-control-"))
+            .collect();
+        assert!(cors.is_empty(), "{method} {target}: {cors:?}");
+    }
+}

rustfs/src/server/http.rs

@@ -18,6 +18,9 @@ use crate::app::object::request_body::{BodyReadControl, ObservedBody};
 use crate::auth::IAMAuth;
 use crate::auth_keystone;
 use crate::config;
+#[cfg(feature = "gateway-stack")]
+use crate::server::gateway_stack::{GatewayAssembly, GatewayStack, keystone_enabled_in_configuration};
+use crate::server::stack::{S3StackService, select_stack};
 use crate::server::{
     Http12ShutdownOutcome, ReadinessGateLayer, RemoteAddr, ShutdownHandle,
     compress::{HttpCompressionConfig, PathAwareHttpCompressionPredicate, PathCategoryInjectionLayer},
@@ -259,6 +262,7 @@ const EVENT_HTTP_TRANSPORT_FAILED: &str = "http_transport_failed";
 const EVENT_SOCKET_FALLBACK: &str = "socket_fallback";
 const EVENT_HTTP_BIND_FAILED: &str = "http_bind_failed";
 const EVENT_HTTP_STARTUP_ENDPOINTS: &str = "http_startup_endpoints";
+const EVENT_S3_STACK_SELECTED: &str = "s3_stack_selected";
 const EVENT_HTTP_HOST_ROUTING: &str = "http_host_routing";
 const EVENT_HTTP_COMPRESSION_STATE: &str = "http_compression_state";
 const EVENT_API_RATE_LIMIT_STATE: &str = "api_rate_limit_state";
@@ -1460,6 +1464,41 @@ fn control_request_body(body: s3s::Body) -> ControlRequestBody {
     ControlRequestBody(body)
 }
 
+/// The S3 service mounted under the middleware: the legacy stack or the gateway stack, selected
+/// at startup (`server/stack.rs`).
+type S3EdgeService = S3StackService<S3Service, GatewayEdgeService>;
+
+/// A build without the `gateway-stack` feature has no gateway stack to select, so this side of
+/// [`S3EdgeService`] is never constructed.
+#[cfg(not(feature = "gateway-stack"))]
+type GatewayEdgeService = S3Service;
+
+/// The gateway stack where the legacy service is mounted: the same response and error types, so
+/// the layers around it do not change with the stack.
+#[cfg(feature = "gateway-stack")]
+#[derive(Clone)]
+struct GatewayEdgeService(GatewayStack);
+
+#[cfg(feature = "gateway-stack")]
+impl<B> Service<HttpRequest<B>> for GatewayEdgeService
+where
+    B: http_body::Body<Data = Bytes> + Send + 'static,
+    B::Error: std::error::Error + Send + Sync + 'static,
+{
+    type Response = Response<s3s::Body>;
+    type Error = s3s::HttpError;
+    type Future = Pin<Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, request: HttpRequest<B>) -> Self::Future {
+        let answer = self.0.call(request);
+        Box::pin(async move { Ok(answer.await.map(s3s::Body::http_body_unsync)) })
+    }
+}
+
 #[derive(Clone)]
 struct ControlBodyDigestService<S> {
     inner: S,
@@ -2452,6 +2491,48 @@ async fn start_http_server_inner(
         rustfs_config::ENV_HTTP_REQUEST_BODY_READ_TIMEOUT,
         rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT,
     ));
+
+    // The S3 stack is selected once, for the whole server. A gateway stack that cannot be
+    // assembled fails startup here; the legacy stack is never started in its place.
+    #[cfg(feature = "gateway-stack")]
+    let assemble_gateway = || -> Result<GatewayEdgeService> {
+        let stack = GatewayStack::assemble(GatewayAssembly {
+            server_ctx: Arc::clone(&server_ctx),
+            root_access_key: config.access_key.clone(),
+            root_secret_key: config.secret_key.clone(),
+            region: config.region.clone(),
+            server_domains: vh_domain_hosts.to_vec(),
+            body_read_timeout: http_request_body_read_timeout,
+            keystone_enabled: keystone_enabled_in_configuration(),
+        })
+        .map_err(Error::other)?;
+        info!(
+            event = EVENT_S3_STACK_SELECTED,
+            component = LOG_COMPONENT_SERVER,
+            subsystem = LOG_SUBSYSTEM_STARTUP,
+            stack = "gateway",
+            registered_operations = stack.registered_operations().len(),
+            "S3 stack selected"
+        );
+        Ok(GatewayEdgeService(stack))
+    };
+    #[cfg(not(feature = "gateway-stack"))]
+    let assemble_gateway = || {
+        Err(Error::new(
+            std::io::ErrorKind::Unsupported,
+            "the gateway S3 stack is not built into this binary",
+        ))
+    };
+    let s3_service: S3EdgeService = select_stack(config.s3_stack, s3_service, assemble_gateway)?;
+    if s3_service.selected() == crate::config::S3Stack::Legacy {
+        info!(
+            event = EVENT_S3_STACK_SELECTED,
+            component = LOG_COMPONENT_SERVER,
+            subsystem = LOG_SUBSYSTEM_STARTUP,
+            stack = "legacy",
+            "S3 stack selected"
+        );
+    }
     #[cfg(feature = "http3")]
     let http3_server = if let Some(server_config) = http3_server_config {
         let h3_s3_service = s3_service.clone();
@@ -2889,7 +2970,7 @@ async fn start_http_server_inner(
 struct ConnectionContext {
     http_server: Arc<ConnBuilder<HttpTaskExecutor>>,
     task_owner: HttpTaskExecutor,
-    s3_service: S3Service,
+    s3_service: S3EdgeService,
     compression_config: HttpCompressionConfig,
     is_console: bool,
     /// Whether `RUSTFS_SERVER_DOMAINS` is configured (i.e. s3s virtual-hosted-style routing is active).

rustfs/src/server/mod.rs

@@ -16,6 +16,10 @@ mod audit;
 mod compress;
 pub mod cors;
 mod event;
+#[cfg(feature = "gateway-stack")]
+mod gateway_handlers;
+#[cfg(feature = "gateway-stack")]
+mod gateway_stack;
 mod health;
 mod http;
 mod hybrid;
@@ -29,6 +33,7 @@ mod runtime;
 pub(crate) mod runtime_sources;
 mod service_state;
 mod ssec_transport;
+mod stack;
 pub mod tls_material;
 pub(crate) mod tls_publish;
 

rustfs/src/server/ssec_transport.rs

@@ -92,7 +92,7 @@ fn carries_ssec_headers(headers: &HeaderMap) -> bool {
 /// protocol is only consulted as a second source because the trusted-proxy
 /// layer has already decided whether the peer is allowed to assert it; a
 /// request that arrives direct carries no such assertion.
-fn is_secure_transport(connection_is_tls: bool, client_info: Option<&ClientInfo>) -> bool {
+pub(crate) fn is_secure_transport(connection_is_tls: bool, client_info: Option<&ClientInfo>) -> bool {
     if connection_is_tls {
         return true;
     }

rustfs/src/server/stack.rs

@@ -0,0 +1,253 @@
+// 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 S3 protocol stack one server runs, selected once at startup (rustfs/backlog#2734 T2.5).
+//!
+//! Responsible for: [`S3StackService`], the service mounted where the legacy S3 service used to
+//! be, and [`select_stack`], the only place a stack is chosen. The whole stack is one or the
+//! other: no request is routed between stacks by its operation, and a request the gateway stack
+//! refuses is answered by that refusal and never retried on the legacy stack.
+//! NOT responsible for: assembling either stack (`server/http.rs` builds the legacy service,
+//! `server/gateway_stack.rs` the gateway one), or the middleware around them.
+//! Upstream: `server/http.rs`. Downstream: the legacy S3 service and the gateway stack.
+//!
+//! This file names neither stack's types: it is generic over two services with one response
+//! type, so `server/http.rs` decides what they are and how their bodies meet.
+
+use super::layer::is_sts_query_request;
+use super::prefix::{HEALTH_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, has_path_prefix, is_admin_path};
+use crate::admin::console::is_console_path;
+use crate::config::S3Stack;
+use http::{HeaderMap, Method, Request, Uri};
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll, ready};
+use tower::Service;
+
+/// The MinIO-compatible health probes share this prefix.
+const MINIO_HEALTH_PREFIX: &str = "/minio/health";
+
+/// The S3 service of one server: the legacy stack, or the gateway stack with the legacy router
+/// kept for the routes that are not S3 operations.
+#[derive(Clone)]
+pub(crate) struct S3StackService<L, G> {
+    legacy: L,
+    gateway: Option<G>,
+}
+
+impl<L, G> S3StackService<L, G> {
+    /// The stack this service runs.
+    pub(crate) fn selected(&self) -> S3Stack {
+        match self.gateway {
+            Some(_) => S3Stack::Gateway,
+            None => S3Stack::Legacy,
+        }
+    }
+}
+
+/// Builds the service for the selected stack.
+///
+/// `assemble_gateway` runs only when the gateway stack is selected, and its failure is the
+/// failure of the whole selection: the caller refuses to start.
+///
+/// # Errors
+///
+/// Whatever `assemble_gateway` refused with. The legacy stack is never substituted.
+pub(crate) fn select_stack<L, G, E>(
+    stack: S3Stack,
+    legacy: L,
+    assemble_gateway: impl FnOnce() -> Result<G, E>,
+) -> Result<S3StackService<L, G>, E> {
+    let gateway = match stack {
+        S3Stack::Legacy => None,
+        S3Stack::Gateway => Some(assemble_gateway()?),
+    };
+    Ok(S3StackService { legacy, gateway })
+}
+
+/// Whether the legacy router answers this request while the gateway stack is selected: the admin
+/// and table-catalog APIs, the console, STS, health and profiling, none of which is an S3
+/// operation. Every other request is the gateway's, whatever it answers.
+fn is_legacy_only_route(method: &Method, uri: &Uri, headers: &HeaderMap) -> bool {
+    let path = uri.path();
+    is_admin_path(path)
+        || is_console_path(path)
+        || has_path_prefix(path, HEALTH_PREFIX)
+        || has_path_prefix(path, MINIO_HEALTH_PREFIX)
+        || path == PROFILE_CPU_PATH
+        || path == PROFILE_MEMORY_PATH
+        || is_sts_query_request(method, uri, headers)
+}
+
+impl<L, G, B, R, E> Service<Request<B>> for S3StackService<L, G>
+where
+    L: Service<Request<B>, Response = http::Response<R>, Error = E>,
+    G: Service<Request<B>, Response = http::Response<R>, Error = E>,
+    L::Future: Send + 'static,
+    G::Future: Send + 'static,
+{
+    type Response = http::Response<R>;
+    type Error = E;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if let Some(gateway) = &mut self.gateway {
+            ready!(gateway.poll_ready(cx))?;
+        }
+        self.legacy.poll_ready(cx)
+    }
+
+    fn call(&mut self, request: Request<B>) -> Self::Future {
+        match &mut self.gateway {
+            // REMOVED BY T2.11: the gateway serves these routes itself once they are migrated.
+            Some(gateway) if !is_legacy_only_route(request.method(), request.uri(), request.headers()) => {
+                Box::pin(gateway.call(request))
+            }
+            _ => Box::pin(self.legacy.call(request)),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use http::{Response, StatusCode};
+    use std::convert::Infallible;
+    use std::future::{Ready, ready as ready_future};
+
+    /// Answers every request with its own name.
+    #[derive(Clone)]
+    struct Named(&'static str);
+
+    impl Service<Request<()>> for Named {
+        type Response = Response<&'static str>;
+        type Error = Infallible;
+        type Future = Ready<Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, _request: Request<()>) -> Self::Future {
+            ready_future(Ok(Response::new(self.0)))
+        }
+    }
+
+    fn request(method: Method, target: &str) -> Request<()> {
+        Request::builder()
+            .method(method)
+            .uri(target)
+            .body(())
+            .expect("static request")
+    }
+
+    async fn answered_by(service: &mut S3StackService<Named, Named>, request: Request<()>) -> &'static str {
+        let response = service.call(request).await.expect("infallible");
+        assert_eq!(response.status(), StatusCode::OK);
+        response.into_body()
+    }
+
+    fn gateway_stack() -> S3StackService<Named, Named> {
+        select_stack(S3Stack::Gateway, Named("legacy"), || Ok::<_, Infallible>(Named("gateway"))).expect("assembles")
+    }
+
+    #[test]
+    fn a_gateway_assembly_failure_fails_the_selection_instead_of_falling_back() {
+        let selection = select_stack::<_, Named, _>(S3Stack::Gateway, Named("legacy"), || Err("the assembly was refused"));
+
+        assert_eq!(selection.err(), Some("the assembly was refused"));
+    }
+
+    #[test]
+    fn the_legacy_selection_never_assembles_a_gateway() {
+        let service = select_stack::<_, Named, &str>(S3Stack::Legacy, Named("legacy"), || {
+            panic!("the legacy stack must not assemble the gateway")
+        })
+        .expect("legacy needs no assembly");
+
+        assert_eq!(service.selected(), S3Stack::Legacy);
+    }
+
+    #[tokio::test]
+    async fn the_legacy_stack_answers_every_request_itself() {
+        let mut service = select_stack(S3Stack::Legacy, Named("legacy"), || Ok::<_, Infallible>(Named("gateway")))
+            .expect("legacy needs no assembly");
+
+        for target in ["/", "/bucket/key", "/rustfs/admin/v3/info", "/health"] {
+            assert_eq!(answered_by(&mut service, request(Method::GET, target)).await, "legacy", "{target}");
+        }
+    }
+
+    #[tokio::test]
+    async fn the_gateway_stack_answers_every_s3_request_whatever_the_operation() {
+        let mut service = gateway_stack();
+        assert_eq!(service.selected(), S3Stack::Gateway);
+
+        for (method, target) in [
+            (Method::GET, "/"),
+            (Method::GET, "/bucket/key"),
+            (Method::PUT, "/bucket/key"),
+            (Method::DELETE, "/bucket"),
+            (Method::GET, "/bucket?versions"),
+            (Method::POST, "/bucket/key?uploads"),
+            // Prefixes of the legacy-only routes that are bucket names, not those routes.
+            (Method::GET, "/healthy/key"),
+            (Method::GET, "/rustfs/administrator"),
+            (Method::GET, "/minio/healthcheck"),
+            (Method::GET, "/profile/cpus"),
+        ] {
+            assert_eq!(
+                answered_by(&mut service, request(method.clone(), target)).await,
+                "gateway",
+                "{method} {target}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn the_gateway_stack_leaves_the_non_s3_routes_to_the_legacy_router() {
+        let mut service = gateway_stack();
+
+        for target in [
+            "/rustfs/admin/v3/info",
+            "/minio/admin/v3/info",
+            "/iceberg/v1/config",
+            "/health",
+            "/health/ready",
+            "/minio/health/live",
+            "/profile/cpu",
+            "/profile/memory",
+        ] {
+            assert_eq!(answered_by(&mut service, request(Method::GET, target)).await, "legacy", "{target}");
+        }
+    }
+
+    #[tokio::test]
+    async fn an_sts_form_post_stays_on_the_legacy_router_and_a_plain_post_does_not() {
+        let mut service = gateway_stack();
+        let sts = Request::builder()
+            .method(Method::POST)
+            .uri("/")
+            .header(http::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
+            .body(())
+            .expect("static request");
+
+        assert!(
+            is_sts_query_request(sts.method(), sts.uri(), sts.headers()),
+            "the fixture is an STS request"
+        );
+        assert_eq!(answered_by(&mut service, sts).await, "legacy");
+        assert_eq!(answered_by(&mut service, request(Method::POST, "/")).await, "gateway");
+    }
+}

rustfs/src/startup_server.rs

@@ -41,7 +41,8 @@ const EVENT_SERVER_STARTING: &str = "server_starting";
 const EVENT_ACTION_CREDENTIALS_INITIALIZED: &str = "action_credentials_initialized";
 const EVENT_ACTION_CREDENTIALS_INITIALIZATION_FAILED: &str = "action_credentials_initialization_failed";
 const DEFAULT_CREDENTIALS_WARNING_MESSAGE: &str = "Detected default root credentials; set RUSTFS_ACCESS_KEY and RUSTFS_SECRET_KEY to non-default values for production deployments";
-const GATEWAY_STACK_NOT_ASSEMBLED: &str = "gateway stack is not assembled yet (rustfs/backlog#2734 T2.5)";
+#[cfg(not(feature = "gateway-stack"))]
+const GATEWAY_STACK_NOT_BUILT: &str = "RUSTFS_S3_STACK=gateway needs a binary built with the `gateway-stack` feature, which no release build enables (rustfs/backlog#2734)";
 
 pub(crate) struct StartupListenContext {
     pub(crate) readiness: Arc<GlobalReadiness>,
@@ -82,12 +83,7 @@ pub(crate) async fn init_startup_listen_context(
     config: &Config,
     instance_ctx: &Arc<InstanceContext>,
 ) -> Result<StartupListenContext> {
-    match config.s3_stack {
-        S3Stack::Legacy => {}
-        // REPLACED BY T2.5
-        // Refuse before any listener or storage exists; never fall back to legacy.
-        S3Stack::Gateway => return Err(Error::new(ErrorKind::Unsupported, GATEWAY_STACK_NOT_ASSEMBLED)),
-    }
+    refuse_unbuilt_stack(config.s3_stack)?;
     crate::server::init_console_prefix()?;
     log_sanitized_server_config(config);
     let readiness = Arc::new(GlobalReadiness::new());
@@ -138,6 +134,18 @@ pub(crate) async fn init_startup_listen_context(
     })
 }
 
+/// Refuses a selected S3 stack this binary cannot serve, before any listener or storage exists.
+/// The selection is never replaced by another stack.
+fn refuse_unbuilt_stack(stack: S3Stack) -> Result<()> {
+    match stack {
+        S3Stack::Legacy => Ok(()),
+        #[cfg(feature = "gateway-stack")]
+        S3Stack::Gateway => Ok(()),
+        #[cfg(not(feature = "gateway-stack"))]
+        S3Stack::Gateway => Err(Error::new(ErrorKind::Unsupported, GATEWAY_STACK_NOT_BUILT)),
+    }
+}
+
 pub(crate) async fn prepare_embedded_startup_config(
     address: String,
     access_key: String,
@@ -166,6 +174,10 @@ pub(crate) async fn prepare_embedded_startup_config(
     config.secret_key = secret_key.clone();
     config.region = Some(region.clone());
     config.console_enable = false;
+    // The stack is selected for the whole process, so an embedded server reads the same variable
+    // the binary does.
+    config.s3_stack = S3Stack::from_env()?;
+    refuse_unbuilt_stack(config.s3_stack)?;
 
     Ok(EmbeddedStartupConfig {
         config,
@@ -518,6 +530,26 @@ mod tests {
         ));
     }
 
+    #[cfg(not(feature = "gateway-stack"))]
+    #[test]
+    fn a_binary_without_the_gateway_stack_refuses_to_select_it() {
+        let err = super::refuse_unbuilt_stack(crate::config::S3Stack::Gateway).expect_err("the gateway stack is not built");
+
+        assert_eq!(err.kind(), std::io::ErrorKind::Unsupported);
+        assert!(err.to_string().contains("gateway-stack"), "{err}");
+    }
+
+    #[cfg(feature = "gateway-stack")]
+    #[test]
+    fn a_binary_with_the_gateway_stack_accepts_its_selection() {
+        super::refuse_unbuilt_stack(crate::config::S3Stack::Gateway).expect("the gateway stack is built");
+    }
+
+    #[test]
+    fn the_legacy_stack_is_always_selectable() {
+        super::refuse_unbuilt_stack(crate::config::S3Stack::Legacy).expect("legacy is the default stack");
+    }
+
     #[tokio::test]
     async fn init_startup_listen_context_rejects_console_s3_port_conflict() {
         let mut config = Config::new("127.0.0.1:9001", vec!["/tmp/rustfs-data".to_string()]);

rustfs/src/storage/access.rs

@@ -1357,7 +1357,12 @@ fn object_store_not_initialized_error() -> S3Error {
 }
 
 fn request_iam_store<R: AuthorizationRequest + ?Sized>(req: &R) -> S3Result<Arc<IamSys<ObjectStore>>> {
-    let iam_store = match req.server_context() {
+    iam_store_for(req.server_context())
+}
+
+/// The IAM store of `server_ctx`, or the process default when the request carries no server context.
+fn iam_store_for(server_ctx: Option<&Arc<ServerContextSlot>>) -> S3Result<Arc<IamSys<ObjectStore>>> {
+    let iam_store = match server_ctx {
         Some(server_ctx) => server_ctx
             .installed_app_context()
             .filter(|context| context.iam().is_ready())
@@ -1373,6 +1378,86 @@ fn request_iam_store<R: AuthorizationRequest + ?Sized>(req: &R) -> S3Result<Arc<
     })
 }
 
+/// Resolves the caller of one request as the access phase of either edge needs it: the credential
+/// IAM holds for `access_key` (validated against `session_token`), whether it owns the deployment,
+/// and the request-scoped facts later authorization steps read.
+///
+/// `access_key` is the key whose signature the edge already verified; `None` is an anonymous
+/// request. Resolved against `server_ctx` so each embedded server reads its own IAM domain.
+///
+/// # Errors
+///
+/// What [`check_key_valid_with_context`] refuses: an unknown, disabled or expired credential, a
+/// session token that does not belong to it, or an IAM system that is not initialized.
+pub(crate) async fn resolve_req_info(
+    server_ctx: &Arc<ServerContextSlot>,
+    access_key: Option<&str>,
+    session_token: &str,
+    request_context: Option<RequestContext>,
+) -> S3Result<ReqInfo> {
+    let app_context = server_ctx.app_context();
+    let (cred, is_owner) = match access_key {
+        Some(access_key) => {
+            let (cred, is_owner) = check_key_valid_with_context(session_token, access_key, app_context.as_deref()).await?;
+            (Some(cred), is_owner)
+        }
+        None => (None, false),
+    };
+    let region = match app_context.as_deref() {
+        Some(context) => context.region().get(),
+        None => runtime_sources::current_region(),
+    };
+    Ok(ReqInfo {
+        cred,
+        is_owner,
+        region,
+        request_context,
+        ..Default::default()
+    })
+}
+
+/// The operation-aware admission checks every S3 request passes after its caller is resolved and
+/// before its own authorization: the signed upload-size capabilities are refused on any operation
+/// but the one each is scoped to, and the license must be valid.
+///
+/// `operation` is the S3 operation name (`PutObject`). Both edges call this with the request they
+/// accepted, so a capability query cannot be silently ignored on one of them.
+///
+/// # Errors
+///
+/// `InvalidRequest` for a misplaced capability query, `AccessDenied` for a refused license, and
+/// `InternalError` when the license cannot be checked.
+pub(crate) fn admit_operation(headers: &HeaderMap, raw_query: Option<&str>, operation: &str) -> S3Result<()> {
+    let auth_type = get_request_auth_type_with_query(headers, raw_query);
+    let verified_presigned = matches!(auth_type, AuthType::Presigned);
+    let verified_sigv4 = matches!(auth_type, AuthType::Presigned | AuthType::Signed);
+    // The size capability is intentionally scoped to the single-object
+    // PutObject operation. Validate this at the operation-aware access
+    // boundary so unsupported GET/HEAD/DELETE/bucket routes cannot silently
+    // ignore a signed capability query.
+    if parse_presigned_put_max_content_length(headers, raw_query, verified_presigned)?.is_some() && operation != "PutObject" {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
+        ));
+    }
+    if parse_presigned_multipart_max_total_object_size(headers, raw_query, verified_sigv4)?.is_some()
+        && operation != "CreateMultipartUpload"
+    {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} is only supported for CreateMultipartUpload"),
+        ));
+    }
+    license_check().map_err(|er| match er.kind() {
+        std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"),
+        _ => {
+            tracing::error!("license check failed due to unexpected error: {er}");
+            s3_error!(InternalError, "License validation failed")
+        }
+    })
+}
+
 pub(crate) struct ListBucketsIamAuthorization {
     iam_store: Arc<IamSys<ObjectStore>>,
     prepared: PreparedIamAuth,
@@ -1410,22 +1495,46 @@ impl ListBucketsIamAuthorization {
     }
 }
 
+/// The request facts the ListBuckets discovery decision reads.
+///
+/// Owned by no HTTP framework: the legacy s3s edge and the gateway edge both fill it from their own
+/// request, so the decision cannot differ by the edge that accepted the request.
+pub(crate) struct ListBucketsAuthzRequest<'a> {
+    /// The caller as the access phase resolved it.
+    pub(crate) req_info: &'a ReqInfo,
+    pub(crate) headers: &'a HeaderMap,
+    /// The query exactly as it arrived, without its `?`.
+    pub(crate) raw_query: Option<&'a str>,
+    pub(crate) remote_addr: Option<std::net::SocketAddr>,
+    pub(crate) client_info: Option<&'a ClientInfo>,
+    pub(crate) server_context: Option<&'a Arc<ServerContextSlot>>,
+}
+
 /// Prepare the IAM-only authorization used to decide which buckets are visible in ListBuckets.
 /// Bucket policies are intentionally excluded from this discovery decision, matching MinIO.
-pub(crate) async fn prepare_list_buckets_iam_authorization<T>(req: &S3Request<T>) -> S3Result<ListBucketsIamAuthorization> {
-    let req_info = req_info_ref(req)?;
+///
+/// # Errors
+///
+/// `AccessDenied` for an anonymous caller, and `InternalError` when the IAM store is not ready.
+pub(crate) async fn prepare_list_buckets_iam_authorization(
+    req: ListBucketsAuthzRequest<'_>,
+) -> rustfs_s3_types::S3Result<ListBucketsIamAuthorization> {
+    let req_info = req.req_info;
     let Some(cred) = req_info.cred.as_ref() else {
         return Err(ApiError::access_denied().into());
     };
-    let iam_store = request_iam_store(req)?;
+    let iam_store = iam_store_for(req.server_context)?;
     let account = cred.access_key.clone();
     let groups = cred.groups.clone();
     let claims = cred.claims.clone().unwrap_or_default();
     let action = Action::S3Action(S3Action::ListAllMyBucketsAction);
     let root_access_key = root_access_key();
     let principal = AuthzPrincipal::new(cred, &root_access_key);
-    let facts = request_authz_facts(
-        req,
+    let facts = AuthzFacts::from_request_parts(
+        req.headers,
+        req.raw_query,
+        req.remote_addr,
+        req.client_info,
         AuthzTarget {
             action,
             bucket: "",
@@ -1433,8 +1542,12 @@ pub(crate) async fn prepare_list_buckets_iam_authorization<T>(req: &S3Request<T>
             version_id: None,
             location_constraint: None,
         },
+        OffsetDateTime::now_utc(),
     );
-    let base_conditions = authorization_conditions(req, &principal, &facts)?;
+    // The service-level listing has no bucket, object, tag or lock input, so its condition values
+    // are the request's own: the same values `authorization_conditions` yields for this action.
+    let mut base_conditions = build_conditions(&principal, &facts);
+    forward_secret_headers(req.headers, &mut base_conditions);
     let mut bucket_conditions = base_conditions.clone();
     bucket_conditions.insert("prefix".to_string(), vec![String::new()]);
     bucket_conditions.insert("delimiter".to_string(), vec!["/".to_string()]);
@@ -1466,8 +1579,7 @@ pub(crate) async fn prepare_list_buckets_iam_authorization<T>(req: &S3Request<T>
 
 /// Preserve the top-level IAM denial audit emitted before ListBuckets falls back
 /// to bucket-level visibility checks.
-pub(crate) fn log_list_buckets_iam_implicit_deny<T>(req: &S3Request<T>) -> S3Result<()> {
-    let req_info = req_info_ref(req)?;
+pub(crate) fn log_list_buckets_iam_implicit_deny(req_info: &ReqInfo) {
     let denial = DenialContext {
         quiet: true,
         suppress_audit: true,
@@ -1479,7 +1591,6 @@ pub(crate) fn log_list_buckets_iam_implicit_deny<T>(req: &S3Request<T>) -> S3Res
         is_owner: req_info.is_owner,
     };
     denial.log("iam_implicit_deny", Action::S3Action(S3Action::ListAllMyBucketsAction));
-    Ok(())
 }
 
 pub(crate) fn delete_object_authorize_action(version_id: Option<&str>) -> Action {
@@ -2676,33 +2787,10 @@ impl S3Access for FS {
         // second embedded server validates keys against its own root identity
         // and IAM domain; the slot resolves the process default when it has
         // not been installed, keeping single-instance behavior unchanged.
-        let app_context = self.server_ctx().app_context();
-        let (cred, is_owner) = if let Some(input_cred) = cx.credentials() {
-            let (cred, is_owner) = check_key_valid_with_context(
-                get_session_token(cx.uri(), cx.headers()).unwrap_or_default(),
-                &input_cred.access_key,
-                app_context.as_deref(),
-            )
-            .await?;
-            (Some(cred), is_owner)
-        } else {
-            (None, false)
-        };
-
+        let access_key = cx.credentials().map(|credentials| credentials.access_key.clone());
+        let session_token = get_session_token(cx.uri(), cx.headers()).unwrap_or_default().to_owned();
         let request_context = cx.extensions_mut().get::<RequestContext>().cloned();
-
-        let region = match app_context.as_deref() {
-            Some(context) => context.region().get(),
-            None => runtime_sources::current_region(),
-        };
-
-        let req_info = ReqInfo {
-            cred,
-            is_owner,
-            region,
-            request_context,
-            ..Default::default()
-        };
+        let req_info = resolve_req_info(self.server_ctx(), access_key.as_deref(), &session_token, request_context).await?;
 
         let auth_type = get_request_auth_type_with_query(cx.headers(), cx.uri().query());
         let verified_presigned = matches!(auth_type, AuthType::Presigned);
@@ -2725,37 +2813,7 @@ impl S3Access for FS {
             }
         }
 
-        // The size capability is intentionally scoped to the single-object
-        // PutObject operation. Validate this at the operation-aware access
-        // boundary so unsupported GET/HEAD/DELETE/bucket routes cannot silently
-        // ignore a signed capability query.
-        if parse_presigned_put_max_content_length(cx.headers(), cx.uri().query(), verified_presigned)?.is_some()
-            && cx.s3_op().name() != "PutObject"
-        {
-            return Err(S3Error::with_message(
-                S3ErrorCode::InvalidRequest,
-                format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
-            ));
-        }
-        if parse_presigned_multipart_max_total_object_size(cx.headers(), cx.uri().query(), verified_sigv4)?.is_some()
-            && cx.s3_op().name() != "CreateMultipartUpload"
-        {
-            return Err(S3Error::with_message(
-                S3ErrorCode::InvalidRequest,
-                format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} is only supported for CreateMultipartUpload"),
-            ));
-        }
-        license_check().map_err(|er| match er.kind() {
-            std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"),
-            _ => {
-                tracing::error!("license check failed due to unexpected error: {er}");
-                s3_error!(InternalError, "License validation failed")
-            }
-        })?;
-
-        // Verify uniformly here? Or verify separately below?
-
-        Ok(())
+        admit_operation(cx.headers(), cx.uri().query(), cx.s3_op().name())
     }
 
     /// Checks whether the CreateBucket request has accesses to the resources.

rustfs/src/storage/ecfs.rs

@@ -45,6 +45,7 @@ use crate::storage::validate_website_configuration;
 use crate::table_catalog;
 use http::StatusCode;
 use metrics::{counter, histogram};
+use rustfs_gateway_types::compat::s3s_5761ddfe::generated::ops::list_buckets as gateway_list_buckets;
 use rustfs_io_metrics::record_s3_op;
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
 use rustfs_s3_ops::S3Operation;
@@ -740,6 +741,15 @@ impl FS {
     }
 }
 
+/// A ListBuckets value the s3s edge and the gateway's types disagree on. The seam converts every
+/// member both sides hold, so this is a defect in the conversion, never the caller's fault.
+fn list_buckets_seam_error(error: rustfs_gateway_types::compat::ConversionError) -> S3Error {
+    S3Error::with_message(
+        S3ErrorCode::InternalError,
+        format!("failed to convert ListBuckets through the gateway compatibility seam: {error}"),
+    )
+}
+
 #[async_trait::async_trait]
 impl S3 for FS {
     #[instrument(level = "debug", skip(self, req))]
@@ -1596,7 +1606,17 @@ impl S3 for FS {
         // List buckets not associated with a bucket, give it bucket label "*" to denote "all".
         record_s3_op(S3Operation::ListBuckets);
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_list_buckets(req).await
+        // The use case speaks the envelope and the gateway's types; this edge only converts.
+        let envelope = crate::app::RequestEnvelope::from_s3s(&req);
+        let input = gateway_list_buckets::input_from_s3s(req.input).map_err(list_buckets_seam_error)?;
+        let output = usecase.execute_list_buckets(envelope, input).await?;
+        let mut output = gateway_list_buckets::output_to_s3s(output).map_err(list_buckets_seam_error)?;
+        // The seam writes a listing without buckets as an absent member. This edge has always
+        // answered the element, empty or not, and clients index the answer by it.
+        if output.buckets.is_none() {
+            output.buckets = Some(Vec::new());
+        }
+        Ok(S3Response::new(output))
     }
 
     async fn list_multipart_uploads(

rustfs/src/storage/s3_api/bucket.rs

@@ -13,15 +13,14 @@
 // limitations under the License.
 
 use crate::storage::s3_api::common::rustfs_owner;
-use crate::storage::storage_api::s3_api_consumer::bucket::contract::{
-    bucket::BucketInfo,
-    list::{ListObjectVersionsInfo as StorageListObjectVersionsInfo, ListObjectsV2Info as StorageListObjectsV2Info},
+use crate::storage::storage_api::s3_api_consumer::bucket::contract::list::{
+    ListObjectVersionsInfo as StorageListObjectVersionsInfo, ListObjectsV2Info as StorageListObjectsV2Info,
 };
 use crate::storage::storage_api::s3_api_consumer::bucket::to_s3s_xml_etag;
 use percent_encoding::percent_decode_str;
 use s3s::dto::{
-    Bucket, CommonPrefix, DeleteMarkerEntry, EncodingType, ListBucketsOutput, ListObjectVersionsOutput, ListObjectsOutput,
-    ListObjectsV2Output, Object, ObjectStorageClass, ObjectVersion, ObjectVersionStorageClass, Timestamp,
+    CommonPrefix, DeleteMarkerEntry, EncodingType, ListObjectVersionsOutput, ListObjectsOutput, ListObjectsV2Output, Object,
+    ObjectStorageClass, ObjectVersion, ObjectVersionStorageClass, Timestamp,
 };
 use s3s::{S3Error, S3ErrorCode};
 use tracing::debug;
@@ -126,23 +125,6 @@ pub(crate) fn build_list_objects_output(v2: ListObjectsV2Output, request_marker:
     }
 }
 
-pub(crate) fn build_list_buckets_output(bucket_infos: &[BucketInfo]) -> ListBucketsOutput {
-    let buckets: Vec<Bucket> = bucket_infos
-        .iter()
-        .map(|bucket_info| Bucket {
-            creation_date: Some(Timestamp::from(bucket_info.created.unwrap_or(time::OffsetDateTime::UNIX_EPOCH))),
-            name: Some(bucket_info.name.clone()),
-            ..Default::default()
-        })
-        .collect();
-
-    ListBucketsOutput {
-        buckets: Some(buckets),
-        owner: Some(rustfs_owner()),
-        ..Default::default()
-    }
-}
-
 pub(crate) fn parse_list_object_versions_params(
     prefix: Option<String>,
     delimiter: Option<String>,
@@ -450,19 +432,16 @@ fn calculate_next_marker(v2: &ListObjectsV2Output) -> Option<String> {
 #[cfg(test)]
 mod tests {
     use super::{
-        ListObjectVersionsInfo, ListObjectVersionsParams, ListObjectsV2Info, build_list_buckets_output,
-        build_list_object_versions_output, build_list_objects_output, build_list_objects_v2_output,
-        encode_list_objects_v2_continuation_token, parse_list_object_versions_params, parse_list_objects_v2_params,
-        to_s3s_xml_etag,
+        ListObjectVersionsInfo, ListObjectVersionsParams, ListObjectsV2Info, build_list_object_versions_output,
+        build_list_objects_output, build_list_objects_v2_output, encode_list_objects_v2_continuation_token,
+        parse_list_object_versions_params, parse_list_objects_v2_params, to_s3s_xml_etag,
     };
     use crate::storage::s3_api::common::rustfs_owner;
     use crate::storage::storage_api::s3_api_consumer::bucket::StorageObjectInfo as ObjectInfo;
-    use crate::storage::storage_api::s3_api_consumer::bucket::contract::bucket::BucketInfo;
     use rustfs_filemeta::FileInfo;
     use rustfs_utils::http::headers::AMZ_STORAGE_CLASS;
     use s3s::S3ErrorCode;
     use s3s::dto::{CommonPrefix, EncodingType, ListObjectsV2Output, Object};
-    use time::OffsetDateTime;
     use uuid::Uuid;
 
     #[test]
@@ -525,36 +504,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_build_list_buckets_output_maps_bucket_infos_and_owner() {
-        let bucket_infos = vec![
-            BucketInfo {
-                name: "bucket-a".to_string(),
-                created: Some(OffsetDateTime::UNIX_EPOCH),
-                ..Default::default()
-            },
-            BucketInfo {
-                name: "bucket-b".to_string(),
-                created: None,
-                ..Default::default()
-            },
-        ];
-
-        let output = build_list_buckets_output(&bucket_infos);
-        let buckets = output.buckets.as_ref().expect("buckets should be present");
-        let owner = output.owner.as_ref().expect("owner should be present");
-
-        assert_eq!(buckets.len(), 2);
-        assert_eq!(buckets[0].name.as_deref(), Some("bucket-a"));
-        assert_eq!(buckets[0].creation_date, Some(s3s::dto::Timestamp::from(OffsetDateTime::UNIX_EPOCH)));
-        assert_eq!(buckets[1].name.as_deref(), Some("bucket-b"));
-        assert_eq!(buckets[1].creation_date, Some(s3s::dto::Timestamp::from(OffsetDateTime::UNIX_EPOCH)));
-
-        let expected_owner = rustfs_owner();
-        assert_eq!(owner.display_name, expected_owner.display_name);
-        assert_eq!(owner.id, expected_owner.id);
-    }
-
     #[test]
     fn test_list_objects_marker_echoes_request_value() {
         let output = build_list_objects_output(ListObjectsV2Output::default(), Some("m-1".to_string()));

rustfs/src/storage/storage_api.rs

@@ -148,18 +148,20 @@ pub(crate) use super::sse::{
 pub(crate) mod access_consumer {
     pub(crate) use super::super::access::{
         AuthorizationRequest, BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard,
-        MultipartStagingOperation, ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard,
-        PostObjectPolicyHeaders, PostObjectRequestMarker, ReqInfo, RequestedCopySource, RequestedObjectLegalHold,
-        RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags, StagedMultipartPartAuthorization,
-        TABLE_DATA_PLANE_LIST_CURSOR_PREFIX, TableDataPlaneListAccess, TableDataPlaneListCursorPosition,
-        TableDataPlanePublicationGuards, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard,
-        authorize_copy_object_tags, authorize_request, authorize_request_core, bucket_config_mutation_incarnation,
-        delete_object_authorize_action, has_bypass_governance_header, load_bucket_generation_from_store,
-        log_list_buckets_iam_implicit_deny, odm_read_generation, prepare_list_buckets_iam_authorization,
-        prepare_odm_read_generation, recursive_force_delete_has_authenticated_caller, replication_proxy_allowed,
-        replication_request_authorized, req_info_mut, req_info_ref, retained_table_data_plane_publication_guards,
-        set_requested_object_lock_retain_until,
+        ListBucketsAuthzRequest, MultipartStagingOperation, ObjectTagConditions, OdmReadGenerationGuard,
+        PendingDeleteBucketGenerationGuard, PostObjectPolicyHeaders, PostObjectRequestMarker, ReqInfo, RequestedCopySource,
+        RequestedObjectLegalHold, RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags,
+        StagedMultipartPartAuthorization, TABLE_DATA_PLANE_LIST_CURSOR_PREFIX, TableDataPlaneListAccess,
+        TableDataPlaneListCursorPosition, TableDataPlanePublicationGuards, apply_bucket_generation_guard,
+        apply_copy_source_bucket_generation_guard, authorize_copy_object_tags, authorize_request, authorize_request_core,
+        bucket_config_mutation_incarnation, delete_object_authorize_action, has_bypass_governance_header,
+        load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
+        prepare_list_buckets_iam_authorization, prepare_odm_read_generation, recursive_force_delete_has_authenticated_caller,
+        replication_proxy_allowed, replication_request_authorized, req_info_mut, req_info_ref,
+        retained_table_data_plane_publication_guards, set_requested_object_lock_retain_until,
     };
+    #[cfg(feature = "gateway-stack")]
+    pub(crate) use super::super::access::{admit_operation, resolve_req_info};
     pub(crate) use super::super::denial_audit::DenialAuditor;
     pub(crate) use super::super::helper::{AuditRequestView, DenialScope};
 }
@@ -369,16 +371,12 @@ pub(crate) mod heal_control_startup_consumer {
 pub(crate) mod s3_api_consumer {
     pub(crate) mod bucket {
         pub(crate) use super::super::super::s3_api::bucket::{
-            ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output,
-            build_list_objects_output, build_list_objects_v2_output, encode_list_objects_v2_continuation_token,
-            parse_list_object_versions_params, parse_list_objects_v2_params,
+            ListObjectVersionsParams, ListObjectsV2Params, build_list_object_versions_output, build_list_objects_output,
+            build_list_objects_v2_output, encode_list_objects_v2_continuation_token, parse_list_object_versions_params,
+            parse_list_objects_v2_params,
         };
 
         pub(crate) mod contract {
-            pub(crate) mod bucket {
-                pub(crate) use super::super::super::super::contract::bucket::BucketInfo;
-            }
-
             pub(crate) mod list {
                 pub(crate) use super::super::super::super::contract::list::{ListObjectVersionsInfo, ListObjectsV2Info};
             }

rustfs/src/storage_api.rs

@@ -232,6 +232,13 @@ pub(crate) mod server {
             pub(crate) use crate::storage::storage_api::request_context_consumer::RequestContext;
         }
 
+        /// The access-phase functions the gateway stack's handlers run, shared with the legacy
+        /// access hooks.
+        #[cfg(feature = "gateway-stack")]
+        pub(crate) mod access {
+            pub(crate) use crate::storage::storage_api::access_consumer::{admit_operation, resolve_req_info};
+        }
+
         pub(crate) mod rpc {
             pub(crate) use crate::storage::storage_api::rpc_consumer::InternodeRpcService;
             #[cfg(test)]

rustfs/tests/embedded_gateway_stack_test.rs

@@ -0,0 +1,402 @@
+// 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 S3 stack inside an embedded RustFS server (rustfs/backlog#2734 T2.5): one
+// operation, ListBuckets, is served end to end, and every other S3 operation is answered
+// `501 NotImplemented`. Requests are signed by the AWS SDK and travel through the whole
+// middleware stack, so what is asserted here is what a client sees.
+
+#![recursion_limit = "256"]
+
+use aws_sdk_s3::config::{Credentials, Region};
+use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::{Client, Config};
+use rustfs::app::context::get_global_app_context;
+use rustfs::embedded::{RustFSServer, RustFSServerBuilder, find_available_port};
+use rustfs_io_metrics::s3_http_metrics::s3_http_metrics_snapshot;
+use rustfs_s3_ops::S3Operation;
+use rustfs_storage_api::{BucketOperations, MakeBucketOptions};
+use std::collections::BTreeMap;
+use std::future::Future;
+
+mod common;
+
+const ROOT_ACCESS_KEY: &str = "gatewaystackroot";
+const ROOT_SECRET_KEY: &str = "gateway-stack-root-secret";
+const USER_SECRET_KEY: &str = "gateway-stack-user-secret";
+const ENV_S3_STACK: &str = "RUSTFS_S3_STACK";
+const ENV_KEYSTONE_ENABLE: &str = "RUSTFS_KEYSTONE_ENABLE";
+const ENV_KEYSTONE_AUTH_URL: &str = "RUSTFS_KEYSTONE_AUTH_URL";
+/// A Keystone endpoint nothing listens on: these tests are about start-up, not about tokens.
+const UNREACHABLE_KEYSTONE: &str = "http://127.0.0.1:1/v3";
+
+fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client {
+    let config = Config::builder()
+        .credentials_provider(Credentials::new(access_key, secret_key, None, None, "test"))
+        .region(Region::new("us-east-1"))
+        .endpoint_url(endpoint)
+        .force_path_style(true)
+        .behavior_version_latest()
+        .build();
+    Client::from_conf(config)
+}
+
+/// Runs `test` against an embedded server started with `RUSTFS_S3_STACK` set to `stack`.
+fn with_stack<F, Fut>(stack: Option<&'static str>, test: F)
+where
+    F: FnOnce(RustFSServer) -> Fut + Send + 'static,
+    Fut: Future<Output = ()> + 'static,
+{
+    with_environment(vec![(ENV_S3_STACK, stack)], test);
+}
+
+/// Runs `test` against an embedded server started under `environment`.
+fn with_environment<F, Fut>(environment: Vec<(&'static str, Option<&'static str>)>, test: F)
+where
+    F: FnOnce(RustFSServer) -> Fut + Send + 'static,
+    Fut: Future<Output = ()> + 'static,
+{
+    temp_env::with_vars(environment, || {
+        common::run_embedded_test(move || async move {
+            let Some(started) = start_server().await else { return };
+            test(started.expect("start embedded server")).await;
+        });
+    });
+}
+
+/// Starts an embedded server on a free port; `None` when the sandbox allows no port at all.
+async fn start_server() -> Option<Result<RustFSServer, String>> {
+    let port = match find_available_port() {
+        Ok(port) => port,
+        Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
+        Err(err) => panic!("find free port: {err}"),
+    };
+    let started = RustFSServerBuilder::new()
+        .address(format!("127.0.0.1:{port}"))
+        .access_key(ROOT_ACCESS_KEY)
+        .secret_key(ROOT_SECRET_KEY)
+        .build()
+        .await;
+    // Both renderings: the refusal's text may be in either.
+    Some(started.map_err(|error| format!("{error} {error:?}")))
+}
+
+/// The S3 error code and HTTP status an SDK call was refused with.
+fn refusal<E: ProvideErrorMetadata>(error: &SdkError<E>) -> (Option<String>, Option<u16>) {
+    let status = error.raw_response().map(|response| response.status().as_u16());
+    (error.code().map(str::to_owned), status)
+}
+
+/// One series of the S3 request counter: method, operation and outcome.
+type Series = (&'static str, &'static str, &'static str);
+
+/// How often each series of the S3 request counter has been counted in this process.
+fn s3_request_counts() -> BTreeMap<Series, u64> {
+    s3_http_metrics_snapshot()
+        .into_iter()
+        .map(|series| ((series.method, series.operation, series.outcome), series.total))
+        .collect()
+}
+
+/// Every series counted since `before`, with how often. Reading the whole counter, not one
+/// series, is what makes a second count under another label visible.
+fn counted_since(before: &BTreeMap<Series, u64>) -> BTreeMap<Series, u64> {
+    s3_request_counts()
+        .into_iter()
+        .filter_map(|(series, total)| {
+            let counted = total - before.get(&series).copied().unwrap_or(0);
+            (counted > 0).then_some((series, counted))
+        })
+        .collect()
+}
+
+async fn create_bucket_in_the_store(bucket: &str) {
+    get_global_app_context()
+        .expect("embedded server context")
+        .object_store()
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket in the store");
+}
+
+async fn create_user_without_a_policy(access_key: &str) {
+    get_global_app_context()
+        .expect("embedded server context")
+        .iam()
+        .handle()
+        .create_user(
+            access_key,
+            &rustfs_madmin::AddOrUpdateUserReq {
+                secret_key: USER_SECRET_KEY.to_string(),
+                policy: None,
+                status: rustfs_madmin::AccountStatus::Enabled,
+            },
+        )
+        .await
+        .expect("create user without a policy");
+}
+
+#[test]
+fn the_gateway_stack_lists_the_caller_s_buckets_and_answers_every_other_operation_501() {
+    with_stack(Some("gateway"), |server| async move {
+        // CreateBucket is not served by this stack yet, so the buckets are made in the store.
+        create_bucket_in_the_store("gateway-stack-alpha").await;
+        create_bucket_in_the_store("gateway-stack-beta").await;
+        let client = s3_client(&server.endpoint(), server.access_key(), server.secret_key());
+
+        let listing = client.list_buckets().send().await.expect("signed ListBuckets");
+        let mut names: Vec<&str> = listing.buckets().iter().filter_map(|bucket| bucket.name()).collect();
+        names.sort_unstable();
+        assert_eq!(names, ["gateway-stack-alpha", "gateway-stack-beta"]);
+        assert!(listing.owner().and_then(|owner| owner.id()).is_some(), "the listing names its owner");
+        assert!(
+            listing.buckets().iter().all(|bucket| bucket.creation_date().is_some()),
+            "every bucket carries its creation date"
+        );
+
+        let get = client
+            .get_object()
+            .bucket("gateway-stack-alpha")
+            .key("object")
+            .send()
+            .await
+            .expect_err("GetObject is not registered");
+        assert_eq!(refusal(&get), (Some("NotImplemented".to_owned()), Some(501)));
+
+        let put = client
+            .put_object()
+            .bucket("gateway-stack-alpha")
+            .key("object")
+            .body(ByteStream::from_static(b"payload"))
+            .send()
+            .await
+            .expect_err("PutObject is not registered");
+        assert_eq!(refusal(&put), (Some("NotImplemented".to_owned()), Some(501)));
+
+        let create = client
+            .create_bucket()
+            .bucket("gateway-stack-gamma")
+            .send()
+            .await
+            .expect_err("CreateBucket is not registered");
+        assert_eq!(refusal(&create), (Some("NotImplemented".to_owned()), Some(501)));
+
+        let listing = client.list_buckets().send().await.expect("ListBuckets after the refusals");
+        assert_eq!(listing.buckets().len(), 2, "a refused write created nothing");
+
+        server.shutdown().await;
+    });
+}
+
+#[test]
+fn the_gateway_stack_refuses_callers_it_cannot_admit() {
+    with_stack(Some("gateway"), |server| async move {
+        create_bucket_in_the_store("gateway-stack-alpha").await;
+        create_user_without_a_policy("gatewaystacknopolicy").await;
+        let endpoint = server.endpoint();
+
+        // Anonymous: no credentials at all.
+        let anonymous = reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("http client")
+            .get(format!("{endpoint}/"))
+            .send()
+            .await
+            .expect("anonymous ListBuckets is answered");
+        assert_eq!(anonymous.status().as_u16(), 403);
+        let request_id = anonymous
+            .headers()
+            .get("x-amz-request-id")
+            .and_then(|value| value.to_str().ok())
+            .map(str::to_owned)
+            .expect("the refusal names its request");
+        assert_eq!(
+            anonymous.headers().get("x-request-id").and_then(|value| value.to_str().ok()),
+            Some(request_id.as_str()),
+            "one identifier names the request in both headers"
+        );
+        let body = anonymous.text().await.expect("refusal body");
+        assert!(body.contains("<Code>AccessDenied</Code>"), "{body}");
+
+        // The right access key with the wrong secret.
+        let wrong_secret = s3_client(&endpoint, ROOT_ACCESS_KEY, "not-the-root-secret")
+            .list_buckets()
+            .send()
+            .await
+            .expect_err("a wrong secret signs nothing");
+        assert_eq!(refusal(&wrong_secret), (Some("SignatureDoesNotMatch".to_owned()), Some(403)));
+
+        // An access key nobody issued: the caller's fault, not the server's.
+        let unknown_key = s3_client(&endpoint, "gatewaystackunknown", USER_SECRET_KEY)
+            .list_buckets()
+            .send()
+            .await
+            .expect_err("an unknown access key is refused");
+        assert_eq!(refusal(&unknown_key), (Some("InvalidAccessKeyId".to_owned()), Some(403)));
+
+        // A real user whose policy grants nothing: it may neither list every bucket nor see one.
+        let no_permission = s3_client(&endpoint, "gatewaystacknopolicy", USER_SECRET_KEY)
+            .list_buckets()
+            .send()
+            .await
+            .expect_err("a caller without permission lists nothing");
+        assert_eq!(refusal(&no_permission), (Some("AccessDenied".to_owned()), Some(403)));
+
+        // The root caller is still served by the same server.
+        let listing = s3_client(&endpoint, server.access_key(), server.secret_key())
+            .list_buckets()
+            .send()
+            .await
+            .expect("signed ListBuckets");
+        assert_eq!(listing.buckets().len(), 1);
+
+        server.shutdown().await;
+    });
+}
+
+/// The S3 request counter belongs to RustFS's layer in front of both stacks, and the operation
+/// label is set by the code a handler calls. That only works if the gateway polls its handler
+/// inside that layer's scope, on the request's own task, which is what this observes.
+#[test]
+fn the_gateway_stack_counts_each_request_once_under_the_label_the_legacy_stack_uses() {
+    with_stack(Some("gateway"), |server| async move {
+        let endpoint = server.endpoint();
+
+        let before = s3_request_counts();
+        s3_client(&endpoint, server.access_key(), server.secret_key())
+            .list_buckets()
+            .send()
+            .await
+            .expect("signed ListBuckets");
+        assert_eq!(
+            counted_since(&before),
+            BTreeMap::from([(("GET", S3Operation::ListBuckets.as_str(), "2xx"), 1)]),
+            "one served request is counted once, as ListBuckets"
+        );
+
+        let before = s3_request_counts();
+        let refused = s3_client(&endpoint, ROOT_ACCESS_KEY, "not-the-root-secret")
+            .list_buckets()
+            .send()
+            .await
+            .expect_err("a wrong secret signs nothing");
+        assert_eq!(refusal(&refused), (Some("SignatureDoesNotMatch".to_owned()), Some(403)));
+        assert_eq!(
+            counted_since(&before),
+            BTreeMap::from([(("GET", "unknown", "4xx"), 1)]),
+            "a request refused by authentication is counted once, under no operation"
+        );
+
+        server.shutdown().await;
+    });
+}
+
+/// The pinned gateway admits a request that carries only `X-Auth-Token` as anonymous, while the
+/// RustFS code behind it reads the Keystone identity. Until the gateway authenticates the token
+/// (rustfs/gateway#1586) the two settings together refuse to start, and nothing is served.
+#[test]
+fn the_gateway_stack_refuses_to_start_with_keystone_enabled() {
+    let environment = [
+        (ENV_S3_STACK, Some("gateway")),
+        (ENV_KEYSTONE_ENABLE, Some("true")),
+        (ENV_KEYSTONE_AUTH_URL, Some(UNREACHABLE_KEYSTONE)),
+    ];
+    temp_env::with_vars(environment, || {
+        common::run_embedded_test(|| async {
+            let Some(started) = start_server().await else { return };
+            let refusal = match started {
+                Ok(server) => {
+                    server.shutdown().await;
+                    panic!("the gateway stack started with Keystone enabled");
+                }
+                Err(refusal) => refusal,
+            };
+            for named in ["RUSTFS_S3_STACK=gateway", "RUSTFS_KEYSTONE_ENABLE", "rustfs/gateway#1586"] {
+                assert!(refusal.contains(named), "{named} is missing from: {refusal}");
+            }
+        });
+    });
+}
+
+#[test]
+fn the_legacy_stack_starts_with_keystone_enabled() {
+    let environment = vec![
+        (ENV_S3_STACK, None),
+        (ENV_KEYSTONE_ENABLE, Some("true")),
+        (ENV_KEYSTONE_AUTH_URL, Some(UNREACHABLE_KEYSTONE)),
+    ];
+    with_environment(environment, |server| async move {
+        let listing = s3_client(&server.endpoint(), server.access_key(), server.secret_key())
+            .list_buckets()
+            .send()
+            .await
+            .expect("signed ListBuckets on the legacy stack with Keystone enabled");
+        assert!(listing.buckets().is_empty());
+
+        server.shutdown().await;
+    });
+}
+
+#[test]
+fn the_legacy_stack_still_serves_every_operation() {
+    with_stack(None, |server| async move {
+        let client = s3_client(&server.endpoint(), server.access_key(), server.secret_key());
+
+        client
+            .create_bucket()
+            .bucket("legacy-stack-alpha")
+            .send()
+            .await
+            .expect("CreateBucket on the legacy stack");
+        client
+            .put_object()
+            .bucket("legacy-stack-alpha")
+            .key("object")
+            .body(ByteStream::from_static(b"payload"))
+            .send()
+            .await
+            .expect("PutObject on the legacy stack");
+        let before = s3_request_counts();
+        let listing = client.list_buckets().send().await.expect("ListBuckets on the legacy stack");
+        assert_eq!(
+            counted_since(&before),
+            BTreeMap::from([(("GET", S3Operation::ListBuckets.as_str(), "2xx"), 1)]),
+            "the label the gateway stack is held to is the one this stack counts under"
+        );
+        let names: Vec<&str> = listing.buckets().iter().filter_map(|bucket| bucket.name()).collect();
+        assert_eq!(names, ["legacy-stack-alpha"]);
+        assert!(listing.owner().and_then(|owner| owner.id()).is_some(), "the listing names its owner");
+
+        let missing = client
+            .get_object()
+            .bucket("legacy-stack-alpha")
+            .key("missing")
+            .send()
+            .await
+            .expect_err("a missing key");
+        assert_eq!(refusal(&missing), (Some("NoSuchKey".to_owned()), Some(404)));
+
+        create_user_without_a_policy("legacystacknopolicy").await;
+        let no_permission = s3_client(&server.endpoint(), "legacystacknopolicy", USER_SECRET_KEY)
+            .list_buckets()
+            .send()
+            .await
+            .expect_err("a caller without permission lists nothing on either stack");
+        assert_eq!(refusal(&no_permission), (Some("AccessDenied".to_owned()), Some(403)));
+
+        server.shutdown().await;
+    });
+}

scripts/check_gateway_stack_release_boundary.sh

@@ -0,0 +1,226 @@
+#!/usr/bin/env bash
+# Keeps the `gateway-stack` cargo feature of the rustfs crate out of every release build while the
+# gateway S3 stack is under construction (rustfs/backlog#2734 T2.5).
+#
+# Usage: scripts/check_gateway_stack_release_boundary.sh [--root DIR]
+#        scripts/check_gateway_stack_release_boundary.sh --self-test
+#
+# Three facts are checked, none of which needs Cargo:
+#   1. rustfs/src/server/gateway_stack.rs refuses to compile without debug assertions, so a
+#      release-profile build that enables the feature fails in the compiler;
+#   2. neither the `default` nor the `full` feature list of rustfs/Cargo.toml enables it, directly
+#      or through a feature that enables it, and the feature itself still exists (a guard whose
+#      subject is gone must fail, not skip);
+#   3. no file that produces a release artifact names the feature or one that enables it.
+# A missing input is an error: every file below exists in this repository.
+
+set -euo pipefail
+
+FEATURE="gateway-stack"
+GUARD_SOURCE="rustfs/src/server/gateway_stack.rs"
+GUARD_CFG='#[cfg(not(debug_assertions))]'
+GUARD_ERROR='compile_error!("gateway-stack is restricted to debug builds");'
+MANIFEST="rustfs/Cargo.toml"
+# Everything that builds or packages a release binary.
+RELEASE_FILES=(
+  ".github/workflows/build.yml"
+  ".github/workflows/docker.yml"
+  "build-rustfs.sh"
+  "docker-buildx.sh"
+  "Dockerfile"
+  "Dockerfile.glibc"
+  "Dockerfile.source"
+  "rustfs.spec"
+  "Justfile"
+  ".config/make/build.mak"
+  ".config/make/build-docker.mak"
+)
+
+fail() {
+  echo "gateway-stack release boundary: $*" >&2
+  return 1
+}
+
+# Prints the value of one `name = [...]` feature line of the manifest.
+feature_list() {
+  local manifest="$1" name="$2"
+  grep -E "^${name}[[:space:]]*=" "$manifest" || true
+}
+
+# Prints the feature and every feature that enables it, directly or through another, one per line.
+enabling_features() {
+  local manifest="$1" known="$FEATURE" grown name
+  while true; do
+    grown="$known"
+    while IFS= read -r name; do
+      grep -qxF -- "$name" <<<"$grown" || grown+=$'\n'"$name"
+    done < <(
+      while IFS= read -r name; do
+        grep -E "^[A-Za-z0-9_-]+[[:space:]]*=.*\"${name}\"" "$manifest" | sed -E 's/[[:space:]]*=.*//' || true
+      done <<<"$known"
+    )
+    [[ "$grown" == "$known" ]] && break
+    known="$grown"
+  done
+  printf '%s\n' "$known"
+}
+
+check_root() {
+  local root="$1" status=0 file line name enabler
+  local -a enablers=("$FEATURE")
+
+  file="${root}/${GUARD_SOURCE}"
+  if [[ ! -f "$file" ]]; then
+    fail "${GUARD_SOURCE} is missing" || status=1
+  else
+    # The cfg must sit on the line directly above the compile_error.
+    line="$(grep -nF "$GUARD_ERROR" "$file" | head -n 1 | cut -d: -f1)"
+    if [[ -z "$line" ]]; then
+      fail "${GUARD_SOURCE} no longer refuses a build without debug assertions" || status=1
+    elif [[ "$(sed -n "$((line - 1))p" "$file" | tr -d '[:space:]')" != "$(tr -d '[:space:]' <<<"$GUARD_CFG")" ]]; then
+      fail "the compile_error in ${GUARD_SOURCE} is not gated on cfg(not(debug_assertions))" || status=1
+    fi
+  fi
+
+  file="${root}/${MANIFEST}"
+  if [[ ! -f "$file" ]]; then
+    fail "${MANIFEST} is missing" || status=1
+  else
+    if [[ -z "$(feature_list "$file" "$FEATURE")" ]]; then
+      fail "${MANIFEST} no longer declares the ${FEATURE} feature this guard protects" || status=1
+    fi
+    mapfile -t enablers < <(enabling_features "$file")
+    for name in default full; do
+      line="$(feature_list "$file" "$name")"
+      if [[ -z "$line" ]]; then
+        fail "${MANIFEST} has no ${name} feature list" || status=1
+      fi
+      for enabler in "${enablers[@]}"; do
+        if [[ "$enabler" == "$name" ]]; then
+          fail "the ${name} feature list of ${MANIFEST} enables ${FEATURE}" || status=1
+        fi
+      done
+    done
+  fi
+
+  for file in "${RELEASE_FILES[@]}"; do
+    if [[ ! -f "${root}/${file}" ]]; then
+      fail "release file ${file} is missing; update RELEASE_FILES" || status=1
+      continue
+    fi
+    for enabler in "${enablers[@]}"; do
+      if grep -qF -- "$enabler" "${root}/${file}"; then
+        fail "${file} names ${enabler}, which enables ${FEATURE}; release artifacts must not" || status=1
+      fi
+    done
+  done
+
+  return "$status"
+}
+
+self_test() {
+  local scratch fixture failures=0
+  scratch="$(mktemp -d)"
+
+  make_fixture() {
+    fixture="${scratch}/$1"
+    mkdir -p "${fixture}/rustfs/src/server" "${fixture}/.github/workflows" "${fixture}/.config/make"
+    printf '%s\n%s\n' "$GUARD_CFG" "$GUARD_ERROR" >"${fixture}/${GUARD_SOURCE}"
+    printf '%s\n' '[features]' 'default = ["ftps"]' 'full = ["ftps", "swift"]' 'gateway-stack = ["dep:rustfs-gateway"]' \
+      >"${fixture}/${MANIFEST}"
+    for file in "${RELEASE_FILES[@]}"; do
+      printf 'cargo build --release -p rustfs\n' >"${fixture}/${file}"
+    done
+  }
+
+  expect() {
+    local wanted="$1" label="$2"
+    if check_root "$fixture" >/dev/null 2>&1; then
+      [[ "$wanted" == "pass" ]] || { echo "self-test: ${label} was accepted" >&2; failures=$((failures + 1)); }
+    else
+      [[ "$wanted" == "fail" ]] || { echo "self-test: ${label} was refused" >&2; failures=$((failures + 1)); }
+    fi
+  }
+
+  make_fixture valid
+  expect pass "a repository that keeps the boundary"
+
+  make_fixture no-compile-error
+  printf '// nothing here\n' >"${fixture}/${GUARD_SOURCE}"
+  expect fail "a gateway stack without the compile-time refusal"
+
+  make_fixture ungated-compile-error
+  printf '%s\n%s\n' '#[cfg(test)]' "$GUARD_ERROR" >"${fixture}/${GUARD_SOURCE}"
+  expect fail "a compile_error gated on something else"
+
+  make_fixture missing-source
+  rm "${fixture}/${GUARD_SOURCE}"
+  expect fail "a missing guard source"
+
+  make_fixture default-enables
+  sed -i.bak 's/^default = .*/default = ["ftps", "gateway-stack"]/' "${fixture}/${MANIFEST}"
+  expect fail "a default feature list that enables the stack"
+
+  make_fixture full-enables
+  sed -i.bak 's/^full = .*/full = ["ftps", "gateway-stack"]/' "${fixture}/${MANIFEST}"
+  expect fail "a full feature list that enables the stack"
+
+  make_fixture feature-gone
+  sed -i.bak '/^gateway-stack = /d' "${fixture}/${MANIFEST}"
+  expect fail "a manifest without the feature"
+
+  make_fixture full-enables-through-another
+  printf '%s\n' 'hooks = ["gateway-stack"]' >>"${fixture}/${MANIFEST}"
+  sed -i.bak 's/^full = .*/full = ["ftps", "hooks"]/' "${fixture}/${MANIFEST}"
+  expect fail "a full feature list that enables the stack through another feature"
+
+  make_fixture debug-only-enabler
+  printf '%s\n' 'hooks = ["gateway-stack"]' >>"${fixture}/${MANIFEST}"
+  expect pass "a feature outside default and full that enables the stack"
+
+  make_fixture release-file-names-enabler
+  printf '%s\n' 'hooks = ["gateway-stack"]' >>"${fixture}/${MANIFEST}"
+  printf 'cargo build --release -p rustfs --features hooks\n' >"${fixture}/build-rustfs.sh"
+  expect fail "a release file that enables the stack through another feature"
+
+  make_fixture release-workflow
+  printf 'cargo build --release -p rustfs --features gateway-stack\n' >"${fixture}/.github/workflows/build.yml"
+  expect fail "a release workflow that enables the stack"
+
+  make_fixture missing-release-file
+  rm "${fixture}/build-rustfs.sh"
+  expect fail "a missing release file"
+
+  rm -rf "$scratch"
+  if ((failures > 0)); then
+    echo "gateway-stack release boundary self-test: ${failures} failure(s)" >&2
+    return 1
+  fi
+  echo "gateway-stack release boundary self-test passed"
+}
+
+main() {
+  local root
+  root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+  case "${1:-}" in
+    --self-test)
+      [[ $# -eq 1 ]] || { echo "usage: $0 [--root DIR] | --self-test" >&2; exit 2; }
+      self_test
+      ;;
+    --root)
+      [[ $# -eq 2 ]] || { echo "usage: $0 [--root DIR] | --self-test" >&2; exit 2; }
+      check_root "$2"
+      echo "gateway-stack release boundary holds"
+      ;;
+    "")
+      check_root "$root"
+      echo "gateway-stack release boundary holds"
+      ;;
+    *)
+      echo "usage: $0 [--root DIR] | --self-test" >&2
+      exit 2
+      ;;
+  esac
+}
+
+main "$@"

scripts/ci_gate.py

@@ -17,7 +17,7 @@ ROOT = Path(__file__).resolve().parent.parent
 ALWAYS_JOBS = ("classify-changes", "typos", "quick-checks")
 CODE_JOBS = (
     "workspace-lint-and-build", "workspace-tests", "core-contract", "connect-feature-boundaries", "fips-provider", "test-and-lint-rio-v2",
-    "test-and-lint-protocols", "build-rustfs-debug-binary", "uring-integration",
+    "test-and-lint-protocols", "test-and-lint-gateway-stack", "build-rustfs-debug-binary", "uring-integration",
     "e2e-tests", "s3-implemented-tests", "s3-lifecycle-behavior-tests",
 )
 # Specialized boundary, hardware and alternate-feature lanes do not gate a
@@ -33,7 +33,7 @@ NIGHTLY_JOBS = ("connect-feature-boundaries", "fips-provider", "uring-integratio
 # against this revision's expected job contract. The workflow refuses a plan
 # that does not carry the current marker and falls back to the full matrix, so
 # a base planner predating the contract cannot authorize a narrower selection.
-PLAN_CONTRACT = "3"
+PLAN_CONTRACT = "4"
 # These jobs still exercise the server and its black-box harness when only E2E
 # Rust sources change. Production code, manifests and shared test configuration
 # always select the full matrix. No production package depends on e2e_test.
@@ -299,6 +299,8 @@ def configurations(jobs: dict[str, str]) -> list[dict]:
         ("protocol-sftp-tests", "test-and-lint-protocols", "test", ["sftp"]),
         ("protocol-http3-lint", "test-and-lint-protocols", "dev", ["http3", "swift"]),
         ("protocol-http3-tests", "test-and-lint-protocols", "test", ["http3", "swift"]),
+        ("gateway-stack-lint", "test-and-lint-gateway-stack", "dev", ["gateway-stack"]),
+        ("gateway-stack-tests", "test-and-lint-gateway-stack", "test", ["gateway-stack"]),
         ("server-rio-v2", "build-rustfs-debug-binary-rio-v2", "dev", ["e2e-test-hooks", "rio-v2"]),
     ]
     return [{"id": identity, "job": job, "target": "x86_64-unknown-linux-gnu", "profile": profile,