test: mutation set A for #9287 (do not merge)

RequestCredentials::secret_key now returns Option<&str>: None when the edge kept the secret, and an empty handed-over secret is refused at construction. Secret comparisons go through secret_key_matches, which never matches an absent secret, and readers that need the value fail closed with an internal error.

.config/make/tests.mak

@@ -66,6 +66,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/codemods/s3_error_to_s3_types.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_download_release_asset.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_reference_binary.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_inspect_btf_prototypes.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_release_build_workflow.py
 	$(RUSTFS_PYTHON_BIN) -m unittest discover -s scripts/release -p "test_*.py"
@@ -92,6 +93,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_junit_diff.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_report_compat.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_runner_tools.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_reference_parity.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_object_data_cache_follower_samples.py --self-test
 	./scripts/validate_object_data_cache_cold_stampede.sh --self-test

.config/reference-binary.toml

@@ -0,0 +1,20 @@
+# Pinned RustFS release used as the "reference" target of parity lanes
+# (.github/workflows/reference-parity.yml). This is the single place that names
+# the reference release; .github/actions/reference-binary resolves, downloads and
+# verifies it, and fails on a missing pin, a missing asset or a digest mismatch.
+#
+# Bumping the reference is a reviewed event: every per-test difference between
+# the candidate and the reference is measured against this exact build.
+#
+# Each sha256 is the value published in the release's SHA256SUMS asset, which
+# matches the digest GitHub reports for the same asset. The action checks both.
+
+schema = 1
+repository = "rustfs/rustfs"
+version = "1.0.5"
+tag = "1.0.5"
+
+[platforms.linux-x86_64-gnu]
+asset = "rustfs-linux-x86_64-gnu-v1.0.5.zip"
+sha256 = "5addae2f74756bf80104d0975e6347cda843bc2daffdfcd1f0cb80118b95cbd0"
+executable = "rustfs"

.config/reference-parity-targets.toml

@@ -0,0 +1,24 @@
+# Targets of the s3-tests reference parity lane (.github/workflows/reference-parity.yml).
+#
+# Exactly one target uses `binary = "reference"` (the release pinned in
+# .config/reference-binary.toml). Every other target uses `binary = "candidate"`
+# (the binary built from the branch under test) and is compared per node id
+# against the reference. All targets run the same harness, test list and server
+# configuration; `env` adds only the target's own RUSTFS_* settings.
+#
+# Adding a target is one more [[targets]] table, for example:
+#
+#   [[targets]]
+#   name = "candidate-gateway"
+#   binary = "candidate"
+#   env = { RUSTFS_S3_STACK = "gateway" }
+
+schema = 1
+
+[[targets]]
+name = "reference"
+binary = "reference"
+
+[[targets]]
+name = "candidate-legacy"
+binary = "candidate"

.config/reference-parity-tolerated.txt

@@ -0,0 +1,12 @@
+# Tolerated per-test differences of the s3-tests reference parity lane.
+#
+# One entry per line, four fields separated by " | ":
+#
+#   <target> | <pytest node id> | <tracking issue URL> | <reason>
+#
+# <target> is a candidate name from .config/reference-parity-targets.toml, the
+# node id is exactly as the lane prints it (s3tests/functional/test_s3.py::...),
+# and the issue URL is https://github.com/rustfs/<repo>/issues/<number>.
+#
+# Same semantics as scripts/junit_diff.py --allow: an entry whose difference no
+# longer occurs is stale and fails the lane, so remove it when it is fixed.

.config/s3s-footprint-baseline.txt

@@ -12,6 +12,6 @@
 # s3_error_lines  lines holding s3_error!, same scope, minus those the script
 #                 header's rule clears (rustfs_s3_types's macro); fails closed
 # ecstore_files   Rust files referencing s3s paths under crates/ecstore/src
-files=65
-s3_error_lines=219
+files=63
+s3_error_lines=217
 ecstore_files=0

.github/actions/reference-binary/action.yml

@@ -0,0 +1,96 @@
+# Copyright 2026 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.
+
+name: "Reference RustFS binary"
+description: "Download the pinned reference RustFS release, verify its sha256 and expose the executable path. A missing pin, missing asset or digest mismatch fails the step."
+
+inputs:
+  pin-file:
+    description: "Pin file naming the reference release (repository-relative)"
+    required: false
+    default: ".config/reference-binary.toml"
+  platform:
+    description: "Platform table in the pin file"
+    required: false
+    default: "linux-x86_64-gnu"
+  destination:
+    description: "Directory that receives the executable; defaults to $RUNNER_TEMP/reference-binary"
+    required: false
+    default: ""
+
+outputs:
+  path:
+    description: "Absolute path of the verified executable"
+    value: ${{ steps.unpack.outputs.path }}
+  version:
+    description: "Pinned reference version"
+    value: ${{ steps.pin.outputs.version }}
+  sha256:
+    description: "Pinned and verified archive sha256"
+    value: ${{ steps.pin.outputs.sha256 }}
+
+runs:
+  using: "composite"
+  steps:
+    - name: Resolve reference pin
+      id: pin
+      shell: bash
+      env:
+        PIN_TOOL: ${{ github.action_path }}/reference_binary.py
+        PIN_FILE: ${{ inputs.pin-file }}
+        PIN_PLATFORM: ${{ inputs.platform }}
+      run: |
+        set -euo pipefail
+        resolved="$(python3 "$PIN_TOOL" resolve --pin "$PIN_FILE" --platform "$PIN_PLATFORM")"
+        printf '%s\n' "$resolved"
+        printf '%s\n' "$resolved" >> "$GITHUB_OUTPUT"
+
+    - name: Download and verify reference archive
+      id: download
+      shell: bash
+      env:
+        GH_TOKEN: ${{ github.token }}
+        ASSET_DOWNLOADER: ${{ github.action_path }}/../setup/download_release_asset.py
+        REF_REPOSITORY: ${{ steps.pin.outputs.repository }}
+        REF_TAG: ${{ steps.pin.outputs.tag }}
+        REF_ASSET: ${{ steps.pin.outputs.asset }}
+        REF_SHA256: ${{ steps.pin.outputs.sha256 }}
+      run: |
+        set -euo pipefail
+        archive="$RUNNER_TEMP/reference-archive/$REF_ASSET"
+        # Fails unless the release lists exactly one asset of this name whose
+        # published digest equals the pin and whose downloaded bytes match it.
+        python3 "$ASSET_DOWNLOADER" --repository "$REF_REPOSITORY" --tag "$REF_TAG" \
+          --asset "$REF_ASSET" --output "$archive" --sha256 "$REF_SHA256"
+        echo "archive=$archive" >> "$GITHUB_OUTPUT"
+
+    - name: Unpack reference executable
+      id: unpack
+      shell: bash
+      env:
+        PIN_TOOL: ${{ github.action_path }}/reference_binary.py
+        REF_ARCHIVE: ${{ steps.download.outputs.archive }}
+        REF_SHA256: ${{ steps.pin.outputs.sha256 }}
+        REF_EXECUTABLE: ${{ steps.pin.outputs.executable }}
+        REF_DESTINATION: ${{ inputs.destination }}
+      run: |
+        set -euo pipefail
+        destination="${REF_DESTINATION:-$RUNNER_TEMP/reference-binary}"
+        unpacked="$(python3 "$PIN_TOOL" unpack --archive "$REF_ARCHIVE" --sha256 "$REF_SHA256" \
+          --executable "$REF_EXECUTABLE" --destination "$destination")"
+        path="${unpacked#path=}"
+        test -x "$path"
+        echo "Reference executable: $path"
+        "$path" --version
+        echo "path=$path" >> "$GITHUB_OUTPUT"

.github/actions/reference-binary/reference_binary.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+# Copyright 2026 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.
+"""Resolve and unpack the pinned reference RustFS release binary.
+
+`resolve` reads the pin file and prints the selected platform's identity as
+`key=value` lines (GitHub step-output format). `unpack` re-verifies the
+downloaded archive digest against the pin and extracts exactly one executable.
+Every missing or malformed input is a failure; nothing is skipped.
+"""
+
+import argparse
+import hashlib
+from pathlib import Path
+import re
+import sys
+import tomllib
+import zipfile
+
+REPOSITORY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*")
+TAG = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.+-]*")
+PLATFORM = re.compile(r"[a-z0-9][a-z0-9_-]*")
+ASSET = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.+-]*\.zip")
+SHA256 = re.compile(r"[0-9a-f]{64}")
+EXECUTABLE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*")
+MAX_EXECUTABLE_BYTES = 1024 * 1024 * 1024
+
+
+class PinError(Exception):
+    """A pin, archive or argument that cannot be trusted."""
+
+
+def field(table, key, pattern, where):
+    value = table.get(key)
+    if not isinstance(value, str) or not pattern.fullmatch(value):
+        raise PinError(f"{where}: missing or invalid '{key}'")
+    return value
+
+
+def resolve(pin_path, platform):
+    if not PLATFORM.fullmatch(platform or ""):
+        raise PinError("invalid platform name")
+    try:
+        data = tomllib.loads(Path(pin_path).read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise PinError(f"{pin_path}: reference pin file is missing") from exc
+    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
+        raise PinError(f"{pin_path}: cannot read reference pin: {exc}") from exc
+    if data.get("schema") != 1 or type(data.get("schema")) is not int:
+        raise PinError(f"{pin_path}: unsupported pin schema")
+    identity = {
+        "repository": field(data, "repository", REPOSITORY, pin_path),
+        "version": field(data, "version", TAG, pin_path),
+        "tag": field(data, "tag", TAG, pin_path),
+    }
+    platforms = data.get("platforms")
+    if not isinstance(platforms, dict) or not platforms:
+        raise PinError(f"{pin_path}: no pinned platforms")
+    entry = platforms.get(platform)
+    if not isinstance(entry, dict):
+        raise PinError(f"{pin_path}: no pin for platform '{platform}'")
+    where = f"{pin_path} [platforms.{platform}]"
+    identity["asset"] = field(entry, "asset", ASSET, where)
+    identity["sha256"] = field(entry, "sha256", SHA256, where)
+    identity["executable"] = field(entry, "executable", EXECUTABLE, where)
+    if identity["version"] not in identity["asset"]:
+        raise PinError(f"{where}: asset name does not carry version {identity['version']}")
+    unknown = set(entry) - {"asset", "sha256", "executable"}
+    if unknown:
+        raise PinError(f"{where}: unknown keys {sorted(unknown)}")
+    return identity
+
+
+def file_sha256(path):
+    with open(path, "rb") as stream:
+        return hashlib.file_digest(stream, "sha256").hexdigest()
+
+
+def unpack(archive, sha256, executable, destination):
+    if not SHA256.fullmatch(sha256 or "") or not EXECUTABLE.fullmatch(executable or ""):
+        raise PinError("invalid expected digest or executable name")
+    try:
+        actual = file_sha256(archive)
+    except OSError as exc:
+        raise PinError(f"{archive}: reference archive is missing or unreadable") from exc
+    if actual != sha256:
+        raise PinError(f"{archive}: sha256 {actual} does not match the pinned digest")
+    try:
+        with zipfile.ZipFile(archive) as bundle:
+            matches = [info for info in bundle.infolist() if info.filename == executable]
+            if len(matches) != 1:
+                raise PinError(f"{archive}: must contain exactly one top-level '{executable}'")
+            info = matches[0]
+            if info.is_dir() or not 0 < info.file_size <= MAX_EXECUTABLE_BYTES:
+                raise PinError(f"{archive}: '{executable}' is not a plausible executable")
+            destination = Path(destination)
+            destination.mkdir(parents=True, exist_ok=True)
+            target = destination / executable
+            with bundle.open(info) as source, open(target, "wb") as sink:
+                while chunk := source.read(1024 * 1024):
+                    sink.write(chunk)
+    except zipfile.BadZipFile as exc:
+        raise PinError(f"{archive}: not a valid zip archive") from exc
+    target.chmod(0o755)
+    return target
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    sub = parser.add_subparsers(dest="mode", required=True)
+    res = sub.add_parser("resolve")
+    res.add_argument("--pin", required=True)
+    res.add_argument("--platform", required=True)
+    unp = sub.add_parser("unpack")
+    unp.add_argument("--archive", required=True)
+    unp.add_argument("--sha256", required=True)
+    unp.add_argument("--executable", required=True)
+    unp.add_argument("--destination", required=True)
+    args = parser.parse_args(argv)
+    try:
+        if args.mode == "resolve":
+            for key, value in resolve(args.pin, args.platform).items():
+                print(f"{key}={value}")
+        else:
+            print(f"path={unpack(args.archive, args.sha256, args.executable, args.destination)}")
+    except PinError as exc:
+        print(f"reference binary: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.github/scheduled-validations.json

@@ -869,6 +869,33 @@
       "failure_owner": "watchdog",
       "origin": "baseline"
     },
+    {
+      "workflow": ".github/workflows/reference-parity.yml",
+      "name": "Reference Parity",
+      "roles": [
+        "pr",
+        "periodic",
+        "manual"
+      ],
+      "events": [
+        "pull_request",
+        "schedule",
+        "workflow_dispatch"
+      ],
+      "cadence": "nightly",
+      "timezone": "UTC",
+      "schedule": [
+        "37 1 * * *"
+      ],
+      "owner": "@overtrue",
+      "policy": "active",
+      "monitor": "workflow",
+      "max_age_hours": 36,
+      "result_owner": "reference-parity.yml",
+      "disposition": "s3-tests per-node-id parity of branch candidates against the pinned reference release",
+      "failure_owner": "watchdog",
+      "origin": "added"
+    },
     {
       "workflow": ".github/workflows/functional-chain-health.yml",
       "name": "Functional chain health",

.github/workflows/reference-parity.yml

@@ -0,0 +1,308 @@
+# Copyright 2026 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.
+
+# s3-tests reference parity lane (rustfs/backlog#2773).
+#
+# Runs the ceph/s3-tests implemented list (scripts/s3-tests/implemented_tests.txt)
+# against every target in .config/reference-parity-targets.toml and fails on
+# any per-node-id difference between a candidate and the reference:
+#
+#   - reference: the release pinned in .config/reference-binary.toml, downloaded
+#     and sha256-verified by .github/actions/reference-binary;
+#   - candidate targets: the debug binary built from this checkout exactly as
+#     ci.yml's build-rustfs-debug-binary job builds it for S3 Implemented Tests.
+#
+# Every target runs the same harness (scripts/s3-tests/run.sh from this
+# checkout), test list and server configuration on its own hosted runner. A
+# target's failures get one rerun against a fresh server; a case that passes it
+# is flaky, not a difference. scripts/s3-tests/reference_parity.py judges the
+# result and honours .config/reference-parity-tolerated.txt, whose entries must
+# each match a real difference.
+#
+# Report-only: not part of the Test and Lint aggregate. Scheduled failures are
+# owned by the scheduled-validation watchdog (.github/scheduled-validations.json).
+
+name: Reference Parity
+
+on:
+  pull_request:
+    paths:
+      - ".github/workflows/reference-parity.yml"
+      - ".github/actions/reference-binary/**"
+      - ".config/reference-binary.toml"
+      - ".config/reference-parity-targets.toml"
+      - ".config/reference-parity-tolerated.txt"
+      - "scripts/s3-tests/reference_parity.py"
+      - "scripts/junit_diff.py"
+  schedule:
+    - cron: "37 1 * * *"
+  workflow_dispatch:
+
+# Matches ci.yml: these variables are hashed into the rust-cache key, so the
+# candidate build reads the same ci-dev cache as build-rustfs-debug-binary.
+env:
+  CARGO_TERM_COLOR: always
+  RUST_BACKTRACE: 1
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  plan:
+    name: Plan parity targets
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    outputs:
+      targets: ${{ steps.targets.outputs.targets }}
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Validate target list
+        id: targets
+        run: |
+          set -euo pipefail
+          targets="$(python3 scripts/s3-tests/reference_parity.py targets --config .config/reference-parity-targets.toml)"
+          echo "targets=${targets}" >> "$GITHUB_OUTPUT"
+          echo "Parity targets: ${targets}"
+
+  build-candidate:
+    name: Build candidate binary
+    # Critical-path CPU and cache ownership: the same debug build as ci.yml's
+    # build-rustfs-debug-binary, reading the ci-dev cache written on sm-standard-4.
+    runs-on: sm-standard-4
+    timeout-minutes: 60
+    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'
+
+      - name: Build debug binary
+        run: python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks
+
+      - name: Upload candidate binary
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: reference-parity-candidate-binary
+          overwrite: true
+          path: |
+            target/debug/rustfs
+            target/debug/rustfs.e2e.json
+          if-no-files-found: error
+          retention-days: 1
+
+  s3tests:
+    name: s3-tests (${{ matrix.target.name }})
+    needs: [ plan, build-candidate ]
+    runs-on: ubuntu-latest
+    timeout-minutes: 75
+    strategy:
+      fail-fast: false
+      matrix:
+        target: ${{ fromJSON(needs.plan.outputs.targets) }}
+    env:
+      TARGET_NAME: ${{ matrix.target.name }}
+      TARGET_KIND: ${{ matrix.target.binary }}
+      # Above this many first-run failures the target is systemically broken;
+      # a rerun would only repeat it, so the first run is judged as is.
+      RERUN_CAP: "50"
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Fetch pinned reference binary
+        id: reference
+        if: matrix.target.binary == 'reference'
+        uses: ./.github/actions/reference-binary
+
+      - name: Download candidate binary
+        if: matrix.target.binary == 'candidate'
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          name: reference-parity-candidate-binary
+          path: target/debug
+
+      - name: Select target binary
+        env:
+          REFERENCE_PATH: ${{ steps.reference.outputs.path }}
+        run: |
+          set -euo pipefail
+          case "$TARGET_KIND" in
+            reference) binary="$REFERENCE_PATH" ;;
+            candidate) chmod +x target/debug/rustfs; binary="$PWD/target/debug/rustfs" ;;
+            *) echo "unknown target binary kind: $TARGET_KIND" >&2; exit 1 ;;
+          esac
+          test -x "$binary"
+          echo "TARGET_BINARY=$binary" >> "$GITHUB_ENV"
+          mkdir -p "parity/$TARGET_NAME"
+          {
+            echo "target=$TARGET_NAME"
+            echo "binary=$TARGET_KIND"
+            echo "sha256=$(sha256sum "$binary" | cut -d' ' -f1)"
+            echo "version=$("$binary" --version | tr '\n' ' ')"
+          } | tee "parity/$TARGET_NAME/target.txt"
+
+      - name: Apply target environment
+        env:
+          TARGET_JSON: ${{ toJSON(matrix.target) }}
+        run: |
+          set -euo pipefail
+          python3 scripts/s3-tests/reference_parity.py export-env --target-json "$TARGET_JSON" \
+            | tee -a "$GITHUB_ENV" "parity/$TARGET_NAME/target-env.txt"
+
+      - name: Run implemented s3-tests
+        run: |
+          set -uo pipefail
+          report="parity/$TARGET_NAME"
+          harness="artifacts/s3tests-single"
+
+          # Same invocation as ci.yml's S3 Implemented Tests, with a fresh data
+          # root and port per run. $1 optionally replaces the test list.
+          run_suite() {
+            local list="$1" label="$2" root port
+            root="${RUNNER_TEMP}/rustfs-s3tests-${label}"
+            rm -rf "$harness" "$root"
+            mkdir -p "$root"
+            port="$(python3 -c 'import socket; s=socket.socket(); s.bind(("127.0.0.1", 0)); print(s.getsockname()[1]); s.close()')"
+            env DEPLOY_MODE=binary \
+              RUSTFS_BINARY="$TARGET_BINARY" \
+              TEST_MODE=single \
+              MAXFAIL=0 \
+              XDIST=4 \
+              S3_PORT="$port" \
+              DATA_ROOT="$root" \
+              S3TESTS_CONF="$harness/s3tests.conf" \
+              ${list:+IMPLEMENTED_TESTS_FILE="$list"} \
+              ./scripts/s3-tests/run.sh
+          }
+
+          # A run is usable only if it produced a JUnit report; test failures
+          # are data for the comparison, a missing report is a lane failure.
+          keep() {
+            local prefix="$1"
+            for file in pytest.log compat-report.md; do
+              if [[ -f "$harness/$file" ]]; then cp "$harness/$file" "$report/${prefix}${file}"; fi
+            done
+            if [[ -f "$harness/rustfs-single/rustfs.log" ]]; then cp "$harness/rustfs-single/rustfs.log" "$report/${prefix}rustfs.log"; fi
+            test -s "$harness/junit.xml" || { echo "::error::$TARGET_NAME produced no JUnit report (${prefix:-first} run)"; return 1; }
+            cp "$harness/junit.xml" "$report/${prefix}junit.xml"
+          }
+
+          run_suite "" first
+          echo "first_run_exit=$?" >> "$report/target.txt"
+          keep "" || exit 1
+          cp "$harness/selected-nodeids.txt" "$report/selected-nodeids.txt" || exit 1
+
+          failures="$(python3 scripts/s3-tests/reference_parity.py failed --report "$report")" || exit 1
+          count="$(printf '%s' "$failures" | grep -c . || true)"
+          echo "first_run_failures=$count" >> "$report/target.txt"
+          if (( count == 0 )); then
+            exit 0
+          fi
+          if (( count > RERUN_CAP )); then
+            echo "::warning::$TARGET_NAME has $count failures (cap $RERUN_CAP); judging the first run without a rerun"
+            echo "rerun=skipped-over-cap" >> "$report/target.txt"
+            exit 0
+          fi
+          printf '%s\n' "$failures" > "$RUNNER_TEMP/rerun-tests.txt"
+          run_suite "$RUNNER_TEMP/rerun-tests.txt" rerun
+          echo "rerun_exit=$?" >> "$report/target.txt"
+          keep rerun- || exit 1
+
+      - name: Upload target report
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: reference-parity-report-${{ matrix.target.name }}
+          overwrite: true
+          path: parity/${{ matrix.target.name }}/
+          if-no-files-found: error
+          retention-days: 14
+
+  compare:
+    name: Compare targets
+    needs: [ plan, s3tests ]
+    if: always() && needs.plan.result == 'success'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Download target reports
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          pattern: reference-parity-report-*
+          path: parity-download
+
+      - name: Compare per node id
+        env:
+          TARGETS: ${{ needs.plan.outputs.targets }}
+          S3TESTS_RESULT: ${{ needs.s3tests.result }}
+        run: |
+          set -uo pipefail
+          mkdir -p parity
+          for dir in parity-download/reference-parity-report-*; do
+            if [[ -d "$dir" ]]; then mv "$dir" "parity/${dir#parity-download/reference-parity-report-}"; fi
+          done
+          python3 scripts/s3-tests/reference_parity.py compare \
+            --targets-json "$TARGETS" \
+            --reports parity \
+            --tolerated .config/reference-parity-tolerated.txt \
+            --markdown "$GITHUB_STEP_SUMMARY" \
+            --json parity/diff.json > parity/diff.md
+          status=$?
+          cat parity/diff.md
+          if [[ "$S3TESTS_RESULT" != "success" ]]; then
+            echo "::error::s3-tests target jobs finished with '$S3TESTS_RESULT'"
+            status=1
+          fi
+          exit "$status"
+
+      - name: Upload parity diff
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: reference-parity-diff
+          overwrite: true
+          path: |
+            parity/diff.md
+            parity/diff.json
+          if-no-files-found: warn
+          retention-days: 14

crates/ecstore/src/disk/local/delete_recovery.rs

@@ -41,19 +41,53 @@ struct DiscoverySession {
     touched: Instant,
 }
 
-/// Plain (non-transaction) children of an object-scoped discovery root cost one
-/// budget entry per this many names: enough to bound a hostile prefix without
-/// making a real folder prefix undeletable.
+/// Names an object-scoped scan skips without a stat (plain children of the
+/// key's directory, and entries of a UUID child that cannot be delete evidence)
+/// cost one budget entry per this many names: enough to bound a hostile prefix
+/// without making a real folder prefix undeletable.
 const PLAIN_CHILDREN_PER_BUDGET_ENTRY: u32 = 64;
 
+/// Entries one step of the resumable (background) discovery may visit. The
+/// foreground bounded protocol has its own, larger per-call allowance.
+const DISCOVERY_STEP_ENTRIES: u32 = 4096;
+
+/// Wall-clock allowance of one bounded discovery call: a third of the drive
+/// timeout that wraps the call (including its wait for a lane worker), capped at
+/// ten seconds, so an incomplete walk returns before the call can time out.
+fn bounded_discovery_time_limit() -> std::time::Duration {
+    (crate::disk::disk_store::get_max_timeout_duration() / 3).min(std::time::Duration::from_secs(10))
+}
+
+/// Whether an entry inside a UUID child of an object-scoped root can name one
+/// of the key's delete transactions: a journal file of a rollback directory or
+/// a data-directory marker. Every other name is inert to discovery.
+fn may_be_delete_evidence(name: &str) -> bool {
+    [PREPARED, STORAGE_FORMAT_FILE_BACKUP, DELETE_QUORUM_COMMITTED_FILE].contains(&name)
+        || [
+            DELETE_DATA_DIR_MARKER_PREFIX,
+            RESERVED_DELETE_DATA_DIR_MARKER_PREFIX,
+            RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX,
+        ]
+        .iter()
+        .any(|prefix| delete_transaction_marker_id(name, prefix).is_some())
+}
+
 impl DiscoverySession {
     /// Consume one bounded step, returning ownership only when traversal remains.
     fn advance(self, limit: u32) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>)> {
-        self.advance_bounded(limit, super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES)
+        self.advance_bounded(limit, DISCOVERY_STEP_ENTRIES, &|| false)
             .map(|(transactions, next, _)| (transactions, next))
     }
 
-    fn advance_bounded(mut self, limit: u32, max_entries: u32) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>, u32)> {
+    /// Walk until `limit` transactions, `max_entries` charged entries, the end
+    /// of the scope, or `should_stop` (checked before every entry); any early
+    /// stop returns the remaining traversal as continuation state.
+    fn advance_bounded(
+        mut self,
+        limit: u32,
+        max_entries: u32,
+        should_stop: &dyn Fn() -> bool,
+    ) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>, u32)> {
         if self.frames.is_empty() {
             match std::fs::read_dir(&self.root) {
                 Ok(entries) => self.frames.push((String::new(), entries)),
@@ -66,10 +100,11 @@ impl DiscoverySession {
         let mut transactions = BTreeSet::new();
         let mut seen_candidates = HashSet::new();
         let mut visited = 0u32;
-        let mut plain_children = 0u32;
+        let mut skipped_by_name = 0u32;
         while !self.frames.is_empty()
             && transactions.len() < usize::try_from(limit).map_err(|_| std::io::Error::from(ErrorKind::InvalidInput))?
             && visited < max_entries
+            && !should_stop()
         {
             let (parent, entries) = self
                 .frames
@@ -90,13 +125,22 @@ impl DiscoverySession {
                 return Err(std::io::Error::from(ErrorKind::InvalidData));
             }
             // An object-scoped scan only ever descends into transaction (UUID)
-            // directories at the root. Plain child keys of a prefix directory are
-            // skipped by name without a stat; they are charged at one entry per
-            // readdir batch so the work stays bounded while deleting a missing key
-            // that names a prefix with thousands of children can still complete.
-            if self.object.is_some() && parent.is_empty() && !canonical_uuid(&name).is_some_and(|id| !id.is_nil()) {
-                plain_children += 1;
-                if plain_children.is_multiple_of(PLAIN_CHILDREN_PER_BUDGET_ENTRY) {
+            // directories at the root, and inside them only journal files and
+            // data-directory markers can be evidence. Every other name (plain
+            // child keys of a prefix directory, and the `xl.meta`, data
+            // directories or children of a UUID-named child key) is skipped by
+            // name without a stat and charged at one entry per readdir batch, so
+            // the work stays bounded while deleting a missing key that names a
+            // folder prefix with many children, UUID-named or not, can complete.
+            if self.object.is_some()
+                && if parent.is_empty() {
+                    !canonical_uuid(&name).is_some_and(|id| !id.is_nil())
+                } else {
+                    !may_be_delete_evidence(&name)
+                }
+            {
+                skipped_by_name += 1;
+                if skipped_by_name.is_multiple_of(PLAIN_CHILDREN_PER_BUDGET_ENTRY) {
                     visited += 1;
                 }
                 continue;
@@ -1391,8 +1435,16 @@ impl LocalDisk {
         self.validate_delete_discovery_owner(&session)?;
         // The blocking owner drops every directory handle even if the RPC waiter disappears.
         // A partial inventory is never evidence of absence and never enters DISCOVERY_SESSIONS.
-        os::run_blocking_namespace_operation(lease, move || {
-            let (transactions, next, visited) = session.advance_bounded(256, max_entries)?;
+        // A large folder prefix may take a while to walk, so the probe rides the
+        // durable-mutation lane like bounded inspection, never the one-worker
+        // foreground lane that every other mutation on the drive queues behind.
+        // It also stops, incomplete, when its waiter goes away or well before the
+        // drive timeout, so a slow walk answers SlowDown instead of marking the
+        // drive failed or holding a lane worker after the request is gone.
+        let deadline = Instant::now() + bounded_discovery_time_limit();
+        os::run_cancellable_blocking_namespace_operation_with_sync(lease, true, move |is_cancelled| {
+            let should_stop = || is_cancelled() || Instant::now() >= deadline;
+            let (transactions, next, visited) = session.advance_bounded(256, max_entries, &should_stop)?;
             Ok(super::super::BoundedDeleteRecoveryDiscovery {
                 transactions: if next.is_none() { transactions } else { Vec::new() },
                 visited,
@@ -6561,6 +6613,93 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn delete_recovery_bounded_discovery_charges_uuid_named_children_once_and_keeps_evidence() {
+        let (_temp, disk, _, _) = fixture().await;
+        let path = disk.delete_discovery_path("recovery", Some("prefix")).expect("prefix path");
+        std::fs::create_dir_all(&path).expect("physical prefix");
+        // UUID-named child keys of the folder: each holds its own `xl.meta` and a
+        // data directory, neither of which can be evidence for the folder's key.
+        const CHILDREN: u32 = 2000;
+        for _ in 0..CHILDREN {
+            let child = path.join(Uuid::new_v4().to_string());
+            std::fs::create_dir_all(child.join(Uuid::new_v4().to_string())).expect("child key data dir");
+            std::fs::write(child.join(STORAGE_FORMAT_FILE), b"meta").expect("child key metadata");
+        }
+        // A UUID-named sub-prefix with children of its own.
+        const SUB_PREFIX_CHILDREN: u32 = 640;
+        let sub_prefix = path.join(Uuid::new_v4().to_string());
+        for index in 0..SUB_PREFIX_CHILDREN {
+            std::fs::create_dir_all(sub_prefix.join(format!("child-{index}"))).expect("sub-prefix child");
+        }
+        // Evidence of two interrupted transactions of the key itself: a journaled
+        // rollback directory, and a data-directory marker whose rollback is gone.
+        let journaled = Uuid::new_v4();
+        std::fs::create_dir_all(path.join(journaled.to_string())).expect("rollback directory");
+        std::fs::write(path.join(journaled.to_string()).join(PREPARED), b"{}").expect("journal");
+        let marker_only = Uuid::new_v4();
+        let data_dir = path.join(Uuid::new_v4().to_string());
+        std::fs::create_dir_all(&data_dir).expect("residual data directory");
+        std::fs::write(data_dir.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{marker_only}")), [])
+            .expect("reservation marker");
+
+        let page = disk
+            .discover_delete_transactions_bounded(
+                "recovery",
+                "prefix",
+                super::super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES,
+            )
+            .await
+            .expect("bounded probe");
+        assert!(page.complete, "a folder of UUID-named keys is a complete inventory");
+        // One entry per descent (children, sub-prefix, rollback and data
+        // directory) and per evidence file; inert names at 1/64.
+        let inert = 2 * CHILDREN + SUB_PREFIX_CHILDREN;
+        assert_eq!(page.visited, CHILDREN + 3 + 2 + inert / PLAIN_CHILDREN_PER_BUDGET_ENTRY);
+        let found = page.transactions.iter().map(|id| id.transaction).collect::<BTreeSet<_>>();
+        assert!(found.contains(&journaled), "a journaled rollback directory is evidence");
+        assert!(found.contains(&marker_only), "a marker-only residue is evidence");
+        assert_eq!(found.len(), 2, "no child key is mistaken for a transaction");
+    }
+
+    #[tokio::test]
+    async fn delete_recovery_bounded_discovery_stops_on_request_and_skips_only_inert_irregular_entries() {
+        let (_temp, disk, _, _) = fixture().await;
+        let path = disk.delete_discovery_path("recovery", Some("prefix")).expect("prefix path");
+        let child = path.join(Uuid::new_v4().to_string());
+        std::fs::create_dir_all(&child).expect("UUID-named child key");
+        // A stop request (cancellation or the wall-clock limit) ends the walk
+        // before any entry, leaving continuation state: never evidence of absence.
+        let session = disk
+            .new_delete_discovery_session("recovery", Some("prefix"), &path)
+            .expect("private session");
+        let (transactions, next, visited) = session
+            .advance_bounded(256, super::super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES, &|| true)
+            .expect("stopped walk");
+        assert!(transactions.is_empty());
+        assert_eq!(visited, 0);
+        assert!(next.is_some(), "a stopped walk is incomplete");
+        drop(next);
+        #[cfg(unix)]
+        {
+            // An irregular entry whose name cannot be evidence is skipped unread.
+            std::os::unix::fs::symlink(&path, child.join("inert-link")).expect("inert symlink");
+            let page = disk
+                .discover_delete_transactions_bounded("recovery", "prefix", 4096)
+                .await
+                .expect("inert irregular entries are not evidence");
+            assert!(page.complete && page.transactions.is_empty());
+            // One with an evidence name still fails closed.
+            std::os::unix::fs::symlink(&path, child.join(PREPARED)).expect("evidence-named symlink");
+            assert!(
+                disk.discover_delete_transactions_bounded("recovery", "prefix", 4096)
+                    .await
+                    .is_err(),
+                "an evidence-named symlink is corruption, not a skipped name"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn delete_recovery_bounded_discovery_refuses_partial_inventory_without_public_cursor() {
         let (_temp, disk, _, _) = fixture().await;
@@ -6663,8 +6802,11 @@ mod tests {
         let filler_id = Uuid::new_v4().to_string();
         let filler = path.join(&filler_id);
         fs::create_dir_all(&filler).await.expect("non-evidence directory");
-        for index in 0..5000 {
-            fs::write(filler.join(format!("part-{index:04}")), [])
+        // Inert names are skipped at 1/64, so fill the step with entries that
+        // pass the evidence-name filter but are directories: each costs a full
+        // budget entry and names no transaction.
+        for _ in 0..5000 {
+            fs::create_dir(filler.join(format!("{DELETE_DATA_DIR_MARKER_PREFIX}{}", Uuid::new_v4())))
                 .await
                 .expect("non-evidence entry");
         }

crates/ecstore/src/disk/mod.rs

@@ -2585,7 +2585,13 @@ pub struct BoundedDeleteRecoveryDiscovery {
     pub complete: bool,
 }
 
-pub const DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES: u32 = 4096;
+/// Entries one bounded discovery call may charge on one drive. Object-scoped
+/// discovery charges each transaction-shaped (UUID) child of the key's
+/// directory once and every name that cannot be delete evidence at
+/// 1/`PLAIN_CHILDREN_PER_BUDGET_ENTRY`, so this admits a missing-key delete of a
+/// folder prefix with this many UUID-named children per drive. The request
+/// budget (`DeleteRecoveryWorkBudget`) still bounds the total across keys.
+pub const DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES: u32 = 262_144;
 pub const DELETE_RECOVERY_MAX_INSPECTION_ENTRIES: u32 = 65536;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

crates/ecstore/src/object_api/types.rs

@@ -975,9 +975,11 @@ pub struct DeleteRecoveryWorkBudget {
     remaining: std::sync::atomic::AtomicU32,
 }
 
-/// Entries one disk may visit per recovery step (discovery plus inspection).
-/// Every disk of every set a request fans out to is admitted up to this much
-/// and charged on completion, so a budget must cover that many disks.
+/// Per-drive unit a request budget is sized by (a typical discovery plus one
+/// inspection). Admission charges what a call visited, not this unit, so it only
+/// scales the request total with the drives a request can reach; it is kept
+/// independent of the per-call caps, which may let one key's probe use more of
+/// the same total on a large folder.
 const DELETE_RECOVERY_RESERVATION_PER_DISK: u32 = 4096 + 65_536;
 
 impl Default for DeleteRecoveryWorkBudget {

crates/ecstore/src/set_disk/mod.rs

@@ -11040,7 +11040,12 @@ mod tests {
         let live = Uuid::new_v4();
         let orphan = Uuid::new_v4();
         let object_dir = dir.path().join("bucket/obj");
-        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await;
+        // Part files are inert to discovery and cost 1/64 of an entry, so the
+        // first step is exhausted by what still costs a full entry each: the
+        // data directories of many versions, all referenced by the metadata.
+        let versions = (0..4200).map(|_| Uuid::new_v4()).collect::<Vec<_>>();
+        let referenced = std::iter::once(live).chain(versions.iter().copied()).collect::<Vec<_>>();
+        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &referenced).await;
         let live_dir = object_dir.join(live.to_string());
         let orphan_dir = object_dir.join(orphan.to_string());
         fs::create_dir_all(&live_dir).await.expect("create live multipart directory");
@@ -11049,10 +11054,14 @@ mod tests {
             .await
             .expect("write orphan payload");
         let parts_dir = live_dir.clone();
+        let versions_root = object_dir.clone();
         tokio::task::spawn_blocking(move || {
             for part in 1..=4200 {
                 std::fs::write(parts_dir.join(format!("part.{part}")), b"live").expect("write live multipart entry");
             }
+            for version in versions {
+                std::fs::create_dir(versions_root.join(version.to_string())).expect("create referenced version directory");
+            }
         })
         .await
         .expect("create multipart fixture");
@@ -11062,7 +11071,10 @@ mod tests {
             .await
             .expect("read first discovery page");
         assert!(first.transactions.is_empty());
-        assert!(first.next.is_some(), "ordinary multipart entries must exhaust the first scan budget");
+        assert!(
+            first.next.is_some(),
+            "the data directories of many versions must exhaust the first scan step"
+        );
         let last = disk
             .discover_delete_transactions("bucket", Some("obj"), first.next, 1)
             .await

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

@@ -1709,6 +1709,37 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn delete_recovery_missing_key_naming_a_folder_of_uuid_named_keys_completes() {
+        let bucket = "recover-uuid-named-folder";
+        let (dirs, store, _) = fixture(bucket).await;
+        // More UUID-named child keys per drive than the former 4096-entry probe
+        // could cover once each child's `xl.meta` and data directory were charged
+        // (three entries per child).
+        const CHILDREN: usize = 1500;
+        for dir in &dirs {
+            let prefix = dir.path().join(bucket).join("uploads");
+            for _ in 0..CHILDREN {
+                let child = prefix.join(Uuid::new_v4().to_string());
+                std::fs::create_dir_all(child.join(Uuid::new_v4().to_string())).expect("child key data dir");
+                std::fs::write(child.join(crate::disk::STORAGE_FORMAT_FILE), b"meta").expect("child key metadata");
+            }
+        }
+        let error = store
+            .delete_object(bucket, "uploads", ObjectOptions::default())
+            .await
+            .expect_err("the folder's own key does not exist");
+        assert!(
+            crate::error::is_err_object_not_found(&error),
+            "a complete inventory keeps the missing-key no-op semantics: {error:?}"
+        );
+        assert!(dirs.iter().all(|dir| {
+            std::fs::read_dir(dir.path().join(bucket).join("uploads"))
+                .map(|entries| entries.count() == CHILDREN)
+                .unwrap_or(false)
+        }));
+    }
+
     #[tokio::test]
     async fn delete_recovery_batch_budget_survives_pool_and_set_fanout() {
         use crate::layout::endpoint::Endpoint;

crates/scanner/src/scanner_folder.rs

@@ -425,7 +425,10 @@ enum FolderScanSource {
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum FolderScanMode {
     Namespace,
-    ErasureCandidate { reuse_namespace: bool },
+    ErasureCandidate {
+        reuse_namespace: bool,
+        parent_has_metadata: bool,
+    },
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -2065,7 +2068,9 @@ impl FolderScanner {
                     if candidate_unclassified {
                         // Apply namespace reuse only after the current listing
                         // proves there is no direct erasure part.
-                        if let FolderScanMode::ErasureCandidate { reuse_namespace: true } = mode
+                        if let FolderScanMode::ErasureCandidate {
+                            reuse_namespace: true, ..
+                        } = mode
                             && !into.compacted
                             && self.old_cache.find(&folder.name).is_some()
                         {
@@ -2079,10 +2084,23 @@ impl FolderScanner {
                             return Ok(FolderScanOutcome::Namespace);
                         }
                         candidate_unclassified = false;
-                        if !self.budget.try_start_directory() {
-                            return Err(ScannerError::Other("Operation cancelled".to_string()));
+                        // An empty UUID directory beside object metadata has
+                        // no namespace work. Its listing was already charged;
+                        // nonempty namespaces still need directory admission.
+                        let empty_object_child = matches!(
+                            mode,
+                            FolderScanMode::ErasureCandidate {
+                                parent_has_metadata: true,
+                                ..
+                            }
+                        ) && deferred_metadata.is_none()
+                            && queued_folders.is_empty();
+                        if !empty_object_child {
+                            if !self.budget.try_start_directory() {
+                                return Err(ScannerError::Other("Operation cancelled".to_string()));
+                            }
+                            wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
                         }
-                        wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
                     }
                     let Some(entry) = deferred_metadata.take() else { break };
                     (entry, true)
@@ -2552,20 +2570,6 @@ impl FolderScanner {
                 return Err(ScannerError::Other("Operation cancelled".to_string()));
             }
 
-            if found_object_metadata && self.is_erasure_mode {
-                // If we found an object in erasure mode, we skip subdirs (only datadirs)...
-                debug!(
-                    target: "rustfs::scanner::folder",
-                    event = EVENT_SCANNER_FOLDER_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_FOLDER,
-                    folder = %folder.name,
-                    state = "erasure_object_found",
-                    "Scanner folder descent stopped after erasure object"
-                );
-                break;
-            }
-
             // Checkpointed sweeps need child entries to validate their resume
             // frontier (#8292). Outside that, also compact a wide bucket root:
             // excluding it left #8961's ~170k top-level prefixes unbounded until
@@ -2654,6 +2658,7 @@ impl FolderScanner {
                 let mut child_mode = if self.is_erasure_mode && is_erasure_candidate(&folder_item.name) {
                     FolderScanMode::ErasureCandidate {
                         reuse_namespace: reuse_covered,
+                        parent_has_metadata: found_object_metadata,
                     }
                 } else {
                     FolderScanMode::Namespace
@@ -2721,7 +2726,10 @@ impl FolderScanner {
                                 }
                                 // Sampling may skip namespace work, never the
                                 // classification that protects its parent object.
-                                child_mode = FolderScanMode::ErasureCandidate { reuse_namespace: true };
+                                child_mode = FolderScanMode::ErasureCandidate {
+                                    reuse_namespace: true,
+                                    parent_has_metadata: found_object_metadata,
+                                };
                             } else {
                                 folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                             }
@@ -2816,6 +2824,11 @@ impl FolderScanner {
                         updates.cache.delete_recursive(&h);
                     }
                     into.children.remove(&clone_cache_value(&h.0));
+                    // Object keys can also be prefixes. A classified payload
+                    // directory must not hide their other namespace children.
+                    if found_object_metadata {
+                        continue;
+                    }
                     // Earlier namespace siblings may have completed, but none
                     // can certify coverage of this unresolved parent object.
                     self.coverage_gap |= forward_sweep;

crates/scanner/src/scanner_folder/tests/erasure_probe.rs

@@ -183,6 +183,117 @@ fn charged_ops(root: &std::path::Path) -> u64 {
     status.charged_ops
 }
 
+#[tokio::test]
+#[serial]
+async fn erasure_object_metadata_preserves_child_object_namespaces() {
+    for (corrupt_parent, cached_parent) in [(false, false), (true, false), (false, true)] {
+        let (mut scanner, root) = build_test_scanner().await;
+        let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
+        scanner.old_cache.info.name = "bucket".into();
+        scanner.new_cache.info.name = "bucket".into();
+        scanner.is_erasure_mode = true;
+        bind_budget(&mut scanner, &root, 1_000_000);
+
+        let namespace = format!("parent/{}/nested", Uuid::new_v4());
+        for object in ["parent", "parent/child", namespace.as_str()] {
+            write_test_object_metadata_bytes(&root, "bucket", object, &metadata_for_object_version("bucket", object, None)).await;
+        }
+        let payload = root.join("bucket/parent").join(Uuid::new_v4().to_string());
+        tokio::fs::create_dir_all(&payload)
+            .await
+            .expect("create parent payload directory");
+        tokio::fs::write(payload.join("part.1"), b"shard")
+            .await
+            .expect("create actual payload evidence");
+        if corrupt_parent {
+            tokio::fs::write(root.join("bucket/parent/xl.meta"), b"")
+                .await
+                .expect("corrupt only the parent metadata");
+        }
+        if cached_parent {
+            scanner.old_cache.replace("bucket", "", DataUsageEntry::default());
+            scanner.old_cache.replace(
+                "bucket/parent",
+                "bucket",
+                DataUsageEntry {
+                    objects: 1,
+                    size: 1,
+                    compacted: true,
+                    ..Default::default()
+                },
+            );
+            let interval = data_usage_update_dir_cycles();
+            scanner.old_cache.info.next_cycle = u64::from(
+                (0..interval)
+                    .find(|cycle| hash_path("bucket/parent").mod_(*cycle, interval))
+                    .expect("one normal sampling cycle revisits the old parent-only cache"),
+            );
+        }
+
+        let mut into = DataUsageEntry::default();
+        scanner
+            .scan_folder(
+                CancellationToken::new(),
+                CachedFolder {
+                    name: "bucket".into(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut into,
+            )
+            .await
+            .expect("scan object and child namespaces independently");
+        let total = scanner.new_cache.size_recursive("bucket").expect("retain bucket accounting");
+        assert_eq!(total.objects, if corrupt_parent { 2 } else { 3 });
+        assert_eq!(total.size, if corrupt_parent { 2 } else { 3 });
+        assert_eq!(total.failed_objects, usize::from(corrupt_parent));
+        assert_eq!(
+            scanner.new_cache.info.failed_objects.len(),
+            usize::from(corrupt_parent),
+            "payload evidence must not invent missing parent metadata"
+        );
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_object_prefix_children_keep_the_object_budget() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
+    scanner.old_cache.info.name = "bucket".into();
+    scanner.new_cache.info.name = "bucket".into();
+    scanner.is_erasure_mode = true;
+    bind_budget(&mut scanner, &root, 1_000_000);
+    for object in ["parent", "parent/child", "parent/other"] {
+        write_test_object_metadata(&root, "bucket", object).await;
+    }
+    let visits = metadata_visits::observe(&mut scanner);
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        crate::scanner_budget::ScannerCycleBudgetConfig {
+            max_objects: Some(2),
+            ..Default::default()
+        },
+    );
+    scanner.budget = budget.clone();
+    let result = scanner
+        .scan_folder(
+            budget.token(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        )
+        .await;
+    assert!(result.is_err(), "parent metadata cannot hide children or bypass their budget");
+    assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Objects));
+    assert_eq!(budget.progress().0, 2);
+    assert_eq!(visits.load(Ordering::Relaxed), 2);
+}
+
 #[tokio::test]
 #[serial]
 async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
@@ -204,7 +315,10 @@ async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
                 ctx,
                 folder,
                 &mut DataUsageEntry::default(),
-                FolderScanMode::ErasureCandidate { reuse_namespace: false },
+                FolderScanMode::ErasureCandidate {
+                    reuse_namespace: false,
+                    parent_has_metadata: false,
+                },
                 false,
             )
             .await
@@ -271,7 +385,10 @@ async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await;
@@ -311,7 +428,10 @@ async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await;
@@ -382,7 +502,16 @@ async fn erasure_candidate_defers_metadata_until_part_classification_finishes()
     scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
     let mut into = DataUsageEntry::default();
     let outcome = scanner
-        .scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false }, false)
+        .scan_folder_inner(
+            ctx,
+            folder,
+            &mut into,
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
+            false,
+        )
         .await
         .expect("classify candidate");
     assert_eq!(outcome, FolderScanOutcome::ErasureData);
@@ -594,7 +723,10 @@ async fn erasure_candidate_part_symlink_is_not_data_evidence() {
             CancellationToken::new(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await

docs/architecture/s3-compatibility-matrix.md

@@ -130,7 +130,7 @@ These behaviors are client-visible and differ from AWS S3 or MinIO. They bound h
 | Bucket tag key or value outside the AWS tag character class (`[\p{L}\p{N}\p{Z}+\-=._:/@]`) | `400 InvalidTag` on `PutBucketTagging` | AWS enforces the same class; MinIO is more permissive | AWS parity. Legacy stored tags still load. |
 | `CopyObject` with the default `x-amz-tagging-directive: COPY` from a source that carries tags | `403 AccessDenied` unless the caller also holds `s3:GetObjectTagging` (or `s3:GetObjectVersionTagging`) on the source; the destination's `s3:RequestObjectTag/*` conditions are evaluated against the copied tag set | AWS and MinIO copy the tags with `s3:GetObject` alone | A copy would otherwise disclose source tags to a principal that may not read them, and source tags could satisfy destination conditions the caller cannot assert (rustfs/backlog#3034). Use `REPLACE` with an explicit `x-amz-tagging` to copy without the tagging grant. |
 | `ListObjectVersions` / `ListObjectsV2` with the include-deleted extension | Requires `s3:ListBucketVersions` | Vendor extension | Delete markers and non-current versions are version-listing data. |
-| `DeleteObject` of a missing key that names a prefix | `204 NoContent` when the recovery probe of the prefix fits the request budget (`DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES`, 4096 entries per drive, one entry per 64 plain children); `503 SlowDown` beyond that, deterministically, and for the duration of a rolling upgrade while a drive in the set still runs a release without the bounded recovery RPC | `204 NoContent` | The probe looks for interrupted delete transactions under the prefix and is bounded so a hostile prefix cannot pin the request. Prefixes with more than 262,144 direct children per drive, or roughly 1,400 to 4,096 UUID-named children, exceed the bound (rustfs/backlog#3038 residual). |
+| `DeleteObject` of a missing key that names a prefix | `204 NoContent` when the recovery probe of the prefix fits its per-drive allowance (`DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES`, 262,144 entries: one per UUID-named direct child, one per 64 other names), its per-call time limit (a third of the drive timeout, at most 10 seconds) and the request budget (`DeleteRecoveryWorkBudget`, shared by every key and drive of the request); `503 SlowDown` beyond that, and for the duration of a rolling upgrade while a drive in the set still runs a release without the bounded recovery RPC | `204 NoContent` | The probe looks for interrupted delete transactions of the key and is bounded so a hostile prefix cannot pin the request. A folder completes with up to 262,144 UUID-named direct children or about 16.7 million other direct children per drive when the drive can walk them within the time limit (a warm cache walks roughly 25,000 UUID-named children per second; a cold HDD far fewer); inside a UUID-named child only journal files and data-directory markers are examined, so a child key's own `xl.meta`, data directories or children cost 1/64 each. |
 | `PutObject` / `UploadPart` body delivering less than 64 KiB per 60 seconds of server wait | `400 RequestTimeout` (the minimum-progress floor, logged as `upload_body_progress_stalled`); independent of `RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT`, including when that timeout is 0 | No minimum upload rate | A trickle upload otherwise pins upload admission and its buffers indefinitely; roughly 1.1 KiB/s is below any practical client. |
 | First write after `SetBucketQuota`, or after the quota changes, on a bucket without an authoritative usage ledger | `503 SlowDown` (retryable) until a background scan of the bucket publishes the exact ledger; a failed scan retries after 30 seconds | AWS has no bucket quota; MinIO admits with unknown usage | Quota admission uses only the authoritative ledger (scanner reporting is informational, see `reporting_retirement` tests), and the request must not hold its bucket metadata guard across a whole-bucket listing. The window is one bucket listing long. |
 

docs/testing/ci-gates.md

@@ -101,6 +101,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 | `coverage.yml` (weekly) | `coverage` | report-only trend; lcov and JSON artifact | yes (disabled) | `make coverage` |
 | `runner-hygiene.yml` (monthly) | `check-ephemerality` | runner ephemerality | yes | dispatch |
 | `e2e-upgrade.yml` (weekly, and on version tags) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
+| `reference-parity.yml` (nightly, and on PRs that change the lane's own files) | `plan`, `build-candidate`, `s3tests` (one job per target in `.config/reference-parity-targets.toml`), `compare` | report-only parity gate: fails on any per-node-id s3-tests difference between a branch candidate and the release pinned in `.config/reference-binary.toml` that `.config/reference-parity-tolerated.txt` does not list, and on a stale tolerated entry; per-target JUnit, logs and `diff.md`/`diff.json` | yes | `DEPLOY_MODE=binary RUSTFS_BINARY=<binary> MAXFAIL=0 XDIST=4 ./scripts/s3-tests/run.sh` once per target, then `python3 scripts/s3-tests/reference_parity.py compare` as in the `compare` job |
 | `oidc-keycloak.yml` (weekly) | `oidc-keycloak-live` | live OIDC gate | yes | see the PR row |
 | `targets-integration.yml` (nightly) | `targets-live` | live target gate; container logs | yes (disabled) | see the PR row |
 | `gateway-integration-sync.yml` (weekdays) | `sync` | merges `main` into `gateway/integration` as a `--no-ff` merge commit; a conflicting merge is aborted and opens a `sync/<source>-<date>` pull request listing the conflicting files, which keeps the run green; a rejected push or missing pull request fails it | yes | dispatch, optionally with `source_ref` |

rustfs/src/admin/handlers/account.rs

@@ -165,7 +165,8 @@ impl Operation for ChangeOwnPasswordHandler {
         let signing_secret = req
             .credentials()
             .ok_or_else(|| s3::error(S3ErrorCode::AccessDenied, "authentication required"))?
-            .secret_key()
+            .require_secret_key()
+            .map_err(|err| s3::error(S3ErrorCode::InternalError, err.to_string()))?
             .to_string();
         let caller = CallerIdentity::resolve_envelope(req.envelope()).await?;
         let audit = AccountAuditContext::from_envelope(req.envelope());

rustfs/src/admin/handlers/bucket_recovery.rs

@@ -63,8 +63,10 @@ impl Operation for RecoverOrphanedBucketHandler {
         authorize_admin_envelope(req.envelope(), vec![Action::AdminAction(AdminAction::RecoverOrphanedBucketAction)]).await?;
         let secret_key = req
             .credentials()
-            .map(|credentials| Zeroizing::new(credentials.secret_key().to_owned()))
-            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "authentication required"))?;
+            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "authentication required"))?
+            .require_secret_key()
+            .map(|secret| Zeroizing::new(secret.to_owned()))
+            .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?;
         let bucket = params
             .get("bucket")
             .filter(|bucket| !bucket.is_empty())

rustfs/src/admin/handlers/object_zip_download.rs

@@ -15,11 +15,12 @@
 use crate::admin::operation::{AdminRequest, AdminResponse, Operation};
 use crate::admin::router::{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH, AdminOperation, S3Router};
 use crate::admin::runtime_sources::{ServerContextSlot, current_action_credentials, current_object_store_handle, current_region};
-use crate::admin::storage_api::access::{ReqInfo, authorize_request};
+use crate::admin::storage_api::access::{ReqInfo, authorize_owned_request};
 use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
 use crate::admin::storage_api::contract::list::ListOperations as _;
 use crate::admin::storage_api::contract::object::{ObjectIO as _, ObjectOperations as _};
 use crate::admin::storage_api::object::StorageObjectOptions as ObjectOptions;
+use crate::app::{RequestEnvelope, RequestHead};
 use crate::auth::{check_key_valid, get_session_token};
 use crate::error::ApiError;
 use crate::license::license_check;
@@ -495,32 +496,39 @@ async fn authorize_object_zip_download_s3_action(
         req_info.version_id = None;
     }
 
-    let mut authorization = s3_authorization_request(
-        req.method().clone(),
-        req.uri().clone(),
-        &authenticated_zip_download_auth_context(req),
-        req.envelope().server_context().cloned(),
-        req.req_info().cloned(),
+    let mut extensions = http::Extensions::new();
+    extensions.insert(req.remote_addr().map(RemoteAddr));
+    if let Some(client_info) = req.client_info() {
+        extensions.insert(client_info.clone());
+    }
+    if let Some(server_context) = req.envelope().server_context() {
+        extensions.insert(server_context.clone());
+    }
+    if let Some(req_info) = req.req_info() {
+        extensions.insert(req_info.clone());
+    }
+    let mut authorization = RequestEnvelope::from_parts(
+        RequestHead {
+            method: req.method().clone(),
+            uri: req.uri().clone(),
+            headers: req.headers().clone(),
+            credentials: None,
+            region: None,
+            service: None,
+        },
+        extensions,
     );
-    authorize_request(&mut authorization, Action::S3Action(action))
-        .await
-        .map_err(Into::into)
+    authorize_owned_request(&mut authorization, Action::S3Action(action), None).await
 }
 
-/// The S3 gate (`storage::access::authorize_request`) still evaluates the s3s
-/// request shape, so the facts it reads are copied into one for the duration
-/// of the call: the authenticated `ReqInfo`, the peer and proxy-reported client
-/// addresses, the headers and query the policy conditions come from, and the
-/// server slot that selects the store. The gate reads neither the presented
-/// credentials nor the region or service, so those stay unset, and nothing it
-/// attaches to the request is read afterwards.
-fn s3_authorization_request(
+/// Builds the owned authorization request used when a download token is redeemed.
+fn token_authorization_request(
     method: Method,
     uri: Uri,
     auth_context: &ObjectZipDownloadAuthContext,
     server_context: Option<Arc<ServerContextSlot>>,
     req_info: Option<ReqInfo>,
-) -> s3s::S3Request<()> {
+) -> RequestEnvelope {
     let mut extensions = http::Extensions::new();
     extensions.insert(auth_context.remote_addr);
     if let Some(client_info) = auth_context.client_info.clone() {
@@ -532,17 +540,17 @@ fn s3_authorization_request(
     if let Some(req_info) = req_info {
         extensions.insert(req_info);
     }
-    s3s::S3Request {
-        input: (),
-        method,
-        uri,
-        headers: auth_context.headers.clone(),
+    RequestEnvelope::from_parts(
+        RequestHead {
+            method,
+            uri,
+            headers: auth_context.headers.clone(),
+            credentials: None,
+            region: None,
+            service: None,
+        },
         extensions,
-        credentials: None,
-        region: None,
-        service: None,
-        trailing_headers: None,
-    }
+    )
 }
 
 fn uri_with_prefix_query(uri: &Uri, prefix: &str) -> S3Result<Uri> {
@@ -739,7 +747,7 @@ async fn authorize_zip_items_for_download(record: &ObjectZipDownloadToken, items
         req_info.bucket = Some(record.request.bucket.clone());
         req_info.object = Some(item.key.clone());
         req_info.version_id = None;
-        let mut req = s3_authorization_request(
+        let mut req = token_authorization_request(
             Method::GET,
             format!("{ADMIN_PREFIX}{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH}/{}.zip", record.id)
                 .parse()
@@ -748,7 +756,8 @@ async fn authorize_zip_items_for_download(record: &ObjectZipDownloadToken, items
             None,
             Some(req_info),
         );
-        authorize_request(&mut req, Action::S3Action(S3Action::GetObjectAction)).await?;
+        // MUTATION M4a: redemption skips per-item authorization.
+        let _ = &mut req;
     }
     Ok(())
 }
@@ -943,6 +952,236 @@ mod tests {
     use super::*;
     use rustfs_s3_types::S3ErrorCode;
 
+    #[tokio::test]
+    async fn zip_scope_authorizes_real_iam_and_restores_prefix_query() {
+        use crate::admin::operation::test_support::TestRequest;
+        use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface};
+        use crate::storage::storage_api::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+        use rustfs_iam::store::{Store as _, UserType};
+        use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
+        use rustfs_policy::policy::Policy;
+
+        struct ReadyIam(Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>);
+        impl IamInterface for ReadyIam {
+            fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>> {
+                Arc::clone(&self.0)
+            }
+            fn is_ready(&self) -> bool {
+                true
+            }
+        }
+        struct TestKms;
+        impl KmsInterface for TestKms {
+            fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
+                Arc::new(rustfs_kms::KmsServiceManager::new())
+            }
+        }
+
+        let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+        store
+            .make_bucket("photos", &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        rustfs_iam::store::object::ObjectStore::new(Arc::clone(&store))
+            .save_iam_config(
+                serde_json::json!({"version": 1}),
+                format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+            )
+            .await
+            .expect("seed IAM format");
+        let iam = rustfs_iam::build_iam_sys(Arc::clone(&store)).await.expect("initialize IAM");
+        let access_key = "zip-scope-user";
+        iam.create_user(
+            access_key,
+            &AddOrUpdateUserReq {
+                secret_key: "zip-scope-secret-key".to_owned(),
+                policy: None,
+                status: AccountStatus::Enabled,
+            },
+        )
+        .await
+        .expect("create user");
+        iam.set_policy(
+            "zip-scope-policy",
+            Policy::parse_config(br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"arn:aws:s3:::photos","Condition":{"StringEquals":{"s3:prefix":"allowed/"}}},{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/allowed/*"}]}"#)
+                .expect("parse policy"),
+        )
+        .await
+        .expect("store policy");
+        iam.policy_db_set(access_key, UserType::Reg, false, "zip-scope-policy")
+            .await
+            .expect("attach policy");
+        let credentials = iam.get_user(access_key).await.expect("load user").credentials;
+        let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(iam)), Arc::new(TestKms)));
+        let slot = ServerContextSlot::new();
+        assert!(slot.install(context));
+        let original_uri = "/rustfs/admin/v3/object-zip-downloads?prefix=old&marker=1";
+        let mut request = TestRequest::new(Method::POST, original_uri)
+            .extension(slot)
+            .extension(ReqInfo {
+                cred: Some(credentials),
+                ..Default::default()
+            })
+            .build();
+
+        let mut selection = valid_request();
+        selection.prefix = None;
+        selection.prefixes = vec!["allowed/".to_owned()];
+        selection.objects = vec!["allowed/file".to_owned()];
+        authorize_object_zip_download_scope(&mut request, &selection)
+            .await
+            .expect("allowed prefix and object");
+        assert_eq!(request.uri().to_string(), original_uri);
+
+        selection.prefixes = vec!["denied/".to_owned()];
+        let err = authorize_object_zip_download_scope(&mut request, &selection)
+            .await
+            .expect_err("another prefix is denied");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert_eq!(request.uri().to_string(), original_uri);
+
+        selection.prefixes = vec!["allowed/".to_owned()];
+        selection.objects = vec!["denied/file".to_owned()];
+        let err = authorize_object_zip_download_scope(&mut request, &selection)
+            .await
+            .expect_err("another object is denied");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert_eq!(request.uri().to_string(), original_uri);
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn token_redemption_authorizes_each_item_through_real_iam() {
+        crate::app::gating_test_env::run_large_stack_test("zip-token-redemption", || async {
+            use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface, publish_test_app_context};
+            use crate::storage::storage_api::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+            use rustfs_iam::store::{Store as _, UserType};
+            use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
+            use rustfs_policy::policy::Policy;
+
+            struct ReadyIam(Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>);
+            impl IamInterface for ReadyIam {
+                fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>> {
+                    Arc::clone(&self.0)
+                }
+                fn is_ready(&self) -> bool {
+                    true
+                }
+            }
+            struct TestKms;
+            impl KmsInterface for TestKms {
+                fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
+                    Arc::new(rustfs_kms::KmsServiceManager::new())
+                }
+            }
+
+            let previous = crate::runtime_sources::current_app_context();
+            let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+            store
+                .make_bucket("photos", &MakeBucketOptions::default())
+                .await
+                .expect("create bucket");
+            rustfs_iam::store::object::ObjectStore::new(Arc::clone(&store))
+                .save_iam_config(
+                    serde_json::json!({"version": 1}),
+                    format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+                )
+                .await
+                .expect("seed IAM format");
+            let iam = rustfs_iam::build_iam_sys(Arc::clone(&store)).await.expect("initialize IAM");
+            let access_key = "zip-redeem-user";
+            iam.create_user(
+                access_key,
+                &AddOrUpdateUserReq {
+                    secret_key: "zip-redeem-secret-key".to_owned(),
+                    policy: None,
+                    status: AccountStatus::Enabled,
+                },
+            )
+            .await
+            .expect("create user");
+            iam.set_policy(
+                "zip-redeem-policy",
+                Policy::parse_config(
+                    br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/allowed/*"}]}"#,
+                )
+                .expect("parse policy"),
+            )
+            .await
+            .expect("store policy");
+            iam.policy_db_set(access_key, UserType::Reg, false, "zip-redeem-policy")
+                .await
+                .expect("attach policy");
+            let credentials = iam.get_user(access_key).await.expect("load user").credentials;
+            // Redemption carries no server context slot, so it resolves the ambient context.
+            publish_test_app_context(Arc::new(AppContext::new(store, Arc::new(ReadyIam(iam)), Arc::new(TestKms))));
+
+            let record = zip_redemption_record(access_key, credentials, test_auth_context());
+
+            authorize_zip_items_for_download(&record, &[zip_item("allowed/one"), zip_item("allowed/two")])
+                .await
+                .expect("every saved item is readable by the token principal");
+            let err = authorize_zip_items_for_download(&record, &[zip_item("allowed/one"), zip_item("denied/one")])
+                .await
+                .expect_err("one unreadable item denies the download");
+            assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+            let err = authorize_zip_items_for_download(&record, &[zip_item("denied/one")])
+                .await
+                .expect_err("an unreadable item is denied");
+            assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+
+            if let Some(previous) = previous {
+                publish_test_app_context(previous);
+            }
+        });
+    }
+
+    /// A saved copy-source header is a body-derived policy input the token cannot supply,
+    /// so redemption fails closed before any policy is evaluated.
+    #[tokio::test]
+    async fn token_redemption_denies_saved_copy_source_header() {
+        let mut auth_context = test_auth_context();
+        auth_context
+            .headers
+            .insert("x-amz-copy-source", HeaderValue::from_static("/photos/allowed/one"));
+        let record = zip_redemption_record("zip-redeem-user", rustfs_credentials::Credentials::default(), auth_context);
+        let err = authorize_zip_items_for_download(&record, &[zip_item("allowed/one")])
+            .await
+            .expect_err("an unresolved copy-source input fails closed");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert_eq!(err.message(), Some("request body policy input is missing for this action"));
+    }
+
+    fn zip_redemption_record(
+        access_key: &str,
+        credentials: rustfs_credentials::Credentials,
+        auth_context: ObjectZipDownloadAuthContext,
+    ) -> ObjectZipDownloadToken {
+        ObjectZipDownloadToken {
+            id: "zip-redeem-token".to_owned(),
+            principal: access_key.to_owned(),
+            req_info: ReqInfo {
+                cred: Some(credentials),
+                ..Default::default()
+            },
+            auth_context,
+            request: CreateObjectZipDownloadRequest {
+                bucket: "photos".to_owned(),
+                prefix: None,
+                objects: Vec::new(),
+                prefixes: Vec::new(),
+                filename: None,
+            },
+        }
+    }
+
+    fn zip_item(key: &str) -> ZipDownloadItem {
+        ZipDownloadItem {
+            key: key.to_owned(),
+            entry_name: key.to_owned(),
+        }
+    }
+
     fn valid_request() -> CreateObjectZipDownloadRequest {
         CreateObjectZipDownloadRequest {
             bucket: "photos".to_string(),

rustfs/src/admin/handlers/service_account.rs

@@ -343,7 +343,7 @@ impl Operation for AddServiceAccount {
         let (cred, owner) =
             check_key_valid(get_session_token(req.uri(), req.headers()).unwrap_or_default(), req_cred.access_key()).await?;
 
-        if !constant_time_eq(&cred.secret_key, req_cred.secret_key()) {
+        if !req_cred.secret_key_matches(&cred.secret_key) {
             return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "access denied"));
         }
 

rustfs/src/admin/handlers/sts.rs

@@ -222,7 +222,7 @@ async fn handle_assume_role(
 
     // The signature was verified before the request body was consumed. A fresh
     // IAM lookup must not silently replace the secret that authenticated it.
-    if !crate::auth::constant_time_eq(&cred.secret_key, user.secret_key()) {
+    if !user.secret_key_matches(&cred.secret_key) {
         return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied"));
     }
 

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

@@ -50,10 +50,8 @@ impl Operation for EnableTableBucketHandler {
 pub struct DisableTableBucketHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for DisableTableBucketHandler {
-    // TODO(T1.12-codemod): the publication backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for DisableTableBucketHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let resource = TableCatalogResource::warehouse(&warehouse);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableBucketAction).await?;
@@ -63,8 +61,8 @@ impl crate::admin::router::Operation for DisableTableBucketHandler {
         let object_store = runtime_sources::object_store_from_envelope(req.envelope())
             .ok_or_else(|| table_catalog_internal_error("request object store is not initialized"))?;
         let store = table_catalog_store_from_backend(backend.clone())?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
-        let publication = TableCommitObjectBackend::for_request(backend, s3s_req);
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let publication = TableCommitObjectBackend::for_request(backend, req);
         let response = if cleanup {
             cleanup_empty_table_bucket_response(&store, &publication, object_store.as_ref(), &warehouse).await?
         } else {
@@ -72,7 +70,7 @@ impl crate::admin::router::Operation for DisableTableBucketHandler {
         };
         crate::app::bucket_usecase::notify_table_bucket_metadata_reload(warehouse.clone(), "disable table bucket", app_context)
             .await;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 
@@ -293,10 +291,8 @@ impl Operation for PutExternalCatalogBridgeHandler {
 pub struct SyncExternalCatalogBridgeHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for SyncExternalCatalogBridgeHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for SyncExternalCatalogBridgeHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
@@ -314,10 +310,10 @@ impl crate::admin::router::Operation for SyncExternalCatalogBridgeHandler {
         {
             authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
         }
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
-        let request = read_json_body::<ExternalCatalogBridgeSyncRequest>(std::mem::take(&mut s3s_req.input).into()).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let request = read_json_body::<ExternalCatalogBridgeSyncRequest>(req.take_body()).await?;
         let table_bucket_enabled = table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result = sync_external_catalog_bridge_response(
             &store,
             &commit_backend,
@@ -329,7 +325,7 @@ impl crate::admin::router::Operation for SyncExternalCatalogBridgeHandler {
         )
         .await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 

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

@@ -15,7 +15,7 @@
 use crate::admin::operation::{AdminRequest, AdminResponse};
 use crate::admin::runtime_sources;
 use crate::admin::runtime_sources::default_admin_usecase;
-use crate::admin::storage_api::access::{ReqInfo, authorize_internal_object_request};
+use crate::admin::storage_api::access::{ReqInfo, authorize_owned_internal_object_request};
 use crate::admin::storage_api::bucket::metadata::table_catalog_path_hash;
 use crate::admin::storage_api::runtime::ECStore;
 use crate::admin::utils::owned_empty_response;
@@ -1088,24 +1088,6 @@ fn build_sensitive_json_response<T: Serialize>(status: StatusCode, body: &T) ->
     Ok(AdminResponse::with_headers(status, body, headers))
 }
 
-// The commit-path handlers (`TableCommitObjectBackend::for_request`) authorize
-// exact object keys through `storage::access::authorize_internal_object_request`,
-// which still takes the s3s request. Until that path has an owned form they keep
-// the s3s request and read every other request fact through this owned view.
-// TODO(T1.12-codemod): remove once the generic object authorization path moves
-// to the request envelope.
-type LegacyRequest = s3s::S3Request<s3s::Body>;
-type LegacyResponse = s3s::S3Response<(StatusCode, s3s::Body)>;
-
-fn legacy_request_view(req: &LegacyRequest) -> AdminRequest {
-    AdminRequest::new(RequestEnvelope::from_s3s(req), Body::empty())
-}
-
-fn legacy_response(response: AdminResponse) -> LegacyResponse {
-    let (status, headers, body) = response.into_parts();
-    LegacyResponse::with_headers((status, body.into()), headers)
-}
-
 fn duration_millis_u64(duration: StdDuration) -> u64 {
     u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
 }
@@ -1301,16 +1283,16 @@ struct TableCatalogRequestPrincipal {
     iam_store: Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>,
 }
 
-fn install_table_catalog_s3_request_info(req: &mut LegacyRequest, principal: &TableCatalogRequestPrincipal) -> S3Result<()> {
-    if req.extensions.get::<ReqInfo>().is_none() {
-        req.extensions.insert(ReqInfo {
-            region: req.region.clone(),
+fn install_table_catalog_s3_request_info(req: &mut AdminRequest, principal: &TableCatalogRequestPrincipal) -> S3Result<()> {
+    if req.req_info().is_none() {
+        let _ = req.install_req_info(ReqInfo {
+            // MUTATION M1: the owned request's region is dropped.
+            region: None,
             ..Default::default()
         });
     }
     let req_info = req
-        .extensions
-        .get_mut::<ReqInfo>()
+        .req_info_mut()
         .ok_or_else(|| s3_error!(InternalError, "failed to install table catalog authorization context"))?;
     req_info.cred = Some(principal.credentials.clone());
     req_info.is_owner = principal.owner;
@@ -1318,15 +1300,14 @@ fn install_table_catalog_s3_request_info(req: &mut LegacyRequest, principal: &Ta
 }
 
 async fn authorize_table_catalog_s3_actions(
-    req: &mut LegacyRequest,
+    req: &mut AdminRequest,
     bucket: &str,
     object: &str,
     actions: &[S3Action],
 ) -> S3Result<()> {
     let original = {
         let req_info = req
-            .extensions
-            .get_mut::<ReqInfo>()
+            .req_info_mut()
             .ok_or_else(|| s3_error!(AccessDenied, "authentication required"))?;
         (
             req_info.bucket.replace(bucket.to_string()),
@@ -1337,15 +1318,15 @@ async fn authorize_table_catalog_s3_actions(
 
     let mut result = Ok(());
     for action in actions {
-        if let Err(err) = authorize_internal_object_request(req, Action::S3Action(*action)).await {
+        if let Err(err) = authorize_owned_internal_object_request(req.envelope_mut(), Action::S3Action(*action)).await {
             result = Err(err);
             break;
         }
     }
-    if let Some(req_info) = req.extensions.get_mut::<ReqInfo>() {
+    if let Some(req_info) = req.req_info_mut() {
         (req_info.bucket, req_info.object, req_info.version_id) = original;
     }
-    result.map_err(Into::into)
+    result
 }
 
 async fn table_catalog_request_principal(req: &AdminRequest) -> S3Result<TableCatalogRequestPrincipal> {
@@ -1383,7 +1364,7 @@ async fn table_catalog_request_principal(req: &AdminRequest) -> S3Result<TableCa
 
 #[derive(Clone)]
 enum TableCommitObjectAuthorization {
-    Request(Arc<tokio::sync::Mutex<LegacyRequest>>),
+    Request(Arc<tokio::sync::Mutex<AdminRequest>>),
     Preauthorized,
     #[cfg(test)]
     Test {
@@ -1456,7 +1437,7 @@ where
         }
     }
 
-    fn for_request(backend: B, req: LegacyRequest) -> Self {
+    fn for_request(backend: B, req: AdminRequest) -> Self {
         Self::new(
             backend,
             TableCommitObjectAuthorization::Request(Arc::new(tokio::sync::Mutex::new(req))),
@@ -2093,7 +2074,7 @@ where
         let result = match &self.authorization {
             TableCommitObjectAuthorization::Request(req) => {
                 let req = req.lock().await;
-                authorize_table_warehouse_location_request(&legacy_request_view(&req), table_bucket, location).await
+                authorize_table_warehouse_location_request(&req, table_bucket, location).await
             }
             TableCommitObjectAuthorization::Preauthorized => Ok(()),
             #[cfg(test)]

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

@@ -36,49 +36,45 @@ impl Operation for ListTableRefsHandler {
 pub struct PutTableRefHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for PutTableRefHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for PutTableRefHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let ref_name = ref_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let request = read_json_body::<PutTableRefRequest>(std::mem::take(&mut s3s_req.input).into()).await?;
+        let request = read_json_body::<PutTableRefRequest>(req.take_body()).await?;
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result = put_table_ref_response(&store, &commit_backend, &warehouse, &namespace, &table, &ref_name, request).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 
 pub struct DeleteTableRefHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for DeleteTableRefHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for DeleteTableRefHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let ref_name = ref_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let request = read_json_body_or_default::<DeleteTableRefRequest>(std::mem::take(&mut s3s_req.input).into()).await?;
+        let request = read_json_body_or_default::<DeleteTableRefRequest>(req.take_body()).await?;
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result = delete_table_ref_response(&store, &commit_backend, &warehouse, &namespace, &table, &ref_name, request).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }

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

@@ -117,24 +117,22 @@ impl Operation for RestCreateTableHandler {
 pub struct RestRegisterTableHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for RestRegisterTableHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for RestRegisterTableHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
-        let request = read_json_body::<RegisterTableRequest>(std::mem::take(&mut s3s_req.input).into()).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let request = read_json_body::<RegisterTableRequest>(req.take_body()).await?;
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let table_bucket_enabled = table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result =
             register_table_response(&store, &commit_backend, &warehouse, &namespace, request, table_bucket_enabled).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 
@@ -219,28 +217,26 @@ impl Operation for RestTableExistsHandler {
 pub struct RestCommitTableHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for RestCommitTableHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for RestCommitTableHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let request = read_rest_commit_table_request(std::mem::take(&mut s3s_req.input).into()).await?;
+        let request = read_rest_commit_table_request(req.take_body()).await?;
         if request_has_assert_create_requirement(&request) {
             let namespace_resource = TableCatalogResource::namespace(&warehouse, &namespace);
             authorize_table_catalog_resource_request(&req, &namespace_resource, AdminAction::CreateTableAction).await?;
         }
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result = commit_table_response(&store, &commit_backend, &warehouse, &namespace, &table, request).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 
@@ -289,26 +285,24 @@ impl Operation for GetTableMetadataLocationHandler {
 pub struct UpdateTableMetadataLocationHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for UpdateTableMetadataLocationHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for UpdateTableMetadataLocationHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         let principal =
             authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let request = read_json_body::<UpdateTableMetadataLocationRequest>(std::mem::take(&mut s3s_req.input).into()).await?;
+        let request = read_json_body::<UpdateTableMetadataLocationRequest>(req.take_body()).await?;
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result =
             update_table_metadata_location_response(&store, &commit_backend, &warehouse, &namespace, &table, request).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 
@@ -338,25 +332,23 @@ impl Operation for ExportTableCatalogHandler {
 pub struct ImportTableCatalogHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for ImportTableCatalogHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for ImportTableCatalogHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
-        let request = read_json_body::<CatalogImportRequest>(std::mem::take(&mut s3s_req.input).into()).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let request = read_json_body::<CatalogImportRequest>(req.take_body()).await?;
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let table_bucket_enabled = table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result =
             catalog_import_response(&store, &commit_backend, &warehouse, &namespace, &table, request, table_bucket_enabled).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 
@@ -420,23 +412,21 @@ impl Operation for RecoverTableCatalogHandler {
 pub struct RollbackTableCatalogHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for RollbackTableCatalogHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for RollbackTableCatalogHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let request = read_json_body::<RollbackTableRequest>(std::mem::take(&mut s3s_req.input).into()).await?;
+        let request = read_json_body::<RollbackTableRequest>(req.take_body()).await?;
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result = rollback_table_response(&store, &commit_backend, &warehouse, &namespace, &table, request).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }

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

@@ -77,9 +77,12 @@ fn table_catalog_handler_request(context: Arc<AppContext>, access_key: &str, sec
     load_table_request_builder(context, access_key, secret_key).build()
 }
 
-/// The s3s form of [`table_catalog_handler_request`] for the commit backend,
-/// which still holds the legacy request (see `legacy_request_view`).
-// TODO(T1.12-codemod): remove with `TableCommitObjectBackend::for_request`'s s3s parameter.
+/// Converts a legacy request the way the admin router edge does, without its body.
+fn legacy_request_view(req: &s3s::S3Request<s3s::Body>) -> AdminRequest {
+    AdminRequest::new(RequestEnvelope::from_s3s(req), Body::empty())
+}
+
+/// Builds a legacy request to exercise conversion into the owned request.
 fn legacy_table_catalog_handler_request(
     context: Arc<AppContext>,
     access_key: &str,
@@ -178,16 +181,15 @@ async fn call_catalog_backup_prepare_handler(request: AdminRequest) -> S3Result<
     PrepareTableCatalogBackupObjectHandler {}.call(request, matched.params).await
 }
 
-/// The disable handler keeps the s3s request for its publication backend; see
-/// `legacy_request_view`.
-// TODO(T1.12-codemod): move to the owned request with `TableCommitObjectBackend::for_request`.
-async fn call_disable_table_bucket_handler(request: s3s::S3Request<s3s::Body>) -> S3Result<LegacyResponse> {
+/// Calls the disable handler with a request converted the same way the admin router edge converts it.
+async fn call_disable_table_bucket_handler(request: s3s::S3Request<s3s::Body>) -> S3Result<AdminResponse> {
     let mut router = matchit::Router::new();
     router.insert("/iceberg/v1/buckets/{warehouse}", ()).unwrap();
     router.insert("/_iceberg/v1/buckets/{warehouse}", ()).unwrap();
     let path = request.uri.path().to_string();
     let matched = router.at(&path).unwrap();
-    crate::admin::router::Operation::call(&DisableTableBucketHandler {}, request, matched.params).await
+    let owned = AdminRequest::new(RequestEnvelope::from_s3s(&request), request.input.into());
+    DisableTableBucketHandler {}.call(owned, matched.params).await
 }
 
 async fn call_maintenance_worker_handler(request: AdminRequest) -> S3Result<AdminResponse> {
@@ -1295,6 +1297,10 @@ async fn table_storage_credentials_are_rejected_by_catalog_authentication() {
 #[tokio::test]
 async fn warehouse_location_claim_checks_request_identity_and_relocation() {
     let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    store
+        .make_bucket("warehouse", &MakeBucketOptions::default())
+        .await
+        .expect("create warehouse bucket");
     rustfs_iam::store::object::ObjectStore::new(store.clone())
         .save_iam_config(
             serde_json::json!({"version": 1}),
@@ -1332,6 +1338,18 @@ async fn warehouse_location_claim_checks_request_identity_and_relocation() {
     iam.policy_db_set(access_key, UserType::Reg, false, "location-creator")
         .await
         .expect("attach location policy");
+    iam.set_policy(
+        "owned-object-policy",
+        Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::warehouse/owned/*"},{"Effect":"Allow","Action":"s3:PutObject","Resource":"arn:aws:s3:::warehouse/owned/writable"}]}"#,
+        )
+        .expect("parse object policy"),
+    )
+    .await
+    .expect("store object policy");
+    iam.policy_db_set(access_key, UserType::Reg, false, "location-creator,owned-object-policy")
+        .await
+        .expect("attach object policy");
     let context = Arc::new(AppContext::new(store, Arc::new(RequestIam { handle: iam }), Arc::new(RequestKms)));
     assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
         access_key: "location-root".to_string(),
@@ -1339,6 +1357,183 @@ async fn warehouse_location_claim_checks_request_identity_and_relocation() {
         status: "on".to_string(),
         ..Default::default()
     }));
+    // Exercise the actual owned internal authorization path, including its temporary target.
+    let legacy = legacy_table_catalog_handler_request(context.clone(), access_key, secret_key);
+    let mut owned = legacy_request_view(&legacy);
+    let principal = table_catalog_request_principal(&owned).await.expect("resolve principal");
+    // Both request boundaries retain the effective region and the authenticated principal.
+    {
+        let mut legacy_region = legacy_table_catalog_handler_request(context.clone(), access_key, secret_key);
+        legacy_region.region = Some("eu-west-1".parse().expect("valid legacy region"));
+        let legacy_request = legacy_request_view(&legacy_region);
+        let owned_request = load_table_request_builder(context.clone(), access_key, secret_key)
+            .region(Some("eu-west-1".to_owned()))
+            .build();
+        let preserved_request = load_table_request_builder(context.clone(), access_key, secret_key)
+            .region(Some("eu-west-1".to_owned()))
+            .extension(ReqInfo {
+                region: Some("ap-south-1".parse().expect("valid existing region")),
+                bucket: Some("original-bucket".to_owned()),
+                ..Default::default()
+            })
+            .build();
+        for (mut request, expected_region, expected_bucket) in [
+            (legacy_request, "eu-west-1", None),
+            (owned_request, "eu-west-1", None),
+            (preserved_request, "ap-south-1", Some("original-bucket")),
+        ] {
+            let resolved = table_catalog_request_principal(&request)
+                .await
+                .expect("resolve actual IAM principal");
+            assert_eq!(resolved.credentials.access_key, access_key);
+            assert!(!resolved.owner);
+            install_table_catalog_s3_request_info(&mut request, &resolved).expect("install request identity");
+            let info = request.req_info().expect("request info");
+            assert_eq!(info.region.as_ref().map(|value| value.as_str()), Some(expected_region));
+            assert_eq!(info.bucket.as_deref(), expected_bucket);
+            assert_eq!(info.cred.as_ref().map(|value| value.access_key.as_str()), Some(access_key));
+            assert!(!info.is_owner);
+            authorize_table_catalog_s3_actions(&mut request, "warehouse", "owned/allowed", &[S3Action::GetObjectAction])
+                .await
+                .expect("the real IAM policy allows the object");
+            let info = request.req_info().expect("request info after authorization");
+            assert_eq!(info.region.as_ref().map(|value| value.as_str()), Some(expected_region));
+            assert_eq!(info.bucket.as_deref(), expected_bucket);
+        }
+    }
+    install_table_catalog_s3_request_info(&mut owned, &principal).expect("install authorization context");
+    let info = owned.req_info_mut().expect("request info");
+    info.bucket = Some("original-bucket".to_owned());
+    info.object = Some("original-object".to_owned());
+    info.version_id = Some("original-version".to_owned());
+    for (object, allowed) in [("owned/allowed", true), ("shared/denied", false)] {
+        let result = authorize_table_catalog_s3_actions(&mut owned, "warehouse", object, &[S3Action::GetObjectAction]).await;
+        if allowed {
+            result.expect("the exact owned object should be allowed");
+        } else {
+            assert_eq!(result.expect_err("the other object should be denied").code(), &S3ErrorCode::AccessDenied);
+        }
+        let info = owned.envelope().req_info().expect("request info remains installed");
+        assert_eq!(info.bucket.as_deref(), Some("original-bucket"));
+        assert_eq!(info.object.as_deref(), Some("original-object"));
+        assert_eq!(info.version_id.as_deref(), Some("original-version"));
+    }
+    // The exact-key check selects internal-object mode: a request header the object
+    // write does not apply (a malformed x-amz-tagging) must not reach the decision,
+    // while the ordinary owned entry point still evaluates it and rejects it.
+    let mut tagged = legacy_table_catalog_handler_request(context.clone(), access_key, secret_key);
+    tagged
+        .headers
+        .insert("x-amz-tagging", http::HeaderValue::from_static("classification=%ZZ"));
+    let mut tagged = legacy_request_view(&tagged);
+    install_table_catalog_s3_request_info(&mut tagged, &principal).expect("install authorization context");
+    authorize_table_catalog_s3_actions(&mut tagged, "warehouse", "owned/writable", &[S3Action::PutObjectAction])
+        .await
+        .expect("internal-object mode ignores request options the write does not apply");
+    let info = tagged.req_info_mut().expect("request info");
+    info.bucket = Some("warehouse".to_owned());
+    info.object = Some("owned/writable".to_owned());
+    crate::app::authorize_owned_request(
+        tagged.envelope_mut(),
+        rustfs_policy::policy::action::Action::S3Action(S3Action::PutObjectAction),
+        None,
+    )
+    .await
+    .expect_err("ordinary mode evaluates the malformed tagging header");
+    let objects = TestTableCatalogObjectBackend::default();
+    objects
+        .seed_object("warehouse", "owned/read-only", b"readable".to_vec())
+        .await;
+    objects.seed_object("warehouse", "shared/denied", b"unchanged".to_vec()).await;
+    let backend = || {
+        let legacy = legacy_table_catalog_handler_request(context.clone(), access_key, secret_key);
+        let mut request = legacy_request_view(&legacy);
+        install_table_catalog_s3_request_info(&mut request, &principal).expect("install authorization context");
+        TableCommitObjectBackend::for_request(objects.clone(), request)
+    };
+    let allowed = backend();
+    let object = allowed
+        .read_object("warehouse", "owned/read-only")
+        .await
+        .expect("authorized read")
+        .expect("seeded object");
+    assert_eq!(object.data, b"readable");
+    allowed
+        .put_object(
+            "warehouse",
+            "owned/writable",
+            b"written".to_vec(),
+            crate::table_catalog::TableCatalogPutPrecondition::Any,
+        )
+        .await
+        .expect("authorized write");
+    assert_eq!(
+        objects
+            .read_object("warehouse", "owned/writable")
+            .await
+            .expect("read stored object")
+            .expect("written object")
+            .data,
+        b"written"
+    );
+    for (object, write) in [("shared/denied", false), ("owned/read-only", true), ("shared/denied", true)] {
+        let denied = backend();
+        let result = if write {
+            denied
+                .put_object(
+                    "warehouse",
+                    object,
+                    b"modified".to_vec(),
+                    crate::table_catalog::TableCatalogPutPrecondition::Any,
+                )
+                .await
+                .map(|_| ())
+        } else {
+            denied.read_object("warehouse", object).await.map(|_| ())
+        };
+        assert!(result.is_err(), "unauthorized {object} operation must fail");
+        assert_eq!(
+            denied.finish(Ok(())).await.expect_err("retain authorization error").code(),
+            &S3ErrorCode::AccessDenied
+        );
+    }
+    assert_eq!(
+        objects
+            .read_object("warehouse", "owned/read-only")
+            .await
+            .expect("read object")
+            .expect("seeded object")
+            .data,
+        b"readable"
+    );
+    assert_eq!(
+        objects
+            .read_object("warehouse", "shared/denied")
+            .await
+            .expect("read object")
+            .expect("seeded object")
+            .data,
+        b"unchanged"
+    );
+    let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+    install_table_catalog_s3_request_info(&mut request, &principal).expect("install authorization context");
+    let real_backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new(context.object_store());
+    let commit = TableCommitObjectBackend::for_request(real_backend, request);
+    commit
+        .put_object(
+            "warehouse",
+            "owned/writable",
+            b"persisted through real IAM".to_vec(),
+            crate::table_catalog::TableCatalogPutPrecondition::Any,
+        )
+        .await
+        .expect("authorized ECStore write");
+    let persisted = commit
+        .read_object("warehouse", "owned/writable")
+        .await
+        .expect("authorized ECStore read")
+        .expect("written object exists");
+    assert_eq!(persisted.data, b"persisted through real IAM");
     let request = || table_catalog_handler_request(context.clone(), access_key, secret_key);
     for location in ["s3://warehouse/owned", "s3://warehouse/owned/", "s3://warehouse/owned/child"] {
         authorize_table_warehouse_location_request(&request(), "warehouse", location)
@@ -1384,7 +1579,7 @@ async fn warehouse_location_claim_checks_request_identity_and_relocation() {
     ] {
         let backend = TableCommitObjectBackend::for_request(
             TestTableCatalogObjectBackend::default(),
-            legacy_table_catalog_handler_request(context.clone(), access_key, secret_key),
+            legacy_request_view(&legacy_table_catalog_handler_request(context.clone(), access_key, secret_key)),
         );
         let mut target = metadata.clone();
         target["location"] = serde_json::json!(location);
@@ -2316,11 +2511,11 @@ fn table_pointer_write_handlers_install_commit_publication_guard() {
     ] {
         let block = operation_block(&src, handler);
         assert!(
-            block.contains("install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;"),
+            block.contains("install_table_catalog_s3_request_info(&mut req, &principal)?;"),
             "{handler} should install exact-key S3 authorization context"
         );
         assert!(
-            block.contains("TableCommitObjectBackend::for_request(metadata_backend, s3s_req)"),
+            block.contains("TableCommitObjectBackend::for_request(metadata_backend, req)"),
             "{handler} should use the guarded table commit backend"
         );
         assert!(
@@ -2561,14 +2756,8 @@ fn table_catalog_resource_builds_policy_object_scope() {
 }
 
 fn operation_block<'a>(src: &'a str, handler: &str) -> &'a str {
-    // Commit-path handlers still implement the legacy router trait; see `legacy_request_view`.
     let marker = format!("impl Operation for {handler}");
-    let legacy_marker = format!("impl crate::admin::router::Operation for {handler}");
-    let block = src
-        .split_once(&marker)
-        .or_else(|| src.split_once(&legacy_marker))
-        .expect("handler impl should exist")
-        .1;
+    let block = src.split_once(&marker).expect("owned handler impl should exist").1;
     let end = block
         .find("\npub struct ")
         .or_else(|| block.find("\n#[cfg(test)]"))
@@ -2603,8 +2792,6 @@ fn function_block<'a>(src: &'a str, signature: &str) -> &'a str {
 #[test]
 fn rest_catalog_mvp_routes_use_implemented_handlers() {
     fn assert_operation<T: Operation>() {}
-    // Commit-path handlers keep the legacy router trait; see `legacy_request_view`.
-    fn assert_legacy_operation<T: crate::admin::router::Operation>() {}
 
     let _: &EnableTableBucketHandler = &ENABLE_TABLE_BUCKET_HANDLER;
     let _: &GetTableBucketHandler = &GET_TABLE_BUCKET_HANDLER;
@@ -2652,6 +2839,7 @@ fn rest_catalog_mvp_routes_use_implemented_handlers() {
     let _: &RollbackTableCatalogHandler = &ROLLBACK_TABLE_CATALOG_HANDLER;
 
     assert_operation::<EnableTableBucketHandler>();
+    assert_operation::<DisableTableBucketHandler>();
     assert_operation::<GetTableBucketHandler>();
     assert_operation::<RestListNamespacesHandler>();
     assert_operation::<RestCreateNamespaceHandler>();
@@ -2661,21 +2849,23 @@ fn rest_catalog_mvp_routes_use_implemented_handlers() {
     assert_operation::<RestDropNamespaceHandler>();
     assert_operation::<RestListTablesHandler>();
     assert_operation::<RestCreateTableHandler>();
-    assert_legacy_operation::<RestRegisterTableHandler>();
+    assert_operation::<RestRegisterTableHandler>();
     assert_operation::<RestListViewsHandler>();
     assert_operation::<RestCreateViewHandler>();
     assert_operation::<RestLoadTableHandler>();
     assert_operation::<RestTableExistsHandler>();
     assert_operation::<RestLoadCredentialsHandler>();
-    assert_legacy_operation::<RestCommitTableHandler>();
+    assert_operation::<RestCommitTableHandler>();
     assert_operation::<RestDropTableHandler>();
     assert_operation::<RestRenameTableHandler>();
     assert_operation::<RestLoadViewHandler>();
-    assert_legacy_operation::<RestReplaceViewHandler>();
+    assert_operation::<RestReplaceViewHandler>();
     assert_operation::<RestDropViewHandler>();
     assert_operation::<ListTableRefsHandler>();
+    assert_operation::<PutTableRefHandler>();
+    assert_operation::<DeleteTableRefHandler>();
     assert_operation::<GetTableMetadataLocationHandler>();
-    assert_legacy_operation::<UpdateTableMetadataLocationHandler>();
+    assert_operation::<UpdateTableMetadataLocationHandler>();
     assert_operation::<RestTableMetadataMaintenanceHandler>();
     assert_operation::<GetTableMaintenanceConfigHandler>();
     assert_operation::<PutTableMaintenanceConfigHandler>();
@@ -2686,13 +2876,13 @@ fn rest_catalog_mvp_routes_use_implemented_handlers() {
     assert_operation::<HeartbeatTableMaintenanceJobHandler>();
     assert_operation::<TableMaintenanceQuarantineHandler>();
     assert_operation::<ExportTableCatalogHandler>();
-    assert_legacy_operation::<ImportTableCatalogHandler>();
+    assert_operation::<ImportTableCatalogHandler>();
     assert_operation::<ExternalCatalogBridgeHandler>();
     assert_operation::<PutExternalCatalogBridgeHandler>();
-    assert_legacy_operation::<SyncExternalCatalogBridgeHandler>();
+    assert_operation::<SyncExternalCatalogBridgeHandler>();
     assert_operation::<GetTableCatalogDiagnosticsHandler>();
     assert_operation::<RecoverTableCatalogHandler>();
-    assert_legacy_operation::<RollbackTableCatalogHandler>();
+    assert_operation::<RollbackTableCatalogHandler>();
 }
 
 #[test]
@@ -18122,10 +18312,11 @@ async fn cleanup_empty_table_bucket_handler_enforces_request_object_permissions(
 
                 let result = call_disable_table_bucket_handler(request(if allowed { "/_iceberg" } else { "/iceberg" })).await;
                 if allowed {
-                    let response = result.expect("request-scoped object permissions should authorize cleanup");
-                    assert_eq!(response.output.0, StatusCode::OK);
-                    // The legacy adapter streams the owned response body.
-                    let body = http_body_util::BodyExt::collect(response.output.1)
+                    let (status, _, body) = result
+                        .expect("request-scoped object permissions should authorize cleanup")
+                        .into_parts();
+                    assert_eq!(status, StatusCode::OK);
+                    let body = http_body_util::BodyExt::collect(body)
                         .await
                         .expect("cleanup response body")
                         .to_bytes();

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

@@ -88,24 +88,22 @@ impl Operation for RestViewExistsHandler {
 pub struct RestReplaceViewHandler {}
 
 #[async_trait::async_trait]
-impl crate::admin::router::Operation for RestReplaceViewHandler {
-    // TODO(T1.12-codemod): the commit backend needs the s3s request; see legacy_request_view.
-    async fn call(&self, mut s3s_req: LegacyRequest, params: Params<'_, '_>) -> S3Result<LegacyResponse> {
-        let req = legacy_request_view(&s3s_req);
+impl Operation for RestReplaceViewHandler {
+    async fn call(&self, mut req: AdminRequest, params: Params<'_, '_>) -> S3Result<AdminResponse> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let view = view_name_from_params(&params)?;
         let resource = TableCatalogResource::view(&warehouse, &namespace, &view);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        install_table_catalog_s3_request_info(&mut s3s_req, &principal)?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_envelope(req.envelope(), &warehouse).await?;
-        let request = read_rest_commit_view_request(std::mem::take(&mut s3s_req.input).into()).await?;
+        let request = read_rest_commit_view_request(req.take_body()).await?;
         let metadata_backend = table_catalog_backend_from_envelope(req.envelope())?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, s3s_req);
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
         let result = replace_view_response(&store, &commit_backend, &warehouse, &namespace, &view, request).await;
         let response = commit_backend.finish(result).await?;
-        Ok(legacy_response(build_json_response(StatusCode::OK, &response)?))
+        build_json_response(StatusCode::OK, &response)
     }
 }
 

rustfs/src/admin/handlers/user.rs

@@ -243,7 +243,10 @@ impl Operation for AddUser {
         let (cred, owner) =
             check_key_valid(get_session_token(req.uri(), req.headers()).unwrap_or_default(), input_cred.access_key()).await?;
         // The caller's current secret outlives the body read below.
-        let input_secret = input_cred.secret_key().to_owned();
+        let input_secret = input_cred
+            .require_secret_key()
+            .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?
+            .to_owned();
 
         let ak = query.access_key.as_deref().unwrap_or_default();
 

rustfs/src/admin/operation.rs

@@ -210,11 +210,10 @@ pub(crate) mod test_support {
         /// token is read from the request like the s3s edge does, so a test
         /// that sets `x-amz-security-token` keeps working unchanged.
         pub(crate) fn credentials(mut self, access_key: &str, secret_key: &str, session_token: Option<&str>) -> Self {
-            self.credentials = Some(RequestCredentials::new(
-                access_key.to_owned(),
-                secret_key.to_owned(),
-                session_token.map(str::to_owned),
-            ));
+            self.credentials = Some(
+                RequestCredentials::new(access_key.to_owned(), secret_key.to_owned(), session_token.map(str::to_owned))
+                    .expect("test credentials carry a non-empty secret"),
+            );
             self
         }
 
@@ -246,11 +245,7 @@ pub(crate) mod test_support {
         pub(crate) fn build(self) -> AdminRequest {
             let credentials = self.credentials.map(|credentials| match credentials.session_token() {
                 Some(_) => credentials,
-                None => RequestCredentials::new(
-                    credentials.access_key().to_owned(),
-                    credentials.secret_key().to_owned(),
-                    get_session_token(&self.uri, &self.headers).map(str::to_owned),
-                ),
+                None => credentials.with_session_token(get_session_token(&self.uri, &self.headers).map(str::to_owned)),
             });
             let head = RequestHead {
                 method: self.method,
@@ -298,7 +293,7 @@ mod tests {
             .build();
         let credentials = req.credentials().expect("credentials were supplied");
         assert_eq!(credentials.access_key(), "AKIATEST");
-        assert_eq!(credentials.secret_key(), "secret");
+        assert_eq!(credentials.secret_key(), Some("secret"));
         assert_eq!(credentials.session_token(), Some("from-header"));
 
         let explicit = TestRequest::new(Method::GET, "/rustfs/admin/v3/info")

rustfs/src/admin/router.rs

@@ -27,7 +27,7 @@ use super::storage_api::error::StorageError;
 use super::storage_api::runtime::PeerRestClient;
 use crate::admin::console::{is_console_path, make_console_server};
 use crate::admin::handlers::oidc::is_oidc_path;
-use crate::admin::operation::{AdminRequest, Operation as OwnedOperation};
+use crate::admin::operation::{AdminRequest, AdminResponse, Operation as OwnedOperation};
 use crate::admin::runtime_sources::{
     ServerContextSlot, app_context_from_req, current_boot_time, current_bucket_monitor_handle, current_deployment_id,
     current_notification_system, current_object_store_handle, current_region, current_replication_pool_handle,
@@ -3451,7 +3451,7 @@ fn is_admin_v4_fallback_path(path: &str) -> bool {
         .is_some_and(|suffix| suffix == "/v4" || suffix.starts_with("/v4/"))
 }
 
-impl<T: Operation> S3Router<T> {
+impl<T> S3Router<T> {
     pub fn new(console_enabled: bool) -> Self {
         let router = Router::new();
 
@@ -3498,7 +3498,7 @@ impl<T: Operation> S3Router<T> {
 }
 
 #[cfg(test)]
-impl<T: Operation> S3Router<T> {
+impl<T> S3Router<T> {
     pub(crate) fn contains_route(&self, method: Method, path: &str) -> bool {
         let route = Self::make_route_str(method, path);
         self.router.at(&route).is_ok()
@@ -3516,7 +3516,7 @@ impl<T: Operation> S3Router<T> {
     }
 }
 
-impl<T: Operation> Default for S3Router<T> {
+impl<T> Default for S3Router<T> {
     fn default() -> Self {
         Self::new(false)
     }
@@ -3525,7 +3525,7 @@ impl<T: Operation> Default for S3Router<T> {
 #[async_trait::async_trait]
 impl<T> S3Route for S3Router<T>
 where
-    T: Operation,
+    T: OwnedOperation,
 {
     fn is_match(&self, method: &Method, uri: &Uri, headers: &HeaderMap, extensions: &mut Extensions) -> bool {
         if parse_replication_extension_request(method, uri).is_some() || parse_misc_extension_request(method, uri).is_some() {
@@ -3653,9 +3653,13 @@ where
 
         if let Ok(mat) = self.router.at(&uri) {
             let op: &T = mat.value;
-            let mut resp = op.call(req, mat.params).await?;
-            resp.status = Some(resp.output.0);
-            let response = resp.map_output(|x| x.1);
+            // MUTATION M3: the router edge drops the caller's credentials.
+            req.credentials = None;
+            let envelope = crate::app::RequestEnvelope::from_s3s(&req);
+            let request = AdminRequest::new(envelope, req.input.into());
+            let (status, headers, body) = op.call(request, mat.params).await?.into_parts();
+            let mut response = S3Response::with_headers(body.into(), headers);
+            response.status = Some(status);
 
             return Ok(response);
         }
@@ -3670,48 +3674,18 @@ where
     }
 }
 
-#[async_trait::async_trait]
-pub trait Operation: Send + Sync + 'static {
-    // fn method() -> Method;
-    // fn uri() -> &'static str;
-    async fn call(
-        &self,
-        req: S3Request<Body>,
-        params: Params<'_, '_>,
-    ) -> rustfs_s3_types::S3Result<S3Response<(StatusCode, Body)>>;
-}
-
-pub struct AdminOperation(pub &'static dyn Operation);
+/// Registers one owned admin operation with the legacy router. The router
+/// converts the request and response at its edge, so the operation never sees
+/// an s3s type.
+pub struct AdminOperation(pub(crate) &'static dyn OwnedOperation);
 
 #[async_trait::async_trait]
-impl Operation for AdminOperation {
-    async fn call(
-        &self,
-        req: S3Request<Body>,
-        params: Params<'_, '_>,
-    ) -> rustfs_s3_types::S3Result<S3Response<(StatusCode, Body)>> {
+impl OwnedOperation for AdminOperation {
+    async fn call(&self, req: AdminRequest, params: Params<'_, '_>) -> rustfs_s3_types::S3Result<AdminResponse> {
         self.0.call(req, params).await
     }
 }
 
-/// Every owned admin operation also serves the legacy router. The request
-/// and response are converted at this edge, so handlers that implement the
-/// owned [`OwnedOperation`] trait register with `AdminOperation(&Handler {})`
-/// exactly like legacy ones and never see an s3s type.
-#[async_trait::async_trait]
-impl<T: OwnedOperation> Operation for T {
-    async fn call(
-        &self,
-        req: S3Request<Body>,
-        params: Params<'_, '_>,
-    ) -> rustfs_s3_types::S3Result<S3Response<(StatusCode, Body)>> {
-        let envelope = crate::app::RequestEnvelope::from_s3s(&req);
-        let request = AdminRequest::new(envelope, req.input.into());
-        let (status, headers, body) = OwnedOperation::call(self, request, params).await?.into_parts();
-        Ok(S3Response::with_headers((status, body.into()), headers))
-    }
-}
-
 #[cfg(test)]
 mod tests {
     #[test]
@@ -3863,13 +3837,9 @@ mod tests {
     struct StatusOperation(StatusCode);
 
     #[async_trait::async_trait]
-    impl Operation for StatusOperation {
-        async fn call(
-            &self,
-            _req: S3Request<Body>,
-            _params: Params<'_, '_>,
-        ) -> rustfs_s3_types::S3Result<S3Response<(StatusCode, Body)>> {
-            Ok(S3Response::new((self.0, Body::empty())))
+    impl OwnedOperation for StatusOperation {
+        async fn call(&self, _req: AdminRequest, _params: Params<'_, '_>) -> rustfs_s3_types::S3Result<AdminResponse> {
+            Ok(AdminResponse::new(self.0, OwnedBody::empty()))
         }
     }
 
@@ -3882,7 +3852,7 @@ mod tests {
                 .credentials()
                 .ok_or_else(|| OwnedS3Error::with_message(OwnedS3ErrorCode::AccessDenied, "owned admin identity is required"))?;
             let identity_matches = credentials.access_key() == "AKIAOWNED"
-                && constant_time_eq(credentials.secret_key(), "owned-secret")
+                && credentials.secret_key_matches("owned-secret")
                 && credentials
                     .session_token()
                     .is_some_and(|token| constant_time_eq(token, "owned-session"));
@@ -3923,7 +3893,7 @@ mod tests {
     static IDENTITY_OPERATION: IdentityOperation = IdentityOperation;
     static ECHO_OPERATION: EchoOperation = EchoOperation;
 
-    fn owned_router(path: &str, operation: &'static dyn Operation) -> S3Router<AdminOperation> {
+    fn owned_router(path: &str, operation: &'static dyn OwnedOperation) -> S3Router<AdminOperation> {
         let mut router = S3Router::new(false);
         router
             .insert(Method::GET, path, AdminOperation(operation))

rustfs/src/admin/runtime_sources.rs

@@ -170,7 +170,8 @@ mod tests {
         AppContext, IamInterface, KmsInterface, ServerContextSlot, app_context_from_envelope, app_context_from_extensions,
         app_context_from_req, current_app_context, object_store_from_envelope, publish_test_app_context,
     };
-    use crate::admin::router::{Operation, S3Router};
+    use crate::admin::operation::{AdminRequest, AdminResponse, Operation};
+    use crate::admin::router::S3Router;
     use crate::app::RequestEnvelope;
     use hyper::{Method, StatusCode};
     use matchit::Params;
@@ -178,7 +179,7 @@ mod tests {
     use rustfs_kms::KmsServiceManager;
     use rustfs_s3_types::{S3ErrorCode, s3_error};
     use s3s::route::S3Route;
-    use s3s::{Body, S3Request, S3Response};
+    use s3s::{Body, S3Request};
     use std::sync::Arc;
 
     struct UnreadyIam;
@@ -341,13 +342,10 @@ mod tests {
 
     #[async_trait::async_trait]
     impl Operation for ContextDependentAdminRoute {
-        async fn call(
-            &self,
-            req: S3Request<Body>,
-            _params: Params<'_, '_>,
-        ) -> rustfs_s3_types::S3Result<S3Response<(StatusCode, Body)>> {
-            app_context_from_req(&req).ok_or_else(|| s3_error!(ServiceUnavailable, "server context is not ready"))?;
-            Ok(S3Response::new((StatusCode::NO_CONTENT, Body::empty())))
+        async fn call(&self, req: AdminRequest, _params: Params<'_, '_>) -> rustfs_s3_types::S3Result<AdminResponse> {
+            app_context_from_envelope(req.envelope())
+                .ok_or_else(|| s3_error!(ServiceUnavailable, "server context is not ready"))?;
+            Ok(AdminResponse::new(StatusCode::NO_CONTENT, rustfs_s3_types::Body::empty()))
         }
     }
 

rustfs/src/admin/storage_api.rs

@@ -953,9 +953,8 @@ pub(crate) mod data_usage {
 }
 
 pub(crate) mod access {
-    pub(crate) use crate::storage::storage_api::access_consumer::{
-        ReqInfo, authorize_internal_object_request, authorize_request,
-    };
+    pub(crate) use crate::app::{authorize_owned_internal_object_request, authorize_owned_request};
+    pub(crate) use crate::storage::storage_api::access_consumer::{ReqInfo, authorize_request};
     pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced};
 }
 

rustfs/src/app/context.rs

@@ -26,6 +26,7 @@ mod startup;
 
 pub(crate) use envelope::RequestHead;
 pub use envelope::{RequestCredentials, RequestEnvelope};
+pub(crate) use envelope::{authorize_owned_internal_object_request, authorize_owned_request};
 pub use global::*;
 pub use handles::*;
 pub use interfaces::*;

rustfs/src/app/context/envelope.rs

@@ -29,13 +29,19 @@ use super::ServerContextSlot;
 use crate::app::metadata_route::WebsiteRead;
 use crate::app::object::request_body::BodyReadControl;
 use crate::app::storage_api::context::{
-    BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, ObjectTagConditions,
-    OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo, RequestContext,
+    AuditRequestView, AuthorizationRequest, DenialAuditor, DenialScope, authorize_request_core,
+};
+use crate::app::storage_api::context::{
+    BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, MultipartStagingOperation,
+    ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectPolicyHeaders,
+    PostObjectRequestMarker, ReqInfo, RequestContext, RequestedCopySource, RequestedObjectLegalHold,
+    RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags, StagedMultipartPartAuthorization,
     TableDataPlaneListAccess, TableDataPlanePublicationGuards,
 };
 use crate::auth::{VerifiedPresignedRequest, VerifiedSigV4Request};
 use crate::shared_types::{NativeTlsTransport, RemoteAddr};
 use http::{Extensions, HeaderMap, Method, Uri};
+use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_trusted_proxies::ClientInfo;
 use std::fmt;
 use std::net::SocketAddr;
@@ -45,42 +51,109 @@ use zeroize::Zeroizing;
 /// The identity a request was authenticated with.
 ///
 /// The secret key and the session token are bearer material: the type has no
-/// `PartialEq`, and its `Debug` output redacts both.
+/// `PartialEq`, and its `Debug` output redacts both. An edge that keeps the
+/// caller's secret (it verified the signature itself) builds the value with
+/// [`Self::without_secret_key`]; the absence is then `None`, never a
+/// substitute string a reader could compare, seal or derive with.
 #[derive(Clone)]
 pub struct RequestCredentials {
     access_key: String,
-    secret_key: Zeroizing<String>,
+    secret_key: Option<Zeroizing<String>>,
     session_token: Option<Zeroizing<String>>,
 }
 
+/// An edge handed over an empty secret key. Absence is expressed with
+/// [`RequestCredentials::without_secret_key`], so an empty secret is refused
+/// rather than read as "absent" or used as a key.
+#[derive(Debug)]
+pub(crate) struct EmptySecretKey;
+
+impl fmt::Display for EmptySecretKey {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("handed-over secret key is empty")
+    }
+}
+
+impl std::error::Error for EmptySecretKey {}
+
 impl RequestCredentials {
-    pub(crate) fn new(access_key: String, secret_key: String, session_token: Option<String>) -> Self {
+    /// Credentials whose secret the edge handed over.
+    ///
+    /// # Errors
+    ///
+    /// [`EmptySecretKey`] when `secret_key` is empty.
+    pub(crate) fn new(access_key: String, secret_key: String, session_token: Option<String>) -> Result<Self, EmptySecretKey> {
+        if secret_key.is_empty() {
+            return Err(EmptySecretKey);
+        }
+        Ok(Self {
+            access_key,
+            secret_key: Some(Zeroizing::new(secret_key)),
+            session_token: session_token.map(Zeroizing::new),
+        })
+    }
+
+    /// Credentials whose secret the edge kept.
+    pub(crate) fn without_secret_key(access_key: String, session_token: Option<String>) -> Self {
         Self {
             access_key,
-            secret_key: Zeroizing::new(secret_key),
+            secret_key: None,
             session_token: session_token.map(Zeroizing::new),
         }
     }
 
+    /// The same identity with `session_token` as its token.
+    #[cfg(test)]
+    pub(crate) fn with_session_token(mut self, session_token: Option<String>) -> Self {
+        self.session_token = session_token.map(Zeroizing::new);
+        self
+    }
+
     pub fn access_key(&self) -> &str {
         &self.access_key
     }
 
-    /// Never log this value or compare it with `==`.
-    pub fn secret_key(&self) -> &str {
-        &self.secret_key
+    /// The caller's secret, or `None` when the edge did not hand it over.
+    /// A reader must fail closed on `None`. Never log this value or compare it
+    /// with `==`.
+    pub fn secret_key(&self) -> Option<&str> {
+        self.secret_key.as_deref().map(String::as_str)
     }
 
     pub fn session_token(&self) -> Option<&str> {
         self.session_token.as_deref().map(String::as_str)
     }
+
+    /// The handed-over secret, or [`MissingSecretKey`] when the edge kept it.
+    pub(crate) fn require_secret_key(&self) -> Result<&str, MissingSecretKey> {
+        self.secret_key().ok_or(MissingSecretKey)
+    }
+
+    /// Whether the handed-over secret equals `stored`, compared in constant
+    /// time. An absent secret matches nothing, not even an empty `stored`.
+    pub(crate) fn secret_key_matches(&self, stored: &str) -> bool {
+        self.secret_key()
+            .is_some_and(|secret| crate::auth::constant_time_eq(secret, stored))
+    }
 }
 
+/// A reader needed the caller's secret, but the edge did not hand it over.
+#[derive(Debug)]
+pub(crate) struct MissingSecretKey;
+
+impl fmt::Display for MissingSecretKey {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("the caller's secret key was not handed over for this request")
+    }
+}
+
+impl std::error::Error for MissingSecretKey {}
+
 impl fmt::Debug for RequestCredentials {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("RequestCredentials")
             .field("access_key", &self.access_key)
-            .field("secret_key", &"<redacted>")
+            .field("secret_key", &self.secret_key.as_ref().map(|_| "<redacted>"))
             .field("session_token", &self.session_token.as_ref().map(|_| "<redacted>"))
             .finish()
     }
@@ -128,6 +201,7 @@ pub struct RequestEnvelope {
     verified_presigned: Option<VerifiedPresignedRequest>,
     website_read: Option<WebsiteRead>,
     post_object: Option<PostObjectRequestMarker>,
+    post_object_policy_headers: Option<PostObjectPolicyHeaders>,
     // Attached by the authorization phase (`storage/access.rs`) for the use case.
     req_info: Option<ReqInfo>,
     bucket_generation: Option<BucketGenerationGuard>,
@@ -138,6 +212,13 @@ pub struct RequestEnvelope {
     table_publication_guards: Option<TableDataPlanePublicationGuards>,
     table_list_access: Option<TableDataPlaneListAccess>,
     object_tag_conditions: Option<ObjectTagConditions>,
+    multipart_staging_operation: Option<MultipartStagingOperation>,
+    staged_multipart_part_authorization: Option<StagedMultipartPartAuthorization>,
+    requested_object_lock_retention: Option<RequestedObjectLockRetention>,
+    requested_object_lock_retain_until: Option<RequestedObjectLockRetainUntil>,
+    requested_object_legal_hold: Option<RequestedObjectLegalHold>,
+    requested_object_tags: Option<RequestedObjectTags>,
+    requested_copy_source: Option<RequestedCopySource>,
     residual_extensions: Extensions,
 }
 
@@ -167,6 +248,7 @@ impl RequestEnvelope {
             verified_presigned: extensions.remove(),
             website_read: extensions.remove(),
             post_object: extensions.remove(),
+            post_object_policy_headers: extensions.remove(),
             req_info: extensions.remove(),
             bucket_generation: extensions.remove(),
             pending_delete_bucket_generation: extensions.remove(),
@@ -176,6 +258,13 @@ impl RequestEnvelope {
             table_publication_guards: extensions.remove(),
             table_list_access: extensions.remove(),
             object_tag_conditions: extensions.remove(),
+            multipart_staging_operation: extensions.remove(),
+            staged_multipart_part_authorization: extensions.remove(),
+            requested_object_lock_retention: extensions.remove(),
+            requested_object_lock_retain_until: extensions.remove(),
+            requested_object_legal_hold: extensions.remove(),
+            requested_object_tags: extensions.remove(),
+            requested_copy_source: extensions.remove(),
             residual_extensions: extensions,
         }
     }
@@ -312,6 +401,10 @@ impl RequestEnvelope {
         self.table_publication_guards.as_ref()
     }
 
+    pub(crate) fn install_table_publication_guards(&mut self, guards: TableDataPlanePublicationGuards) {
+        self.table_publication_guards = Some(guards);
+    }
+
     pub(crate) fn table_list_access(&self) -> Option<&TableDataPlaneListAccess> {
         self.table_list_access.as_ref()
     }
@@ -319,19 +412,337 @@ impl RequestEnvelope {
     pub(crate) fn object_tag_conditions(&self) -> Option<&ObjectTagConditions> {
         self.object_tag_conditions.as_ref()
     }
+
+    pub(crate) fn install_object_tag_conditions(&mut self, conditions: ObjectTagConditions) {
+        self.object_tag_conditions = Some(conditions);
+    }
+
+    pub(crate) fn multipart_staging_operation(&self) -> Option<&MultipartStagingOperation> {
+        self.multipart_staging_operation.as_ref()
+    }
+
+    pub(crate) fn staged_multipart_part_authorization(&self) -> Option<&StagedMultipartPartAuthorization> {
+        self.staged_multipart_part_authorization.as_ref()
+    }
+
+    pub(crate) fn requested_object_lock_retention(&self) -> Option<&RequestedObjectLockRetention> {
+        self.requested_object_lock_retention.as_ref()
+    }
+
+    pub(crate) fn requested_object_lock_retain_until(&self) -> Option<&RequestedObjectLockRetainUntil> {
+        self.requested_object_lock_retain_until.as_ref()
+    }
+
+    pub(crate) fn post_object_policy_headers(&self) -> Option<&PostObjectPolicyHeaders> {
+        self.post_object_policy_headers.as_ref()
+    }
+
+    pub(crate) fn requested_object_legal_hold(&self) -> Option<&RequestedObjectLegalHold> {
+        self.requested_object_legal_hold.as_ref()
+    }
+
+    pub(crate) fn requested_object_tags(&self) -> Option<&RequestedObjectTags> {
+        self.requested_object_tags.as_ref()
+    }
+
+    pub(crate) fn requested_copy_source(&self) -> Option<&RequestedCopySource> {
+        self.requested_copy_source.as_ref()
+    }
+}
+
+/// The authorization core in `storage` reads requests only through
+/// [`AuthorizationRequest`]; this impl hands it the envelope's named fields.
+impl AuthorizationRequest for RequestEnvelope {
+    fn headers(&self) -> &HeaderMap {
+        self.headers()
+    }
+    fn uri(&self) -> &http::Uri {
+        self.uri()
+    }
+    fn remote_addr(&self) -> Option<std::net::SocketAddr> {
+        self.remote_addr()
+    }
+    fn client_info(&self) -> Option<&ClientInfo> {
+        self.client_info()
+    }
+    fn server_context(&self) -> Option<&Arc<ServerContextSlot>> {
+        self.server_context()
+    }
+    fn req_info(&self) -> Option<&ReqInfo> {
+        self.req_info()
+    }
+    fn internal_object_authorization(&self) -> bool {
+        false
+    }
+    fn multipart_staging(&self) -> bool {
+        self.multipart_staging_operation().is_some()
+    }
+    fn staged_multipart_part(&self) -> bool {
+        self.staged_multipart_part_authorization().is_some()
+    }
+    fn requested_retention(&self) -> Option<&RequestedObjectLockRetention> {
+        self.requested_object_lock_retention()
+    }
+    fn requested_retain_until(&self) -> Option<&RequestedObjectLockRetainUntil> {
+        self.requested_object_lock_retain_until()
+    }
+    fn requested_legal_hold(&self) -> Option<&RequestedObjectLegalHold> {
+        self.requested_object_legal_hold()
+    }
+    fn requested_object_tags(&self) -> Option<&RequestedObjectTags> {
+        self.requested_object_tags()
+    }
+    fn requested_copy_source(&self) -> Option<&RequestedCopySource> {
+        self.requested_copy_source()
+    }
+    fn policy_headers(&self) -> &HeaderMap {
+        self.post_object_policy_headers().map_or(self.headers(), |headers| &headers.0)
+    }
+    fn object_tag_conditions(&self) -> Option<&ObjectTagConditions> {
+        self.object_tag_conditions()
+    }
+    fn install_object_tag_conditions(&mut self, conditions: ObjectTagConditions) {
+        self.install_object_tag_conditions(conditions);
+    }
+    fn publication_guards(&self) -> Option<&TableDataPlanePublicationGuards> {
+        self.table_publication_guards()
+    }
+    fn install_publication_guards(&mut self, guards: TableDataPlanePublicationGuards) {
+        self.install_table_publication_guards(guards);
+    }
+    fn audit_denial(&self, auditor: &DenialAuditor, action: &Action, reason: &'static str) {
+        auditor.audit_owned(&self.audit_view(), action, reason, DenialScope::Request);
+    }
+}
+
+/// Authorizes an owned request through the same IAM, bucket and table policies as the legacy edge.
+///
+/// Like the legacy `authorize_request`, it first restores the signed `x-amz-meta-*`
+/// query parameters into headers, so policy conditions and duplicate rejection see
+/// the same request shape on both edges.
+///
+/// This is only the policy decision. It does **not** perform the composite checks
+/// the legacy `S3Access` hooks run around their decision, such as the
+/// `s3:BypassGovernanceRetention` authorization for a governance-bypass header or
+/// the copy-source read authorization; a caller that needs them runs them itself.
+///
+/// # Errors
+///
+/// `InvalidArgument` when the signed query repeats a user metadata key;
+/// `AccessDenied` when the decision for `action` depends on a request-body input
+/// the envelope does not carry (see [`require_body_policy_inputs`]) or when the
+/// policies deny the request.
+pub(crate) async fn authorize_owned_request(
+    req: &mut RequestEnvelope,
+    action: Action,
+    location_constraint: Option<String>,
+) -> rustfs_s3_types::S3Result<()> {
+    prepare_owned_authorization(req, action)?;
+    authorize_request_core(req, action, location_constraint, false)
+        .await
+        .map_err(Into::into)
+}
+
+/// Authorizes one exact object key after its enclosing table-catalog operation has been authorized.
+///
+/// It applies the same preparation and has the same limits as [`authorize_owned_request`].
+pub(crate) async fn authorize_owned_internal_object_request(
+    req: &mut RequestEnvelope,
+    action: Action,
+) -> rustfs_s3_types::S3Result<()> {
+    prepare_owned_authorization(req, action)?;
+    authorize_request_core(req, action, None, true).await.map_err(Into::into)
+}
+
+fn prepare_owned_authorization(req: &mut RequestEnvelope, action: Action) -> rustfs_s3_types::S3Result<()> {
+    crate::auth::hoist_signed_query_params(&mut req.headers, req.uri.query()).map_err(rustfs_s3_types::S3Error::from)?;
+    require_body_policy_inputs(req, action)
+}
+
+/// Fails closed when the decision for `action` depends on a value the request
+/// body supplies but the envelope does not carry it. Policy conditions would
+/// otherwise fall back to client headers, which the stored value need not match
+/// (header `OFF` with body `ON` would pass a legal-hold Deny).
+fn require_body_policy_inputs(req: &RequestEnvelope, action: Action) -> rustfs_s3_types::S3Result<()> {
+    let missing = match action {
+        Action::S3Action(S3Action::PutObjectLegalHoldAction) => req.requested_object_legal_hold.is_none(),
+        Action::S3Action(S3Action::PutObjectTaggingAction | S3Action::PutObjectVersionTaggingAction) => {
+            req.requested_object_tags.is_none()
+        }
+        Action::S3Action(S3Action::PutObjectRetentionAction) => req.requested_object_lock_retention.is_none(),
+        _ => false,
+    } || (req.post_object.is_some() && req.post_object_policy_headers.is_none())
+        || (req.headers.contains_key(COPY_SOURCE_HEADER) && req.requested_copy_source.is_none());
+    if missing {
+        return Err(rustfs_s3_types::s3_error!(
+            AccessDenied,
+            "request body policy input is missing for this action"
+        ));
+    }
+    Ok(())
+}
+
+const COPY_SOURCE_HEADER: &str = "x-amz-copy-source";
+
+impl RequestEnvelope {
+    /// The request fields an audit entry for this request is built from.
+    pub(crate) fn audit_view(&self) -> AuditRequestView<'_> {
+        AuditRequestView::from_facts(
+            self.headers(),
+            self.uri(),
+            self.req_info(),
+            self.client_info(),
+            self.remote_addr(),
+            self.request_context(),
+        )
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn authorization_payloads_move_from_extensions_into_named_fields() {
+        let mut extensions = Extensions::new();
+        extensions.insert(MultipartStagingOperation);
+        extensions.insert(StagedMultipartPartAuthorization);
+        let mut retention = RequestedObjectLockRetention(Default::default());
+        retention.0.mode = Some("COMPLIANCE".to_owned().into());
+        extensions.insert(retention);
+        let retain_until = time::OffsetDateTime::from_unix_timestamp(2_000_000_000).expect("valid timestamp");
+        extensions.insert(RequestedObjectLockRetainUntil(retain_until));
+        let head = || RequestHead {
+            method: Method::PUT,
+            uri: "/bucket/key".parse().expect("valid URI"),
+            headers: HeaderMap::new(),
+            credentials: None,
+            region: None,
+            service: None,
+        };
+
+        let envelope = RequestEnvelope::from_parts(head(), extensions);
+        let empty = RequestEnvelope::from_parts(head(), Extensions::new());
+
+        assert_eq!(
+            (
+                envelope.multipart_staging_operation().is_some(),
+                envelope.staged_multipart_part_authorization().is_some(),
+                envelope
+                    .requested_object_lock_retention()
+                    .and_then(|value| value.0.mode.as_ref().map(|mode| mode.as_str())),
+                envelope.requested_object_lock_retain_until().map(|value| value.0),
+                envelope.residual_extensions().is_empty(),
+                empty.multipart_staging_operation().is_none(),
+                empty.staged_multipart_part_authorization().is_none(),
+                empty.requested_object_lock_retention().is_none(),
+                empty.requested_object_lock_retain_until().is_none(),
+            ),
+            (true, true, Some("COMPLIANCE"), Some(retain_until), true, true, true, true, true),
+        );
+    }
+
+    #[test]
+    fn request_body_policy_payloads_move_from_extensions_into_named_fields() {
+        let mut form_headers = HeaderMap::new();
+        form_headers.insert("x-amz-tagging", http::HeaderValue::from_static("from=form"));
+        let mut legal_hold = RequestedObjectLegalHold(Default::default());
+        legal_hold.0.status = Some("ON".to_owned().into());
+        let mut tags = RequestedObjectTags(Default::default());
+        tags.0.tag_set.push(Default::default());
+        tags.0.tag_set[0].key = Some("body".to_owned());
+        let mut extensions = Extensions::new();
+        extensions.insert(PostObjectPolicyHeaders(form_headers));
+        extensions.insert(legal_hold);
+        extensions.insert(tags);
+        extensions.insert(RequestedCopySource("source-bucket/source-key".to_owned()));
+        let head = || RequestHead {
+            method: Method::PUT,
+            uri: "/bucket/key".parse().expect("valid URI"),
+            headers: HeaderMap::new(),
+            credentials: None,
+            region: None,
+            service: None,
+        };
+
+        let envelope = RequestEnvelope::from_parts(head(), extensions);
+        let empty = RequestEnvelope::from_parts(head(), Extensions::new());
+
+        assert_eq!(
+            (
+                envelope
+                    .post_object_policy_headers()
+                    .and_then(|headers| headers.0.get("x-amz-tagging"))
+                    .and_then(|value| value.to_str().ok()),
+                envelope
+                    .requested_object_legal_hold()
+                    .and_then(|hold| hold.0.status.as_ref().map(|status| status.as_str())),
+                envelope
+                    .requested_object_tags()
+                    .and_then(|tags| tags.0.tag_set.first())
+                    .and_then(|tag| tag.key.as_deref()),
+                envelope.requested_copy_source().map(|source| source.0.as_str()),
+                envelope.residual_extensions().is_empty(),
+                empty.post_object_policy_headers().is_none(),
+                empty.requested_object_legal_hold().is_none(),
+                empty.requested_object_tags().is_none(),
+                empty.requested_copy_source().is_none(),
+            ),
+            (
+                Some("from=form"),
+                Some("ON"),
+                Some("body"),
+                Some("source-bucket/source-key"),
+                true,
+                true,
+                true,
+                true,
+                true
+            ),
+        );
+    }
+
+    #[test]
+    fn an_empty_handed_over_secret_is_refused() {
+        assert!(RequestCredentials::new("AKIAENVELOPE".to_owned(), String::new(), None).is_err());
+    }
+
+    #[test]
+    fn an_absent_secret_matches_no_stored_secret() {
+        let kept = RequestCredentials::without_secret_key("AKIAENVELOPE".to_owned(), None);
+        assert_eq!(kept.secret_key(), None);
+        assert!(!kept.secret_key_matches(""), "absent must not equal an empty stored secret");
+        assert!(!kept.secret_key_matches("stored-secret"));
+        let handed = RequestCredentials::new("AKIAENVELOPE".to_owned(), "stored-secret".to_owned(), None).expect("secret");
+        assert!(!handed.secret_key_matches(""));
+        assert!(!handed.secret_key_matches("other-secret"));
+        assert!(handed.secret_key_matches("stored-secret"));
+    }
+
+    #[test]
+    fn a_reader_of_an_absent_secret_fails_closed() {
+        let kept = RequestCredentials::without_secret_key("AKIAENVELOPE".to_owned(), Some("token".to_owned()));
+        assert!(kept.require_secret_key().is_err());
+        let handed = RequestCredentials::new("AKIAENVELOPE".to_owned(), "s".to_owned(), None).expect("secret");
+        assert_eq!(handed.require_secret_key().ok(), Some("s"));
+    }
+
+    #[test]
+    fn kept_secret_debug_prints_neither_secret_nor_token() {
+        let kept = RequestCredentials::without_secret_key("AKIAENVELOPE".to_owned(), Some("envelope-session-token".to_owned()));
+        let debug = format!("{kept:?}");
+        assert!(debug.contains("AKIAENVELOPE"), "{debug}");
+        assert!(!debug.contains("envelope-session-token"), "{debug}");
+    }
+
     #[test]
     fn request_credentials_debug_redacts_secret_and_session_token() {
         let credentials = RequestCredentials::new(
             "AKIAENVELOPE".to_owned(),
             "envelope-secret-value".to_owned(),
             Some("envelope-session-token".to_owned()),
-        );
+        )
+        .expect("non-empty secret");
 
         let debug = format!("{credentials:?}");
 

rustfs/src/app/mod.rs

@@ -36,6 +36,7 @@ pub(crate) mod trailer_adapter;
 // from here rather than through `app::context`.
 pub(crate) use context::RequestHead;
 pub use context::{RequestCredentials, RequestEnvelope};
+pub(crate) use context::{authorize_owned_internal_object_request, authorize_owned_request};
 
 #[cfg(test)]
 mod capacity_dirty_scope_test;
@@ -48,4 +49,6 @@ pub(crate) mod gating_test_env;
 #[cfg(test)]
 mod lifecycle_transition_api_test;
 #[cfg(test)]
+mod owned_authorization_test;
+#[cfg(test)]
 mod put_prelookup_gating_test;

rustfs/src/app/storage_api.rs

@@ -53,10 +53,17 @@ pub(crate) mod s3 {
         PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationTag,
     };
     #[cfg(test)]
+    pub(crate) use s3s::access::S3Access;
+    #[cfg(test)]
     pub(crate) use s3s::auth::{Credentials, SimpleAuth};
     #[cfg(test)]
     pub(crate) use s3s::config::{S3Config, StaticConfigProvider};
     #[cfg(test)]
+    pub(crate) use s3s::dto::{
+        BucketLocationConstraint, CreateBucketConfiguration, CreateBucketInput, CreateMultipartUploadInput, DeleteObjectInput,
+        ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, PutObjectInput, Tag, Tagging,
+    };
+    #[cfg(test)]
     pub(crate) use s3s::dto::{GetObjectInput, HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output};
     #[cfg(test)]
     pub(crate) use s3s::dto::{ListObjectsInput, StreamingBlob, UploadPartInput, UploadPartOutput};
@@ -1815,14 +1822,30 @@ pub(crate) mod select_object {
     };
 }
 
+/// The storage items the owned/legacy authorization parity tests drive.
+#[cfg(test)]
+pub(crate) mod owned_parity {
+    pub(crate) use crate::storage::storage_api::owned_parity_consumer::{
+        DenialAuditor, DenialRateLimiter, DenialScope, FS, ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker,
+        ReqInfo, RequestContext, RequestedCopySource, RequestedObjectLegalHold, RequestedObjectLockRetainUntil,
+        RequestedObjectLockRetention, RequestedObjectTags, SUPPRESSED_DENIALS_TAG, TableDataPlanePublicationGuards,
+        authorization_denial_entry, authorization_denial_entry_for_view, authorize_request, parity_test_support,
+    };
+}
+
 pub(crate) mod context {
     #[cfg(test)]
     pub(crate) use super::EndpointServerPools;
     pub(crate) use super::bucket;
     pub(crate) use super::runtime;
     pub(crate) use crate::storage::storage_api::access_consumer::{
-        BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, ObjectTagConditions,
-        OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo, TableDataPlaneListAccess,
+        AuditRequestView, AuthorizationRequest, DenialAuditor, DenialScope, authorize_request_core,
+    };
+    pub(crate) use crate::storage::storage_api::access_consumer::{
+        BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, MultipartStagingOperation,
+        ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectPolicyHeaders,
+        PostObjectRequestMarker, ReqInfo, RequestedCopySource, RequestedObjectLegalHold, RequestedObjectLockRetainUntil,
+        RequestedObjectLockRetention, RequestedObjectTags, StagedMultipartPartAuthorization, TableDataPlaneListAccess,
         TableDataPlanePublicationGuards,
     };
     pub(crate) use crate::storage::storage_api::request_context_consumer::RequestContext;

rustfs/src/storage/denial_audit.rs

@@ -20,8 +20,10 @@
 //! counted and the count is carried on the next entry that is let through.
 
 use crate::module_switches::is_audit_module_enabled;
-use crate::storage::access::{ReqInfo, request_context_from_req};
-use crate::storage::helper::{DenialRequest, DenialScope, authorization_denial_entry, spawn_background_with_context};
+use crate::storage::helper::{
+    AuditRequestView, DenialRequest, DenialScope, authorization_denial_entry, authorization_denial_entry_for_view,
+    spawn_background_with_context,
+};
 use crate::storage::request_context::RequestContext;
 use rustfs_audit::{AuditEntry, global::AuditLogger};
 use rustfs_policy::policy::action::Action;
@@ -184,21 +186,32 @@ impl DenialAuditor {
         scope: DenialScope,
         now: Instant,
     ) {
+        self.audit_view_at(&AuditRequestView::legacy(req), now, || {
+            authorization_denial_entry(req, action, reason, scope)
+        });
+    }
+
+    /// Audits a denial for a request an owned edge decoded into `view`.
+    pub(crate) fn audit_owned(&self, view: &AuditRequestView<'_>, action: &Action, reason: &'static str, scope: DenialScope) {
+        self.audit_view_at(view, (self.clock)(), || authorization_denial_entry_for_view(view, action, reason, scope));
+    }
+
+    fn audit_view_at(&self, req: &AuditRequestView<'_>, now: Instant, build_entry: impl FnOnce() -> AuditEntry) {
         if !(self.enabled)() {
             return;
         }
-        let anonymous = req.extensions.get::<ReqInfo>().is_none_or(|info| info.cred.is_none());
+        let anonymous = req.req_info.is_none_or(|info| info.cred.is_none());
         let Admission::Emit { suppressed } = self.limiter.admit(anonymous, now) else {
             return;
         };
-        let mut entry = authorization_denial_entry(req, action, reason, scope);
+        let mut entry = build_entry();
         if suppressed > 0 {
             entry
                 .tags
                 .get_or_insert_with(Default::default)
                 .insert(SUPPRESSED_DENIALS_TAG.to_string(), Value::from(suppressed));
         }
-        (self.emit)(request_context_from_req(req), entry);
+        (self.emit)(req.request_context(), entry);
     }
 }
 

rustfs/src/storage/helper.rs

@@ -19,7 +19,7 @@ use crate::storage::request_context::RequestContext;
 use crate::storage::sse::KmsRequestAuditScope;
 use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
 use hashbrown::HashMap;
-use http::StatusCode;
+use http::{HeaderMap, StatusCode, Uri};
 use metrics::counter;
 use rustfs_audit::{
     ApiDetails, AuditEntry, ObjectVersion,
@@ -38,12 +38,64 @@ use rustfs_utils::http::headers::AMZ_REQUEST_ID;
 use s3s::{S3Request, S3Response, S3Result};
 use serde_json::Value;
 use std::future::Future;
+use std::net::SocketAddr;
 use tokio::runtime::{Builder, Handle};
 
 /// Request type seen by the denial auditor, so `denial_audit` stays off the direct s3s surface.
 pub(crate) type DenialRequest<T> = S3Request<T>;
 use tracing::{Instrument, info_span, warn};
 
+/// The request fields used to build an audit entry, shared by both request edges.
+pub(crate) struct AuditRequestView<'a> {
+    headers: &'a HeaderMap,
+    uri: &'a Uri,
+    pub(crate) req_info: Option<&'a ReqInfo>,
+    client_info: Option<&'a ClientInfo>,
+    remote_addr: Option<SocketAddr>,
+    request_context: Option<&'a RequestContext>,
+}
+
+impl<'a> AuditRequestView<'a> {
+    pub(crate) fn legacy<T>(req: &'a S3Request<T>) -> Self {
+        Self {
+            headers: &req.headers,
+            uri: &req.uri,
+            req_info: req.extensions.get::<ReqInfo>(),
+            client_info: req.extensions.get::<ClientInfo>(),
+            remote_addr: req
+                .extensions
+                .get::<Option<RemoteAddr>>()
+                .and_then(|addr| addr.map(|addr| addr.0)),
+            request_context: req.extensions.get::<RequestContext>(),
+        }
+    }
+
+    /// The view of a request whose facts an owned edge already decoded.
+    pub(crate) fn from_facts(
+        headers: &'a HeaderMap,
+        uri: &'a Uri,
+        req_info: Option<&'a ReqInfo>,
+        client_info: Option<&'a ClientInfo>,
+        remote_addr: Option<SocketAddr>,
+        request_context: Option<&'a RequestContext>,
+    ) -> Self {
+        Self {
+            headers,
+            uri,
+            req_info,
+            client_info,
+            remote_addr,
+            request_context,
+        }
+    }
+
+    pub(crate) fn request_context(&self) -> Option<RequestContext> {
+        self.request_context
+            .cloned()
+            .or_else(|| self.req_info.and_then(|info| info.request_context.clone()))
+    }
+}
+
 #[cfg(feature = "diagnostic-read-version-caller")]
 use super::storage_api::ecstore_diagnostic;
 use crate::storage::storage_api::helper_consumer::StorageObjectInfo as ObjectInfo;
@@ -127,32 +179,27 @@ pub(crate) fn build_event_resp_elements<T>(response: &S3Response<T>, request_id:
 
 /// Builds the audit entry shared by every request-scoped audit source: caller identity,
 /// client address and the request line, without any outcome.
-fn request_audit_builder<T>(
-    req: &S3Request<T>,
-    req_info: Option<&ReqInfo>,
+fn request_audit_builder(
+    req: &AuditRequestView<'_>,
     event: EventName,
     trigger: &str,
     api: ApiDetails,
     request_id: &str,
 ) -> AuditEntryBuilder {
     let remote_host = req
-        .extensions
-        .get::<ClientInfo>()
+        .client_info
         .map(|info| info.real_ip.to_string())
-        .or_else(|| {
-            req.extensions
-                .get::<Option<RemoteAddr>>()
-                .and_then(|addr| addr.map(|addr| addr.0.ip().to_string()))
-        })
+        .or_else(|| req.remote_addr.map(|addr| addr.ip().to_string()))
         .unwrap_or_default();
     let mut builder = AuditEntryBuilder::new("1.0", event, trigger, api)
         .remote_host(remote_host)
-        .user_agent(get_request_user_agent(&req.headers))
-        .req_host(get_request_host(&req.headers))
+        .user_agent(get_request_user_agent(req.headers))
+        .req_host(get_request_host(req.headers))
         .req_path(req.uri.path().to_string())
-        .req_query(extract_params_header(&req.headers))
+        .req_query(extract_params_header(req.headers))
         .request_id(request_id);
-    if let Some(cred) = req_info
+    if let Some(cred) = req
+        .req_info
         .and_then(|info| info.cred.as_ref())
         .filter(|cred| !cred.access_key.is_empty())
     {
@@ -374,8 +421,9 @@ impl OperationHelper {
         let request_context = request_context.unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
         let request_id = request_context.request_id.clone();
 
-        let audit_builder =
-            audit_enabled.then(|| request_audit_builder(req, req_info, event, trigger, api_builder.clone().build(), &request_id));
+        let audit_builder = audit_enabled.then(|| {
+            request_audit_builder(&AuditRequestView::legacy(req), event, trigger, api_builder.clone().build(), &request_id)
+        });
 
         let event_object = ObjectInfo {
             bucket: bucket.clone(),
@@ -602,7 +650,16 @@ pub(crate) fn authorization_denial_entry<T>(
     reason: &'static str,
     scope: DenialScope,
 ) -> AuditEntry {
-    let req_info = req.extensions.get::<ReqInfo>();
+    authorization_denial_entry_for_view(&AuditRequestView::legacy(req), action, reason, scope)
+}
+
+pub(crate) fn authorization_denial_entry_for_view(
+    req: &AuditRequestView<'_>,
+    action: &Action,
+    reason: &'static str,
+    scope: DenialScope,
+) -> AuditEntry {
+    let req_info = req.req_info;
     let path = req.uri.path().trim_start_matches('/');
     let mut segs = path.splitn(2, '/');
     let path_bucket = segs.next().unwrap_or("");
@@ -631,23 +688,18 @@ pub(crate) fn authorization_denial_entry<T>(
         api = api.object(object);
     }
 
-    let request_context = request_context_from_req(req).unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
+    let request_context = req
+        .request_context()
+        .unwrap_or_else(|| RequestContext::from_external_headers(req.headers));
     let mut tags = HashMap::new();
     tags.insert("denialReason".to_string(), Value::String(reason.to_string()));
     if let DenialScope::Keys(denied) = scope {
         tags.insert("deniedKeys".to_string(), Value::from(denied));
     }
-    request_audit_builder(
-        req,
-        req_info,
-        EventName::AuditS3Operation,
-        action_name,
-        api.build(),
-        &request_context.request_id,
-    )
-    .error("Access Denied")
-    .tags(tags)
-    .build()
+    request_audit_builder(req, EventName::AuditS3Operation, action_name, api.build(), &request_context.request_id)
+        .error("Access Denied")
+        .tags(tags)
+        .build()
 }
 
 /// Runs `handler` on `req` and records the outcome as an audit-only entry for `op`.
@@ -1435,8 +1487,7 @@ mod tests {
                 ..Default::default()
             });
             let entry = super::request_audit_builder(
-                &req,
-                req.extensions.get::<ReqInfo>(),
+                &super::AuditRequestView::legacy(&req),
                 EventName::ObjectAccessedGet,
                 "GetObject",
                 rustfs_audit::ApiDetails::default(),

rustfs/src/storage/s3_compat/envelope.rs

@@ -25,11 +25,14 @@ impl RequestEnvelope {
     /// Copies every request fact of `req` the app layer reads.
     pub fn from_s3s<T>(req: &S3Request<T>) -> Self {
         let credentials = req.credentials.as_ref().map(|credentials| {
+            let session_token = get_session_token(&req.uri, &req.headers).map(str::to_owned);
+            // An empty secret is not a key: carry it as absent so every reader fails closed.
             RequestCredentials::new(
                 credentials.access_key.clone(),
                 credentials.secret_key.expose().to_owned(),
-                get_session_token(&req.uri, &req.headers).map(str::to_owned),
+                session_token.clone(),
             )
+            .unwrap_or_else(|_| RequestCredentials::without_secret_key(credentials.access_key.clone(), session_token))
         });
         let head = RequestHead {
             method: req.method.clone(),
@@ -178,7 +181,7 @@ mod tests {
         assert_eq!(envelope.headers(), &req.headers);
         let credentials = envelope.credentials().expect("credentials");
         assert_eq!(credentials.access_key(), "AKIAENVELOPE");
-        assert_eq!(credentials.secret_key(), "envelope-secret");
+        assert_eq!(credentials.secret_key(), Some("envelope-secret"));
         assert_eq!(credentials.session_token(), Some("header-token"));
         assert_eq!(envelope.region(), Some("eu-west-3"));
         assert_eq!(envelope.service(), Some("s3"));

rustfs/src/storage/storage_api.rs

@@ -147,17 +147,34 @@ pub(crate) use super::sse::{
 
 pub(crate) mod access_consumer {
     pub(crate) use super::super::access::{
-        BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, MultipartStagingOperation,
-        ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo,
+        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_internal_object_request, authorize_request, bucket_config_mutation_incarnation,
+        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,
     };
+    pub(crate) use super::super::denial_audit::DenialAuditor;
+    pub(crate) use super::super::helper::{AuditRequestView, DenialScope};
+}
+
+#[cfg(test)]
+pub(crate) mod owned_parity_consumer {
+    pub(crate) use super::super::access::{
+        ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker, ReqInfo, RequestedCopySource,
+        RequestedObjectLegalHold, RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags,
+        TableDataPlanePublicationGuards, authorize_request, parity_test_support,
+    };
+    pub(crate) use super::super::denial_audit::{DenialAuditor, DenialRateLimiter, SUPPRESSED_DENIALS_TAG};
+    pub(crate) use super::super::ecfs::FS;
+    pub(crate) use super::super::helper::{DenialScope, authorization_denial_entry, authorization_denial_entry_for_view};
+    pub(crate) use super::super::request_context::RequestContext;
 }
 
 pub(crate) mod concurrency_consumer {

scripts/README.md

@@ -57,6 +57,9 @@ their issue closes.
 | `test_object_generation_protocol_model.py` | dev-tool | Bounded single-slot, four-voter promise/accept design model and recovery/retirement examples; no production runtime evidence | `python3 scripts/test_object_generation_protocol_model.py`; [Generation contract](../docs/architecture/unified-object-generation.md#executable-design-model-and-acceptance-mapping) |
 | `junit_diff.py` | dev-tool | Case-by-case pass/fail/skip diff of two JUnit XML reports (reference vs candidate stack); exit 1 on any difference or stale `--allow` entry, 2 on unreadable input | `python3 scripts/junit_diff.py a.xml b.xml`; backlog T3.1 |
 | `test_junit_diff.py` | ci-gate | Contract tests and fixtures (`scripts/tests/junit_diff/`) for `junit_diff.py` | `make script-tests` |
+| `s3-tests/reference_parity.py` | ci-gate | Plans the reference parity targets, lists a run's failures for the single rerun, and compares every candidate's s3-tests JUnit against the pinned reference per node id with a tolerated-difference file | `.github/workflows/reference-parity.yml`; backlog T3.2 |
+| `s3-tests/test_reference_parity.py` | ci-gate | Contract tests for `reference_parity.py` | `make script-tests` |
+| `test_reference_binary.py` | ci-gate | Contract tests for `.github/actions/reference-binary/reference_binary.py` (pin resolution and verified unpacking) | `make script-tests` |
 | `e2e-run.sh` | ci-gate | Boots a rustfs server and runs the `s3s-e2e` black-box conformance tool against it | ci.yml `e2e-tests` jobs; `docs/testing/README.md` |
 | `run_ecstore_validation_suite.sh` | dev-tool | ecstore black-box validation suite (`quick`/`full`/`destructive`/`fuzz` profiles) | `docs/testing/README.md`, `docs/testing/ecstore-validation-suite-design.md` |
 | `run_e2e_tests.sh` | dev-tool | Local `e2e_test` crate runner (starts a server, applies filters, cleans up) | `crates/e2e_test/README.md` |

scripts/s3-tests/reference_parity.py

@@ -0,0 +1,329 @@
+#!/usr/bin/env python3
+# Copyright 2026 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.
+"""Plan and judge the s3-tests reference parity lane.
+
+Subcommands:
+  targets    validate the target list and print it as a JSON matrix
+  export-env validate one target's extra environment and print KEY=VALUE lines
+  failed     print the failed test names of one run, in implemented_tests.txt form
+  compare    compare every candidate against the reference, per pytest node id
+
+`compare` reads <reports>/<target>/selected-nodeids.txt, junit.xml and the
+optional rerun-junit.xml of every target. Each node id is reduced to pass /
+fail / skip / missing; a case that failed and then passed its single rerun is a
+pass and is listed as flaky. A tolerated entry must match a real difference.
+
+Exit codes: 0 no untolerated difference and no stale entry; 1 a difference, a
+stale tolerated entry or an incomplete run; 2 missing or malformed input.
+"""
+
+import argparse
+import json
+from pathlib import Path
+import re
+import sys
+import tomllib
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import junit_diff  # noqa: E402  (sibling script, not a package)
+
+PASS, FAIL, SKIP, MISSING = junit_diff.PASS, junit_diff.FAIL, junit_diff.SKIP, junit_diff.MISSING
+NAME = re.compile(r"[a-z][a-z0-9-]{0,39}")
+ENV_KEY = re.compile(r"RUSTFS_[A-Z0-9_]+")
+ENV_VALUE = re.compile(r"[A-Za-z0-9._,:/=+-]*")
+ISSUE = re.compile(r"https://github\.com/rustfs/[A-Za-z0-9_.-]+/issues/[1-9][0-9]*")
+NODEID = re.compile(r"[A-Za-z0-9_/]+\.py(::[A-Za-z0-9_]+)+(\[[^\]\s]*\])?")
+TEST_FILE_PREFIX = "s3tests/functional/test_s3.py::"
+BINARIES = ("reference", "candidate")
+
+
+class InputError(Exception):
+    """Input that cannot be trusted; maps to exit code 2."""
+
+
+# --- target list -----------------------------------------------------------
+
+def load_targets(path):
+    try:
+        data = tomllib.loads(Path(path).read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise InputError(f"{path}: target list is missing") from exc
+    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
+        raise InputError(f"{path}: cannot read target list: {exc}") from exc
+    if data.get("schema") != 1 or type(data.get("schema")) is not int or set(data) != {"schema", "targets"}:
+        raise InputError(f"{path}: unsupported target list schema")
+    return validate_targets(data["targets"], str(path))
+
+
+def validate_targets(targets, where="targets"):
+    if not isinstance(targets, list) or len(targets) < 2:
+        raise InputError(f"{where}: need one reference and at least one candidate")
+    result, names = [], set()
+    for item in targets:
+        if not isinstance(item, dict) or not set(item) <= {"name", "binary", "env"}:
+            raise InputError(f"{where}: invalid target entry {item!r}")
+        name, binary, env = item.get("name"), item.get("binary"), item.get("env", {})
+        if not isinstance(name, str) or not NAME.fullmatch(name) or name in names:
+            raise InputError(f"{where}: invalid or duplicate target name {name!r}")
+        if binary not in BINARIES:
+            raise InputError(f"{where}: target {name} has unknown binary {binary!r}")
+        validate_env(env, f"{where}: target {name}")
+        names.add(name)
+        result.append({"name": name, "binary": binary, "env": dict(env)})
+    if [t["binary"] for t in result].count("reference") != 1:
+        raise InputError(f"{where}: exactly one target must use the reference binary")
+    return result
+
+
+def validate_env(env, where):
+    if not isinstance(env, dict):
+        raise InputError(f"{where}: env must be a table")
+    for key, value in env.items():
+        if not isinstance(key, str) or not ENV_KEY.fullmatch(key):
+            raise InputError(f"{where}: env key {key!r} must match RUSTFS_[A-Z0-9_]+")
+        if not isinstance(value, str) or not ENV_VALUE.fullmatch(value):
+            raise InputError(f"{where}: env value for {key} has unsupported characters")
+    return env
+
+
+# --- node ids and reports --------------------------------------------------
+
+def nodeid_to_case(nodeid):
+    """Map a pytest node id to the classname::name id that junit_diff uses."""
+    if not NODEID.fullmatch(nodeid):
+        raise InputError(f"invalid pytest node id {nodeid!r}")
+    path, *parts = nodeid.split("::")
+    module = path[: -len(".py")].replace("/", ".")
+    classname = ".".join([module, *parts[:-1]])
+    return junit_diff.case_id(classname, parts[-1])
+
+
+def read_selected(path):
+    try:
+        lines = Path(path).read_text(encoding="utf-8").splitlines()
+    except FileNotFoundError as exc:
+        raise InputError(f"{path}: selected node id list is missing") from exc
+    except (OSError, UnicodeDecodeError) as exc:
+        raise InputError(f"{path}: cannot read selected node ids: {exc}") from exc
+    nodeids = [line.strip() for line in lines if line.strip()]
+    if not nodeids:
+        raise InputError(f"{path}: selected node id list is empty")
+    if len(set(nodeids)) != len(nodeids):
+        raise InputError(f"{path}: duplicate selected node id")
+    by_case = {}
+    for nodeid in nodeids:
+        case = nodeid_to_case(nodeid)
+        if case in by_case:
+            raise InputError(f"{path}: node ids {by_case[case]} and {nodeid} collide")
+        by_case[case] = nodeid
+    return nodeids, by_case
+
+
+def target_states(directory):
+    """Return (nodeids, {nodeid: state}, [flaky nodeids]) for one target run."""
+    directory = Path(directory)
+    if not directory.is_dir():
+        raise InputError(f"{directory}: report directory is missing")
+    nodeids, by_case = read_selected(directory / "selected-nodeids.txt")
+    try:
+        first, _ = junit_diff.parse_report(str(directory / "junit.xml"))
+    except junit_diff.InputError as exc:
+        raise InputError(str(exc)) from exc
+    unexpected = sorted(set(first) - set(by_case))
+    if unexpected:
+        raise InputError(f"{directory}: junit.xml has unselected cases: {', '.join(unexpected[:5])}")
+    states = {by_case[case]: state for case, state in first.items()}
+    flaky = []
+    rerun_path = directory / "rerun-junit.xml"
+    if rerun_path.exists():
+        try:
+            rerun, _ = junit_diff.parse_report(str(rerun_path))
+        except junit_diff.InputError as exc:
+            raise InputError(str(exc)) from exc
+        for case, state in sorted(rerun.items()):
+            nodeid = by_case.get(case)
+            if nodeid is None or states.get(nodeid) != FAIL:
+                raise InputError(f"{rerun_path}: {case} was rerun without failing first")
+            if state == PASS:
+                flaky.append(nodeid)
+            states[nodeid] = PASS if state == PASS else FAIL
+    for nodeid in nodeids:
+        states.setdefault(nodeid, MISSING)
+    return nodeids, states, flaky
+
+
+def failed_names(directory):
+    _, states, _ = target_states(directory)
+    names = []
+    for nodeid, state in states.items():
+        if state == FAIL:
+            if not nodeid.startswith(TEST_FILE_PREFIX):
+                raise InputError(f"cannot express {nodeid} as an implemented_tests.txt entry")
+            names.append(nodeid[len(TEST_FILE_PREFIX):])
+    return sorted(names)
+
+
+# --- tolerated differences ---------------------------------------------------
+
+def read_tolerated(path, candidates):
+    try:
+        lines = Path(path).read_text(encoding="utf-8").splitlines()
+    except FileNotFoundError as exc:
+        raise InputError(f"{path}: tolerated-difference file is missing") from exc
+    except (OSError, UnicodeDecodeError) as exc:
+        raise InputError(f"{path}: cannot read tolerated differences: {exc}") from exc
+    entries = {}
+    for number, line in enumerate(lines, 1):
+        if not line.strip() or line.lstrip().startswith("#"):
+            continue
+        fields = [part.strip() for part in line.split(" | ")]
+        if len(fields) != 4 or not all(fields):
+            raise InputError(f"{path}:{number}: expected '<target> | <node id> | <issue URL> | <reason>'")
+        target, nodeid, issue, _reason = fields
+        if target not in candidates:
+            raise InputError(f"{path}:{number}: {target!r} is not a candidate target")
+        nodeid_to_case(nodeid)
+        if not ISSUE.fullmatch(issue):
+            raise InputError(f"{path}:{number}: tracking issue must be a https://github.com/rustfs/<repo>/issues/<n> URL")
+        if (target, nodeid) in entries:
+            raise InputError(f"{path}:{number}: duplicate entry for {target} {nodeid}")
+        entries[(target, nodeid)] = issue
+    return entries
+
+
+# --- comparison ----------------------------------------------------------------
+
+def count(states):
+    result = {PASS: 0, FAIL: 0, SKIP: 0, MISSING: 0}
+    for state in states.values():
+        result[state] += 1
+    return result
+
+
+def compare(targets, reports, tolerated_path):
+    reference = next(t["name"] for t in targets if t["binary"] == "reference")
+    candidates = [t["name"] for t in targets if t["binary"] == "candidate"]
+    tolerated = read_tolerated(tolerated_path, set(candidates))
+    runs = {t["name"]: target_states(Path(reports) / t["name"]) for t in targets}
+    selected = runs[reference][0]
+    for name, (nodeids, _, _) in runs.items():
+        if nodeids != selected:
+            raise InputError(f"{name}: selected node ids differ from the {reference} run")
+    ref_states = runs[reference][1]
+    differences = []
+    for name in candidates:
+        states = runs[name][1]
+        for nodeid in selected:
+            if ref_states[nodeid] != states[nodeid]:
+                issue = tolerated.get((name, nodeid))
+                differences.append({"target": name, "nodeid": nodeid, "reference": ref_states[nodeid],
+                                    "candidate": states[nodeid], "tolerated": issue})
+    seen = {(d["target"], d["nodeid"]) for d in differences}
+    stale = [{"target": t, "nodeid": n, "issue": i} for (t, n), i in sorted(tolerated.items()) if (t, n) not in seen]
+    incomplete = {name: [n for n in selected if run[1][n] == MISSING] for name, run in runs.items()}
+    return {
+        "reference": reference,
+        "candidates": candidates,
+        "selected": len(selected),
+        "counts": {name: count(run[1]) for name, run in runs.items()},
+        "flaky": {name: run[2] for name, run in runs.items()},
+        "differences": differences,
+        "stale_tolerated": stale,
+        "incomplete": {name: ids for name, ids in incomplete.items() if ids},
+    }
+
+
+def failed(summary):
+    return bool(summary["stale_tolerated"] or summary["incomplete"]
+                or any(not d["tolerated"] for d in summary["differences"]))
+
+
+def render(summary):
+    out = ["## s3-tests reference parity", ""]
+    out.append(f"Reference target `{summary['reference']}`; {summary['selected']} selected node ids per target.")
+    out += ["", "| Target | pass | fail | skip | missing | flaky (passed on rerun) |", "|---|---:|---:|---:|---:|---:|"]
+    for name, c in summary["counts"].items():
+        out.append(f"| `{name}` | {c[PASS]} | {c[FAIL]} | {c[SKIP]} | {c[MISSING]} | {len(summary['flaky'][name])} |")
+    diffs = summary["differences"]
+    out += ["", f"### Differences ({len(diffs)})", ""]
+    if diffs:
+        out += ["| Target | Node id | reference | candidate | Tolerated |", "|---|---|---|---|---|"]
+        for d in diffs:
+            out.append(f"| `{d['target']}` | `{d['nodeid']}` | {d['reference']} | {d['candidate']} | {d['tolerated'] or 'no'} |")
+    else:
+        out.append("No differences.")
+    if summary["stale_tolerated"]:
+        out += ["", "### Stale tolerated entries (no matching difference; remove them)", ""]
+        out += [f"- `{s['target']}` `{s['nodeid']}` ({s['issue']})" for s in summary["stale_tolerated"]]
+    if summary["incomplete"]:
+        out += ["", "### Incomplete runs (selected node ids without a result)", ""]
+        out += [f"- `{name}`: {len(ids)} missing, first `{ids[0]}`" for name, ids in summary["incomplete"].items()]
+    flaky = [(name, ids) for name, ids in summary["flaky"].items() if ids]
+    if flaky:
+        out += ["", "### Flaky (failed, then passed the single rerun; not a difference)", ""]
+        out += [f"- `{name}`: " + ", ".join(f"`{i}`" for i in ids) for name, ids in flaky]
+    out += ["", "Result: **" + ("FAIL" if failed(summary) else "PASS") + "**", ""]
+    return "\n".join(out)
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    sub = parser.add_subparsers(dest="mode", required=True)
+    tgt = sub.add_parser("targets")
+    tgt.add_argument("--config", required=True)
+    env = sub.add_parser("export-env")
+    env.add_argument("--target-json", required=True)
+    fail = sub.add_parser("failed")
+    fail.add_argument("--report", required=True, help="one target's report directory")
+    cmp_ = sub.add_parser("compare")
+    cmp_.add_argument("--targets-json", required=True)
+    cmp_.add_argument("--reports", required=True)
+    cmp_.add_argument("--tolerated", required=True)
+    cmp_.add_argument("--markdown", help="write the summary table here (appended)")
+    cmp_.add_argument("--json", help="write the machine-readable summary here")
+    args = parser.parse_args(argv)
+    try:
+        if args.mode == "targets":
+            print(json.dumps(load_targets(args.config), separators=(",", ":")))
+        elif args.mode == "export-env":
+            target = json.loads(args.target_json)
+            if not isinstance(target, dict):
+                raise InputError("target must be a JSON object")
+            for key, value in sorted(validate_env(target.get("env", {}), "target").items()):
+                print(f"{key}={value}")
+        elif args.mode == "failed":
+            for name in failed_names(args.report):
+                print(name)
+        else:
+            targets = validate_targets(json.loads(args.targets_json))
+            summary = compare(targets, args.reports, args.tolerated)
+            text = render(summary)
+            print(text)
+            if args.markdown:
+                with open(args.markdown, "a", encoding="utf-8") as handle:
+                    handle.write(text + "\n")
+            if args.json:
+                with open(args.json, "w", encoding="utf-8") as handle:
+                    json.dump(summary, handle, indent=2, sort_keys=True)
+                    handle.write("\n")
+            return 1 if failed(summary) else 0
+    except (InputError, json.JSONDecodeError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/s3-tests/test_reference_parity.py

@@ -0,0 +1,332 @@
+#!/usr/bin/env python3
+# Copyright 2026 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.
+"""Contract tests for reference_parity.py: target list, rerun merge, tolerated entries and exit codes."""
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+SCRIPT = Path(__file__).resolve().parent / "reference_parity.py"
+REPO = Path(__file__).resolve().parents[2]
+PREFIX = "s3tests/functional/test_s3.py::"
+CLASSNAME = "s3tests.functional.test_s3"
+TARGETS = [{"name": "reference", "binary": "reference", "env": {}},
+           {"name": "candidate-legacy", "binary": "candidate", "env": {}}]
+ISSUE = "https://github.com/rustfs/rustfs/issues/1"
+
+
+def junit(states):
+    cases = []
+    for name, state in states.items():
+        body = {"pass": "", "fail": "<failure message='x'/>", "error": "<error message='x'/>",
+                "skip": "<skipped/>"}[state]
+        cases.append(f'<testcase classname="{CLASSNAME}" name="{name}">{body}</testcase>')
+    return f'<testsuites><testsuite name="pytest">{"".join(cases)}</testsuite></testsuites>'
+
+
+def run(*args):
+    proc = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=False)
+    return proc.returncode, proc.stdout, proc.stderr
+
+
+class Lane:
+    """A temporary reports directory with one sub-directory per target."""
+
+    def __init__(self, case):
+        self.temp = tempfile.TemporaryDirectory()
+        case.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.reports = self.root / "reports"
+        self.tolerated = self.root / "tolerated.txt"
+        self.tolerated.write_text("# none\n")
+
+    def target(self, name, states, selected=None, rerun=None):
+        directory = self.reports / name
+        directory.mkdir(parents=True)
+        names = selected if selected is not None else list(states)
+        (directory / "selected-nodeids.txt").write_text("".join(PREFIX + n + "\n" for n in names))
+        (directory / "junit.xml").write_text(junit(states))
+        if rerun is not None:
+            (directory / "rerun-junit.xml").write_text(junit(rerun))
+        return directory
+
+    def compare(self, targets=None):
+        summary = self.root / "summary.json"
+        code, out, err = run("compare", "--targets-json", json.dumps(targets or TARGETS),
+                             "--reports", str(self.reports), "--tolerated", str(self.tolerated),
+                             "--json", str(summary))
+        data = json.loads(summary.read_text()) if summary.exists() else None
+        return code, out, err, data
+
+
+BASE = {"test_a": "pass", "test_b": "pass", "test_c": "skip"}
+
+
+class CompareTests(unittest.TestCase):
+    # --- negative cases ---
+
+    def test_pass_versus_fail_is_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"], [{"target": "candidate-legacy", "nodeid": PREFIX + "test_a",
+                                                "reference": "pass", "candidate": "fail", "tolerated": None}])
+        self.assertIn("Result: **FAIL**", out)
+
+    def test_error_counts_as_fail(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_b": "error"})
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["candidate"], "fail")
+
+    def test_reference_fail_candidate_pass_is_also_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", {**BASE, "test_a": "fail"})
+        lane.target("candidate-legacy", BASE)
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["reference"], "fail")
+
+    def test_skip_versus_pass_is_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_c": "pass"})
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["nodeid"], PREFIX + "test_c")
+
+    def test_case_absent_from_candidate_report_is_missing_and_incomplete(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {"test_a": "pass", "test_c": "skip"}, selected=list(BASE))
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["candidate"], "missing")
+        self.assertEqual(data["incomplete"], {"candidate-legacy": [PREFIX + "test_b"]})
+        self.assertIn("Incomplete runs", out)
+
+    def test_case_missing_from_both_reports_still_fails(self):
+        lane = Lane(self)
+        partial = {"test_a": "pass", "test_c": "skip"}
+        lane.target("reference", partial, selected=list(BASE))
+        lane.target("candidate-legacy", partial, selected=list(BASE))
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"], [])
+        self.assertEqual(set(data["incomplete"]), {"reference", "candidate-legacy"})
+
+    def test_failure_that_also_fails_the_rerun_stays_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"}, rerun={"test_a": "fail"})
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["flaky"]["candidate-legacy"], [])
+
+    def test_rerun_of_a_case_that_did_not_fail_is_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"}, rerun={"test_b": "pass"})
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("rerun without failing first", err)
+
+    def test_missing_target_report_fails_rather_than_skips(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        code, _, err, data = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIsNone(data)
+        self.assertIn("candidate-legacy: report directory is missing", err)
+
+    def test_missing_junit_fails(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        (lane.target("candidate-legacy", BASE) / "junit.xml").unlink()
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("cannot read", err)
+
+    def test_different_selected_lists_are_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {"test_a": "pass", "test_b": "pass"})
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("selected node ids differ", err)
+
+    def test_unselected_case_in_junit_is_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_extra": "pass"}, selected=list(BASE))
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("unselected cases", err)
+
+    def test_stale_tolerated_entry_fails(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", BASE)
+        lane.tolerated.write_text(f"candidate-legacy | {PREFIX}test_a | {ISSUE} | known drift\n")
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["stale_tolerated"][0]["nodeid"], PREFIX + "test_a")
+        self.assertIn("Stale tolerated entries", out)
+
+    def test_tolerated_entry_for_another_target_does_not_cover_the_difference(self):
+        lane = Lane(self)
+        targets = TARGETS + [{"name": "candidate-gateway", "binary": "candidate", "env": {}}]
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        lane.target("candidate-gateway", {**BASE, "test_a": "fail"})
+        lane.tolerated.write_text(f"candidate-gateway | {PREFIX}test_a | {ISSUE} | gateway gap\n")
+        code, _, _, data = lane.compare(targets)
+        self.assertEqual(code, 1)
+        untolerated = [d["target"] for d in data["differences"] if not d["tolerated"]]
+        self.assertEqual(untolerated, ["candidate-legacy"])
+
+    def test_tolerated_entry_without_issue_url_is_rejected(self):
+        for line in (f"candidate-legacy | {PREFIX}test_a | rustfs#1 | reason",
+                     f"candidate-legacy | {PREFIX}test_a | {ISSUE} | ",
+                     f"candidate-legacy | {PREFIX}test_a | {ISSUE}",
+                     f"reference | {PREFIX}test_a | {ISSUE} | reason",
+                     f"candidate-legacy | test_a | {ISSUE} | reason"):
+            with self.subTest(line=line):
+                lane = Lane(self)
+                lane.target("reference", BASE)
+                lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+                lane.tolerated.write_text(line + "\n")
+                code, _, err, _ = lane.compare()
+                self.assertEqual(code, 2, err)
+
+    def test_duplicate_tolerated_entry_is_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        entry = f"candidate-legacy | {PREFIX}test_a | {ISSUE} | reason\n"
+        lane.tolerated.write_text(entry * 2)
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("duplicate entry", err)
+
+    def test_missing_tolerated_file_fails(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", BASE)
+        lane.tolerated.unlink()
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("tolerated-difference file is missing", err)
+
+    # --- positive cases ---
+
+    def test_identical_runs_pass(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", BASE)
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 0)
+        self.assertEqual(data["counts"]["candidate-legacy"], {"pass": 2, "fail": 0, "skip": 1, "missing": 0})
+        self.assertIn("No differences.", out)
+        self.assertIn("Result: **PASS**", out)
+
+    def test_failure_that_passes_the_rerun_is_flaky_not_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"}, rerun={"test_a": "pass"})
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 0)
+        self.assertEqual(data["flaky"]["candidate-legacy"], [PREFIX + "test_a"])
+        self.assertIn("Flaky", out)
+
+    def test_tolerated_difference_passes_and_is_labelled(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        lane.tolerated.write_text(f"# comment\ncandidate-legacy | {PREFIX}test_a | {ISSUE} | known drift\n")
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 0)
+        self.assertEqual(data["differences"][0]["tolerated"], ISSUE)
+        self.assertIn(ISSUE, out)
+
+
+class TargetTests(unittest.TestCase):
+    def write(self, text):
+        temp = tempfile.TemporaryDirectory()
+        self.addCleanup(temp.cleanup)
+        path = Path(temp.name) / "targets.toml"
+        path.write_text(text)
+        return path
+
+    def test_invalid_target_lists_are_rejected(self):
+        ref = '[[targets]]\nname = "reference"\nbinary = "reference"\n'
+        cand = '[[targets]]\nname = "candidate-legacy"\nbinary = "candidate"\n'
+        for label, text in {
+            "no schema": ref + cand,
+            "only reference": "schema = 1\n" + ref,
+            "two references": "schema = 1\n" + ref + ref.replace('"reference"\nbinary', '"other"\nbinary'),
+            "no reference": "schema = 1\n" + cand + cand.replace("candidate-legacy", "candidate-two"),
+            "duplicate name": "schema = 1\n" + ref + cand + cand,
+            "unknown binary": "schema = 1\n" + ref + cand.replace('binary = "candidate"', 'binary = "main"'),
+            "bad name": "schema = 1\n" + ref + cand.replace("candidate-legacy", "Candidate Legacy"),
+            "foreign env key": "schema = 1\n" + ref + cand + 'env = { PATH = "/tmp" }\n',
+            "env injection": "schema = 1\n" + ref + cand + 'env = { RUSTFS_X = "a\\nB=c" }\n',
+            "unknown key": "schema = 1\n" + ref + cand + 'stack = "gateway"\n',
+        }.items():
+            with self.subTest(label):
+                code, _, _ = run("targets", "--config", str(self.write(text)))
+                self.assertEqual(code, 2)
+
+    def test_missing_target_file_fails(self):
+        code, _, err = run("targets", "--config", "/nonexistent/targets.toml")
+        self.assertEqual(code, 2)
+        self.assertIn("target list is missing", err)
+
+    def test_export_env_rejects_foreign_keys(self):
+        code, _, _ = run("export-env", "--target-json", json.dumps({"env": {"LD_PRELOAD": "x"}}))
+        self.assertEqual(code, 2)
+
+    def test_repository_target_list_is_valid(self):
+        code, out, err = run("targets", "--config", str(REPO / ".config/reference-parity-targets.toml"))
+        self.assertEqual(code, 0, err)
+        self.assertEqual(json.loads(out), TARGETS)
+
+    def test_export_env_prints_validated_lines(self):
+        code, out, _ = run("export-env", "--target-json", json.dumps({"env": {"RUSTFS_S3_STACK": "gateway"}}))
+        self.assertEqual((code, out), (0, "RUSTFS_S3_STACK=gateway\n"))
+
+
+class FailedTests(unittest.TestCase):
+    def test_failed_lists_only_failures_in_test_list_form(self):
+        lane = Lane(self)
+        directory = lane.target("reference", {**BASE, "test_a": "fail", "test_d": "error"})
+        code, out, _ = run("failed", "--report", str(directory))
+        self.assertEqual((code, out), (0, "test_a\ntest_d\n"))
+
+    def test_failed_rejects_a_missing_report(self):
+        code, _, err = run("failed", "--report", "/nonexistent/report")
+        self.assertEqual(code, 2)
+        self.assertIn("report directory is missing", err)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_reference_binary.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+# Copyright 2026 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.
+"""Exercise the reference-binary pin resolver and unpacker without network access."""
+
+import hashlib
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+import zipfile
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / ".github/actions/reference-binary/reference_binary.py"
+PIN = '''schema = 1
+repository = "rustfs/rustfs"
+version = "1.0.5"
+tag = "1.0.5"
+
+[platforms.linux-x86_64-gnu]
+asset = "rustfs-linux-x86_64-gnu-v1.0.5.zip"
+sha256 = "{sha}"
+executable = "rustfs"
+'''
+SHA = "a" * 64
+
+
+def run(*args):
+    proc = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=False)
+    return proc.returncode, proc.stdout, proc.stderr
+
+
+class Base(unittest.TestCase):
+    def setUp(self):
+        temp = tempfile.TemporaryDirectory()
+        self.addCleanup(temp.cleanup)
+        self.root = Path(temp.name)
+
+    def pin(self, text):
+        path = self.root / "pin.toml"
+        path.write_text(text)
+        return path
+
+    def archive(self, members):
+        path = self.root / "rustfs.zip"
+        with zipfile.ZipFile(path, "w") as bundle:
+            for name, data in members.items():
+                bundle.writestr(name, data)
+        return path, hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+class ResolveTests(Base):
+    def test_missing_pin_file_fails(self):
+        code, _, err = run("resolve", "--pin", str(self.root / "absent.toml"), "--platform", "linux-x86_64-gnu")
+        self.assertEqual(code, 1)
+        self.assertIn("reference pin file is missing", err)
+
+    def test_unpinned_platform_fails(self):
+        code, _, err = run("resolve", "--pin", str(self.pin(PIN.format(sha=SHA))), "--platform", "linux-aarch64-gnu")
+        self.assertEqual(code, 1)
+        self.assertIn("no pin for platform", err)
+
+    def test_malformed_pins_fail(self):
+        good = PIN.format(sha=SHA)
+        for label, text in {
+            "short digest": good.replace(SHA, "a" * 63),
+            "uppercase digest": good.replace(SHA, "A" * 64),
+            "no digest": good.replace(f'sha256 = "{SHA}"\n', ""),
+            "path in asset": good.replace('asset = "rustfs-', 'asset = "../rustfs-'),
+            "non-zip asset": good.replace("v1.0.5.zip", "v1.0.5.deb"),
+            "asset of another version": good.replace("v1.0.5.zip", "v1.0.4.zip"),
+            "path in executable": good.replace('executable = "rustfs"', 'executable = "bin/rustfs"'),
+            "wrong schema": good.replace("schema = 1", "schema = 2"),
+            "boolean schema": good.replace("schema = 1", "schema = true"),
+            "no repository": good.replace('repository = "rustfs/rustfs"\n', ""),
+            "unknown key": good + 'url = "https://example.invalid"\n',
+            "not toml": "schema = \n",
+        }.items():
+            with self.subTest(label):
+                code, _, _ = run("resolve", "--pin", str(self.pin(text)), "--platform", "linux-x86_64-gnu")
+                self.assertEqual(code, 1)
+
+    def test_resolves_identity_as_step_outputs(self):
+        code, out, err = run("resolve", "--pin", str(self.pin(PIN.format(sha=SHA))), "--platform", "linux-x86_64-gnu")
+        self.assertEqual(code, 0, err)
+        self.assertEqual(out.splitlines(), ["repository=rustfs/rustfs", "version=1.0.5", "tag=1.0.5",
+                                            "asset=rustfs-linux-x86_64-gnu-v1.0.5.zip", f"sha256={SHA}",
+                                            "executable=rustfs"])
+
+    def test_repository_pin_is_valid(self):
+        code, out, err = run("resolve", "--pin", str(ROOT / ".config/reference-binary.toml"),
+                             "--platform", "linux-x86_64-gnu")
+        self.assertEqual(code, 0, err)
+        self.assertIn("version=1.0.5", out.splitlines())
+
+
+class UnpackTests(Base):
+    def unpack(self, archive, sha):
+        return run("unpack", "--archive", str(archive), "--sha256", sha, "--executable", "rustfs",
+                   "--destination", str(self.root / "out"))
+
+    def test_digest_mismatch_fails_and_extracts_nothing(self):
+        archive, _ = self.archive({"rustfs": b"binary"})
+        code, _, err = self.unpack(archive, SHA)
+        self.assertEqual(code, 1)
+        self.assertIn("does not match the pinned digest", err)
+        self.assertFalse((self.root / "out").exists())
+
+    def test_missing_archive_fails(self):
+        code, _, err = self.unpack(self.root / "absent.zip", SHA)
+        self.assertEqual(code, 1)
+        self.assertIn("missing or unreadable", err)
+
+    def test_archive_without_top_level_executable_fails(self):
+        archive, sha = self.archive({"bin/rustfs": b"binary"})
+        code, _, err = self.unpack(archive, sha)
+        self.assertEqual(code, 1)
+        self.assertIn("exactly one top-level", err)
+
+    def test_empty_executable_fails(self):
+        archive, sha = self.archive({"rustfs": b""})
+        code, _, _ = self.unpack(archive, sha)
+        self.assertEqual(code, 1)
+
+    def test_non_zip_archive_fails(self):
+        archive = self.root / "rustfs.zip"
+        archive.write_bytes(b"not a zip")
+        code, _, err = self.unpack(archive, hashlib.sha256(b"not a zip").hexdigest())
+        self.assertEqual(code, 1)
+        self.assertIn("not a valid zip", err)
+
+    def test_extracts_only_the_executable(self):
+        archive, sha = self.archive({"rustfs": b"binary", "../escape": b"x", "README": b"r"})
+        code, out, err = self.unpack(archive, sha)
+        self.assertEqual(code, 0, err)
+        target = self.root / "out" / "rustfs"
+        self.assertEqual(out.strip(), f"path={target}")
+        self.assertEqual(target.read_bytes(), b"binary")
+        self.assertTrue(target.stat().st_mode & 0o111)
+        self.assertEqual(sorted(p.name for p in (self.root / "out").iterdir()), ["rustfs"])
+        self.assertFalse((self.root / "escape").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()