test(access): reference FS through the parent import

The layer dependency guard tracks `use crate::...` edges, and naming
`crate::storage::ecfs::FS` from the signed-query regression looked like a
new `infra -> interface` reverse dependency. `access.rs` already reaches
the same type through `use super::ecfs::FS`, so reuse that import instead
of introducing a second, tracked spelling of an edge the module has
always had. No baseline entry, no new edge.

.github/workflows/nightly-gnu.yml

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

crates/ecstore/src/disk/local.rs

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

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

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

crates/ecstore/src/store/init.rs

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

rustfs/src/admin/handlers/kms_dynamic.rs

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

rustfs/src/app/multipart_usecase.rs

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

rustfs/src/app/object/copy.rs

@@ -256,6 +256,12 @@ impl DefaultObjectUsecase {
             validate_archive_content_encoding(&key, content_type.as_deref(), content_encoding.as_deref())?;
             let mut replacement_metadata = metadata.unwrap_or_default();
             namespace_reserved_user_metadata(&mut replacement_metadata);
+            // s3s parsed `input.metadata` before the query hoist ran, so a
+            // hoisted `x-amz-meta-*` is only visible on the headers now.
+            super::super::storage_api::object_usecase::options::extract_metadata_from_mime(
+                &req.headers,
+                &mut replacement_metadata,
+            );
             apply_standard_object_metadata(
                 &mut replacement_metadata,
                 cache_control.as_deref(),

rustfs/src/app/object/extract.rs

@@ -4522,6 +4522,46 @@ mod tests {
         assert!(options.ignore_errors);
     }
 
+    #[test]
+    fn resolve_put_object_extract_options_sees_hoisted_query_controls() {
+        // The AWS SDK presigner moves `x-amz-meta-snowball-*` into the query
+        // string. The request-hoist layer restores them as headers, and the
+        // extraction options must resolve from them before member headers drop
+        // the transport keys — otherwise members land under their bare names
+        // and can overwrite unrelated objects (rustfs/rustfs#9146).
+        let mut query = String::from("X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID&X-Amz-Signature=deadbeef");
+        query.push_str("&x-amz-meta-snowball-auto-extract=true&x-amz-meta-snowball-prefix=batch/");
+        query.push_str("&x-amz-meta-snowball-ignore-dirs=true&x-amz-meta-minio-snowball-ignore-errors=true");
+        let mut req = http::Request::builder()
+            .method(http::Method::PUT)
+            .uri(format!("/bucket/key?{query}"))
+            .body(())
+            .expect("request");
+        let hoisted_query = req.uri().query().map(str::to_owned);
+        crate::auth::hoist_signed_query_params(req.headers_mut(), hoisted_query.as_deref()).expect("hoist");
+
+        assert!(is_put_object_extract_requested(req.headers()));
+        let options = resolve_put_object_extract_options(req.headers()).unwrap();
+        assert_eq!(options.prefix.as_deref(), Some("batch"));
+        assert!(options.ignore_dirs);
+        assert!(options.ignore_errors);
+
+        // Member headers keep the archive's own metadata and drop the transport
+        // keys, so no member object stores the controls that drove extraction.
+        let member_headers = snowball_member_headers(req.headers());
+        assert!(!member_headers.contains_key(AMZ_SNOWBALL_EXTRACT));
+        assert!(!member_headers.contains_key("x-amz-meta-snowball-prefix"));
+    }
+
+    #[test]
+    fn extract_request_discovery_honors_an_explicit_false_header_over_query() {
+        // Header precedence: `x-amz-meta-snowball-auto-extract: false` must not
+        // fall through to a hoisted `true` of the same key.
+        let mut headers = HeaderMap::new();
+        headers.insert(AMZ_SNOWBALL_EXTRACT, HeaderValue::from_static("false"));
+        assert!(!is_put_object_extract_requested(&headers));
+    }
+
     #[test]
     fn resolve_put_object_extract_options_accepts_standard_headers() {
         let mut headers = HeaderMap::new();

rustfs/src/app/storage_api.rs

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

rustfs/src/auth.rs

@@ -14,7 +14,9 @@
 
 use crate::runtime_sources::{AppContext, ServerContextSlot, current_action_credentials, current_ready_iam_handle};
 use http::HeaderMap;
+use http::HeaderValue;
 use http::Uri;
+use http::header::HeaderName;
 use http::header::{IF_MATCH, IF_NONE_MATCH};
 use rustfs_credentials::Credentials;
 use rustfs_iam::error::Error as IamError;
@@ -24,6 +26,7 @@ use rustfs_iam::sys::{
 use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive, is_server_derived_condition_key};
 use rustfs_trusted_proxies::ClientInfo;
 use rustfs_utils::MaskedAccessKey;
+use rustfs_utils::http::metadata_compat::starts_with_ignore_ascii_case;
 use rustfs_utils::http::{AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER};
 use s3s::S3Error;
 use s3s::S3ErrorCode;
@@ -585,6 +588,27 @@ fn policy_secure_transport(client_info: Option<&ClientInfo>) -> bool {
         .unwrap_or(false)
 }
 
+/// Canonicalize an `s3:VersionId` condition value.
+///
+/// The object handlers accept several spellings of one version id (upper
+/// case, unhyphenated, braced, `urn:uuid:`), and the storage layer resolves
+/// all of them to the same version. Policy conditions must see the single
+/// spelling the server itself emits, otherwise a `Deny` with `StringEquals`
+/// can be bypassed by re-spelling the id. `null` in any case and the nil
+/// UUID both name the null version. Strings that are not a UUID are passed
+/// through unchanged; the handlers reject them later.
+pub(crate) fn canonical_policy_version_id(version: &str) -> String {
+    let trimmed = version.trim();
+    if trimmed.eq_ignore_ascii_case("null") {
+        return "null".to_string();
+    }
+    match uuid::Uuid::parse_str(trimmed) {
+        Ok(id) if id.is_nil() => "null".to_string(),
+        Ok(id) => id.to_string(),
+        Err(_) => version.to_string(),
+    }
+}
+
 /// Get condition values for policy evaluation
 ///
 /// # Arguments
@@ -679,9 +703,6 @@ pub fn get_condition_values_with_query_and_client_info(
     let curr_time = OffsetDateTime::now_utc();
     let epoch_time = curr_time.unix_timestamp();
 
-    // Use provided version ID or empty string
-    let vid = version_id.unwrap_or("");
-
     // Determine auth type and signature version from headers and query
     let (auth_type, signature_version) = determine_auth_type_and_version_with_query(header, query);
 
@@ -709,9 +730,10 @@ pub fn get_condition_values_with_query_and_client_info(
     args.insert("username".to_owned(), vec![username]);
     args.insert("principaltype".to_owned(), vec![principal_type.to_string()]);
 
-    // Add version ID
-    if !vid.is_empty() {
-        args.insert("versionid".to_owned(), vec![vid.to_string()]);
+    // Add version ID. Policy conditions must identify the same version the
+    // storage layer resolves, whatever spelling the client used.
+    if let Some(vid) = version_id.filter(|vid| !vid.is_empty()) {
+        args.insert("versionid".to_owned(), vec![canonical_policy_version_id(vid)]);
     }
 
     // Add signature version and auth type
@@ -972,6 +994,125 @@ fn is_request_signature_v2(header: &HeaderMap) -> bool {
     false
 }
 
+/// True when `query` can name a user-metadata parameter at all. A single ASCII
+/// scan, so a request whose query only carries things like `versionId` or a
+/// typed property never pays for a parse.
+fn query_may_carry_user_metadata(query: &str) -> bool {
+    const NEEDLE: &[u8] = b"-meta-";
+    query
+        .as_bytes()
+        .windows(NEEDLE.len())
+        .any(|window| window.eq_ignore_ascii_case(NEEDLE))
+}
+
+/// Restores signature-covered `x-amz-meta-*` user metadata as request headers.
+///
+/// AWS SDK presigners hoist metadata into the query string, while RustFS's
+/// metadata consumers read headers (rustfs/rustfs#9142). Only user metadata is
+/// restored here. s3s has already parsed typed request properties before this
+/// hook runs; restoring Object Lock, encryption or routing headers here would
+/// bypass the corresponding DTO validation and per-property authorization.
+/// Those properties keep their existing handling (rustfs/backlog#3143).
+///
+/// This runs after s3s has verified the signature and before policy evaluation
+/// and the handler read request properties. It must not run earlier: s3s's own
+/// verifier rejects an `x-amz-*` header outside `X-Amz-SignedHeaders`, which is
+/// exactly what a restored parameter becomes.
+///
+/// Only a query the SigV4 signature covers is honored. SigV4 signs the
+/// canonical query string (AWS General Reference, "Create a signed AWS API
+/// request"; s3s-sigv4 verifies every parameter except the signature itself),
+/// so its query parameters are as trustworthy as its headers. SigV2's
+/// string-to-sign covers headers plus a fixed subresource list rather than
+/// arbitrary query parameters, so a SigV2 presigned URL keeps validating after
+/// `X-Amz-Algorithm`/`X-Amz-Signature` markers are appended to it — those
+/// markers must not be allowed to widen it into accepting query properties the
+/// presigner never authorized. An existing header always wins, so this is purely
+/// additive and the unsigned-header guard keeps rejecting anything the signer did
+/// not sign. Keys and values that could not have arrived as a header are
+/// rejected rather than stored: the read path rebuilds them as header fields,
+/// where a control character would fail every `HeadObject` of that object.
+pub fn hoist_signed_query_params(headers: &mut HeaderMap, query: Option<&str>) -> S3Result<()> {
+    let Some(query) = query else {
+        return Ok(());
+    };
+    if !query_may_carry_user_metadata(query) || is_request_signature_v2(headers) {
+        return Ok(());
+    }
+
+    // One pass over the query: recognize the presigned-SigV4 marker set and
+    // stage user metadata parameters. Nothing is validated
+    // or applied until the gate below passes, so a request whose query is not
+    // signature-covered is never rejected here.
+    let mut algorithm = false;
+    let mut credential = false;
+    let mut signature = false;
+    let mut staged: Vec<(String, String)> = Vec::new();
+    let mut seen = std::collections::HashSet::new();
+    let mut duplicate = false;
+    for (name, value) in form_urlencoded::parse(query.as_bytes()) {
+        if name.eq_ignore_ascii_case("awsaccesskeyid") {
+            // A SigV2 presigned URL is recognized by its own credential
+            // parameter; the SigV4 markers below can be appended to it without
+            // invalidating the SigV2 signature, so it must not be widened.
+            if !value.is_empty() {
+                return Ok(());
+            }
+            continue;
+        }
+        if name.eq_ignore_ascii_case("x-amz-algorithm") {
+            algorithm = value == SIGN_V4_ALGORITHM;
+            continue;
+        }
+        if name.eq_ignore_ascii_case("x-amz-credential") {
+            credential = !value.is_empty();
+            continue;
+        }
+        if name.eq_ignore_ascii_case("x-amz-signature") {
+            signature = !value.is_empty();
+            continue;
+        }
+        // Parameter names keep whatever case the signer used (`X-Amz-Meta-User-Id`
+        // is what a hoist produces), so match the prefix the way `HeaderName` would.
+        if !starts_with_ignore_ascii_case(&name, "x-amz-meta-") {
+            continue;
+        }
+        let stored_name = name.to_ascii_lowercase();
+        duplicate |= !seen.insert(stored_name.clone());
+        staged.push((stored_name, value.into_owned()));
+    }
+
+    if !is_request_signature_v4(headers) && !(algorithm && credential && signature) {
+        return Ok(());
+    }
+    if duplicate {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidArgument,
+            "Duplicate x-amz request parameters in the request query string".to_owned(),
+        ));
+    }
+
+    for (name, value) in staged {
+        // The read path rebuilds `x-amz-meta-<key>` as a header name and the
+        // value as a header value (s3s `add_opt_metadata`); anything that could
+        // not have arrived as a header would 500 every Head/Get of the object.
+        let header_name = HeaderName::from_bytes(name.as_bytes()).map_err(|_| {
+            S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid x-amz request parameter name".to_owned())
+        })?;
+        let header_value = HeaderValue::from_bytes(value.as_bytes()).map_err(|_| {
+            S3Error::with_message(
+                S3ErrorCode::InvalidArgument,
+                "x-amz request parameter values must be valid HTTP header values".to_owned(),
+            )
+        })?;
+        if headers.contains_key(&header_name) {
+            continue;
+        }
+        headers.insert(header_name, header_value);
+    }
+    Ok(())
+}
+
 pub(crate) fn is_request_presigned_signature_v4_with_query(header: &HeaderMap, query: Option<&str>) -> bool {
     if let Some(credential) = header.get(AMZ_CREDENTIAL) {
         return !credential.to_str().unwrap_or("").is_empty();
@@ -1244,22 +1385,47 @@ pub(crate) fn reject_unsigned_amz_headers_on_presigned_request(header: &HeaderMa
         if !name.starts_with("x-amz-") || PRESIGNED_UNSIGNED_AMZ_HEADER_ALLOWLIST.contains(&name) {
             continue;
         }
-        if !signed.iter().any(|signed_name| signed_name == name) {
-            warn!(
-                event = EVENT_PRESIGNED_UNSIGNED_AMZ_HEADER,
-                component = LOG_COMPONENT_AUTH,
-                subsystem = LOG_SUBSYSTEM_REQUEST,
-                reason = "unsigned_amz_header",
-                header = name,
-                "Presigned request rejected"
-            );
-            return Err(S3Error::with_message(S3ErrorCode::AccessDenied, UNSIGNED_HEADERS_MESSAGE.to_string()));
+        if signed.iter().any(|signed_name| signed_name == name) {
+            continue;
         }
+        // The query-hoist layer restores a signed `x-amz-*` query parameter as
+        // a header before this rule runs a second time at the access boundary.
+        // Such a header carries exactly the value the signature covers, so it
+        // is as signed as the parameter itself; a header the signer did not
+        // sign cannot match one.
+        if header_as_signed_query_value(header, query, name) {
+            continue;
+        }
+        warn!(
+            event = EVENT_PRESIGNED_UNSIGNED_AMZ_HEADER,
+            component = LOG_COMPONENT_AUTH,
+            subsystem = LOG_SUBSYSTEM_REQUEST,
+            reason = "unsigned_amz_header",
+            header = name,
+            "Presigned request rejected"
+        );
+        return Err(S3Error::with_message(S3ErrorCode::AccessDenied, UNSIGNED_HEADERS_MESSAGE.to_string()));
     }
 
     Ok(())
 }
 
+/// True when `name` is a single-valued header whose value is exactly the value
+/// of a same-named query parameter.
+///
+/// The SigV4 canonical query string covers every parameter except the signature
+/// itself, so a match proves the value the header carries is authorized. Only
+/// a single-valued header can match: the query-hoist layer never produces a
+/// multi-valued one, and a second value would be unsigned.
+fn header_as_signed_query_value(header: &HeaderMap, query: &str, name: &str) -> bool {
+    let mut values = header.get_all(name).iter();
+    let (Some(value), None) = (values.next(), values.next()) else {
+        return false;
+    };
+    form_urlencoded::parse(query.as_bytes())
+        .any(|(param, param_value)| param.eq_ignore_ascii_case(name) && param_value.as_bytes() == value.as_bytes())
+}
+
 /// Parse the RustFS presigned PutObject size capability after authentication.
 ///
 /// The query value is covered by SigV4 when it is present before presigning, but
@@ -2016,6 +2182,40 @@ mod tests {
         assert_eq!(conditions.get("versionid"), Some(&vec!["server-version".to_string()]));
     }
 
+    #[test]
+    fn version_id_condition_uses_canonical_spelling() {
+        let cred = create_test_credentials();
+        let headers = HeaderMap::new();
+        let canonical = "68f21f07-9bae-4b3a-a120-36f48fb9de1a";
+        for spelling in [
+            canonical,
+            "68F21F07-9BAE-4B3A-A120-36F48FB9DE1A",
+            "68f21f079bae4b3aa12036f48fb9de1a",
+            "{68f21f07-9bae-4b3a-a120-36f48fb9de1a}",
+            "urn:uuid:68f21f07-9bae-4b3a-a120-36f48fb9de1a",
+            " 68f21f07-9bae-4b3a-a120-36f48fb9de1a ",
+        ] {
+            let conditions = get_condition_values(&headers, &cred, Some(spelling), None, None);
+            assert_eq!(conditions.get("versionid"), Some(&vec![canonical.to_string()]), "{spelling}");
+        }
+        for spelling in [
+            "null",
+            "NULL",
+            " null ",
+            "00000000-0000-0000-0000-000000000000",
+            "00000000000000000000000000000000",
+        ] {
+            let conditions = get_condition_values(&headers, &cred, Some(spelling), None, None);
+            assert_eq!(conditions.get("versionid"), Some(&vec!["null".to_string()]), "{spelling}");
+        }
+        // Strings the handlers reject are passed through unchanged.
+        let conditions = get_condition_values(&headers, &cred, Some("not-a-uuid"), None, None);
+        assert_eq!(conditions.get("versionid"), Some(&vec!["not-a-uuid".to_string()]));
+        for absent in [None, Some("")] {
+            assert!(!get_condition_values(&headers, &cred, absent, None, None).contains_key("versionid"));
+        }
+    }
+
     #[test]
     fn ghsa_6r96_claim_condition_keys_ignore_spoofed_headers() {
         // The credential carries no groups/roles claims, so these keys are absent --

rustfs/src/server/layer.rs

@@ -1859,6 +1859,7 @@ where
 /// rulings of [`crate::auth::reject_unsigned_amz_headers_on_sigv4_request`]
 /// must be applied here, in front of s3s. Rejections carry the same S3 error
 /// document the access layer would have produced.
+
 #[derive(Clone, Default)]
 pub struct SigV4HeaderGuardLayer;
 
@@ -4118,6 +4119,262 @@ mod tests {
         assert_eq!(calls.load(Ordering::SeqCst), 2);
     }
 
+    use crate::auth::{hoist_signed_query_params, reject_unsigned_amz_headers_on_presigned_request};
+    use http::header::HeaderName;
+
+    /// A SigV4 presigned query: the presigner moved the request's `x-amz-*`
+    /// headers into it and every parameter is covered by the signature.
+    fn presigned_query(extra: &str) -> String {
+        let mut query =
+            String::from("X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID%2F20260101%2Fus-east-1%2Fs3%2Faws4_request");
+        query.push_str("&X-Amz-Date=20260101T000000Z&X-Amz-Expires=300&X-Amz-SignedHeaders=host&X-Amz-Signature=deadbeef");
+        query.push('&');
+        query.push_str(extra);
+        query
+    }
+
+    fn hoist(uri_query: &str) -> Result<http::Request<()>, S3Error> {
+        let mut req = Request::builder()
+            .method(Method::PUT)
+            .uri(format!("/bucket/key?{uri_query}"))
+            .body(())
+            .expect("request");
+        let hoisted_query = req.uri().query().map(str::to_owned);
+        hoist_signed_query_params(req.headers_mut(), hoisted_query.as_deref())?;
+        Ok(req)
+    }
+
+    #[test]
+    fn hoist_restores_presigned_query_metadata_as_headers() {
+        let req = hoist(&presigned_query("x-amz-meta-user-id=u1&x-amz-tagging=project%3Drustfs")).expect("hoist");
+        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("u1"));
+        assert!(req.headers().get("x-amz-tagging").is_none());
+    }
+
+    #[test]
+    fn hoist_leaves_typed_request_properties_unchanged() {
+        for name in [
+            "x-amz-object-lock-legal-hold",
+            "x-amz-object-lock-mode",
+            "x-amz-object-lock-retain-until-date",
+            "x-amz-server-side-encryption",
+            "x-amz-storage-class",
+            "x-amz-copy-source",
+            "x-amz-content-sha256",
+            "x-amz-tagging",
+        ] {
+            let req = hoist(&presigned_query(&format!("{name}=value"))).expect("metadata hoist");
+            assert!(!req.headers().contains_key(name), "typed property must keep its original path: {name}");
+        }
+    }
+
+    #[test]
+    fn hoist_normalizes_parameter_names_like_header_names() {
+        let req = hoist(&presigned_query("X-Amz-Meta-User-Id=u1")).expect("hoist");
+        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("u1"));
+        // `HeaderName` is lowercase, so the stored key is the normalized one.
+        assert_eq!(req.headers().keys().map(|name| name.as_str()).collect::<Vec<_>>(), ["x-amz-meta-user-id"]);
+    }
+
+    #[test]
+    fn hoist_never_overwrites_a_header_the_client_sent() {
+        let mut req = Request::builder()
+            .method(Method::PUT)
+            .uri(format!("/bucket/key?{}", presigned_query("x-amz-meta-user-id=from-query")))
+            .header("x-amz-meta-user-id", "from-header")
+            .body(())
+            .expect("request");
+        let hoisted_query = req.uri().query().map(str::to_owned);
+        hoist_signed_query_params(req.headers_mut(), hoisted_query.as_deref()).expect("hoist");
+        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("from-header"));
+    }
+
+    #[test]
+    fn hoist_skips_sigv4_auth_params_and_non_amz_parameters() {
+        let req = hoist(&presigned_query("versionId=3&x-amz-security-token=token")).expect("hoist");
+        assert!(req.headers().get("x-amz-algorithm").is_none());
+        assert!(req.headers().get("x-amz-credential").is_none());
+        assert!(req.headers().get("x-amz-signature").is_none());
+        assert!(req.headers().get("x-amz-security-token").is_none());
+        assert!(req.headers().get("versionId").is_none());
+    }
+
+    #[test]
+    fn hoist_ignores_a_sigv2_upload_that_only_claims_sigv4_markers() {
+        // The SigV2 string-to-sign does not cover arbitrary query parameters,
+        // so appending the SigV4 presign markers leaves the original signature
+        // valid while making the query look signed. A URL holder must not be
+        // able to turn on archive extraction that way.
+        let mut query = String::from("AWSAccessKeyId=AKID&Expires=1893456000&Signature=deadbeef");
+        query.push_str("&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID&X-Amz-Signature=deadbeef");
+        query.push_str("&x-amz-meta-snowball-auto-extract=true&x-amz-meta-user-id=u1");
+        let req = hoist(&query).expect("hoist");
+        assert!(req.headers().get("x-amz-meta-snowball-auto-extract").is_none());
+        assert!(req.headers().get("x-amz-meta-user-id").is_none());
+    }
+
+    #[test]
+    fn hoist_requires_the_complete_presign_marker_set() {
+        // A lone marker is not a presigned request; `X-Amz-Signature` alone is
+        // what the unsigned-header guard keys on, not proof of presigning.
+        let req = hoist("X-Amz-Signature=deadbeef&x-amz-meta-user-id=u1").expect("hoist");
+        assert!(req.headers().get("x-amz-meta-user-id").is_none());
+    }
+
+    #[test]
+    fn hoist_preserves_an_explicit_false_extraction_header() {
+        // Header precedence: a client that says `false` in a header must not be
+        // overruled by the hoisted form of the same key.
+        let mut req = Request::builder()
+            .method(Method::PUT)
+            .uri(format!("/bucket/key?{}", presigned_query("x-amz-meta-snowball-auto-extract=true")))
+            .header("x-amz-meta-snowball-auto-extract", "false")
+            .body(())
+            .expect("request");
+        let hoisted_query = req.uri().query().map(str::to_owned);
+        hoist_signed_query_params(req.headers_mut(), hoisted_query.as_deref()).expect("hoist");
+        assert_eq!(
+            req.headers()
+                .get("x-amz-meta-snowball-auto-extract")
+                .and_then(|v| v.to_str().ok()),
+            Some("false")
+        );
+    }
+
+    #[test]
+    fn hoist_ignores_a_query_the_signature_does_not_cover() {
+        // SigV2 signs headers plus a fixed subresource list, not arbitrary
+        // query parameters; honoring them would widen a presigned upload.
+        let req = hoist("AWSAccessKeyId=AKID&Signature=deadbeef&x-amz-meta-user-id=u1").expect("hoist");
+        assert!(req.headers().get("x-amz-meta-user-id").is_none());
+
+        // Anonymous: no signature at all.
+        let req = hoist("x-amz-meta-user-id=u1").expect("hoist");
+        assert!(req.headers().get("x-amz-meta-user-id").is_none());
+    }
+
+    #[test]
+    fn hoist_restores_params_on_a_header_signed_sigv4_request() {
+        let mut req = Request::builder()
+            .method(Method::PUT)
+            .uri("/bucket/key?x-amz-meta-user-id=u1")
+            .header(
+                "authorization",
+                "AWS4-HMAC-SHA256 Credential=AKID/20260101/us-east-1/s3/aws4_request, SignedHeaders=host, Signature=deadbeef",
+            )
+            .body(())
+            .expect("request");
+        let hoisted_query = req.uri().query().map(str::to_owned);
+        hoist_signed_query_params(req.headers_mut(), hoisted_query.as_deref()).expect("hoist");
+        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("u1"));
+    }
+
+    #[test]
+    fn hoist_rejects_names_and_values_that_could_not_be_headers() {
+        // A stored control character would make the read path fail every
+        // HeadObject of that object.
+        let err = hoist(&presigned_query("x-amz-meta-user-id=a%0d%0ab")).expect_err("control char");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+
+        let err = hoist(&presigned_query("x-amz-meta-bad%20key=v")).expect_err("bad name");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn hoist_rejects_duplicate_parameters_after_name_normalization() {
+        let err = hoist(&presigned_query("x-amz-meta-User-Id=1&x-amz-meta-user-id=2")).expect_err("duplicate");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn hoisted_metadata_matches_the_header_path_exactly() {
+        let req = hoist(&presigned_query("x-amz-meta-user-id=u1&x-amz-meta-content-type=custom")).expect("hoist");
+        let mut from_query = HeaderMap::new();
+        for name in ["x-amz-meta-user-id", "x-amz-meta-content-type"] {
+            from_query.insert(HeaderName::from_static(name), req.headers().get(name).expect("hoisted header").clone());
+        }
+
+        let mut headers = HeaderMap::new();
+        headers.insert("x-amz-meta-user-id", HeaderValue::from_static("u1"));
+        headers.insert("x-amz-meta-content-type", HeaderValue::from_static("custom"));
+        assert_eq!(from_query, headers);
+    }
+
+    #[tokio::test]
+    async fn hoist_then_guard_accepts_a_valid_metadata_bearing_presigned_put() {
+        // The AWS SDK presigner emits `SignedHeaders=host` and moves
+        // `x-amz-content-sha256` and `x-amz-meta-*` into the query. Only the
+        // metadata is restored as headers; the digest keeps its dedicated
+        // query verification path. The guard accepts the signed metadata.
+        let inner = CountingHybridService::default();
+        let calls = inner.calls();
+        let mut service = SigV4HeaderGuardLayer.layer(inner);
+
+        let mut query =
+            String::from("X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID%2F20261103%2Fus-east-1%2Fs3%2Faws4_request");
+        query.push_str("&X-Amz-Date=20261103T000000Z&X-Amz-Expires=300&X-Amz-SignedHeaders=host&X-Amz-Signature=deadbeef");
+        query.push_str("&x-amz-content-sha256=UNSIGNED-PAYLOAD&x-amz-meta-user-id=u1");
+
+        let mut request = Request::builder()
+            .method(Method::PUT)
+            .uri(format!("/repro/bucket/meta-query.csv?{query}"))
+            .header("content-type", "text/csv")
+            .body(Full::<Bytes>::from(Bytes::new()))
+            .expect("request");
+        // The authorization boundary hoists the signed query properties before
+        // the handler and before RustFS's own unsigned-header guard re-runs.
+        let hoisted_query = request.uri().query().map(str::to_owned);
+        hoist_signed_query_params(request.headers_mut(), hoisted_query.as_deref()).expect("hoist");
+
+        let response = service.call(request).await.expect("inner response");
+
+        let status = response.status();
+        let body = BodyExt::collect(response.into_body()).await.expect("body").to_bytes();
+        let body = String::from_utf8(body.to_vec()).expect("utf8 body");
+        assert_eq!(calls.load(Ordering::SeqCst), 1, "status={status} body={body}");
+        assert_eq!(status, StatusCode::IM_A_TEAPOT, "body={body}");
+    }
+
+    #[test]
+    fn hoist_rejects_a_control_character_value_with_invalid_argument() {
+        let err = hoist(&presigned_query("x-amz-meta-user-id=a%0d%0ab")).expect_err("control char");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[tokio::test]
+    async fn sigv4_header_guard_accepts_a_header_that_matches_a_signed_query_value() {
+        // The hoist layer restores `?x-amz-meta-user-id=u1` as a header before
+        // the access boundary re-runs this rule; the value it carries is the one
+        // the signature covers.
+        let uri = format!("/bucket/key?{}", presigned_query("x-amz-meta-user-id=u1"));
+        let mut headers = HeaderMap::new();
+        headers.insert("x-amz-meta-user-id", HeaderValue::from_static("u1"));
+        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(uri.split('?').nth(1).unwrap())).is_ok());
+    }
+
+    #[tokio::test]
+    async fn sigv4_header_guard_still_rejects_an_unsigned_header() {
+        let uri = format!("/bucket/key?{}", presigned_query("x-amz-meta-user-id=u1"));
+        let query = uri.split('?').nth(1).unwrap();
+
+        // Same name, different value: the signer did not authorize it.
+        let mut headers = HeaderMap::new();
+        headers.insert("x-amz-meta-user-id", HeaderValue::from_static("evil"));
+        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(query)).is_err());
+
+        // No such parameter at all.
+        let mut headers = HeaderMap::new();
+        headers.insert("x-amz-meta-other", HeaderValue::from_static("v"));
+        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(query)).is_err());
+
+        // Multi-valued: the hoist layer never produces one, and the extra value
+        // is unsigned.
+        let mut headers = HeaderMap::new();
+        headers.append("x-amz-meta-user-id", HeaderValue::from_static("u1"));
+        headers.append("x-amz-meta-user-id", HeaderValue::from_static("evil"));
+        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(query)).is_err());
+    }
+
     #[tokio::test]
     async fn virtual_host_style_hint_layer_short_circuits_unroutable_put() {
         let inner = CountingHybridService::default();

rustfs/src/storage/access.rs

@@ -1415,6 +1415,13 @@ impl BulkDenialAudit {
 
 /// Authorizes the request based on the action and credentials.
 pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3Result<()> {
+    // Restore the `x-amz-meta-*` user metadata the SigV4 presigners hoist into
+    // the query string. This runs after s3s has verified the signature — its own
+    // verifier rejects an `x-amz-*` header outside `SignedHeaders`, which is
+    // exactly what a hoisted parameter becomes — and before policy evaluation and
+    // the handler read request properties, so every operation sees one request
+    // shape (rustfs/rustfs#9142).
+    crate::auth::hoist_signed_query_params(&mut req.headers, req.uri.query())?;
     authorize_request_with_location_constraint(req, action, None).await
 }
 
@@ -6667,3 +6674,163 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
     }
 }
+
+#[cfg(test)]
+mod signed_query_lock_permission_regression {
+    use super::FS;
+    use crate::runtime_sources::{AppContext, ServerContextSlot};
+    use crate::storage::storage_api::contract::bucket::{BucketOperations, MakeBucketOptions};
+    use crate::storage::storage_api::contract::object::ObjectOperations;
+    use hmac::{Hmac, KeyInit, Mac};
+    use sha2::{Digest, Sha256};
+    use std::sync::Arc;
+
+    fn presigned_lock_put(bucket: &str, key: &str, hoisted: bool) -> http::Request<s3s::Body> {
+        let now = time::OffsetDateTime::now_utc();
+        let stamp = now
+            .format(
+                &time::format_description::parse_borrowed::<2>("[year][month][day]T[hour][minute][second]Z")
+                    .expect("timestamp format"),
+            )
+            .expect("timestamp");
+        let day = &stamp[..8];
+        let scope = format!("{day}/us-east-1/s3/aws4_request");
+        let signed_headers = if hoisted {
+            "host"
+        } else {
+            "host;x-amz-object-lock-legal-hold"
+        };
+        let mut params = std::collections::BTreeMap::from([
+            ("X-Amz-Algorithm", "AWS4-HMAC-SHA256".to_owned()),
+            ("X-Amz-Credential", format!("review-local-access/{scope}")),
+            ("X-Amz-Date", stamp.clone()),
+            ("X-Amz-Expires", "300".to_owned()),
+            ("X-Amz-SignedHeaders", signed_headers.to_owned()),
+        ]);
+        if hoisted {
+            params.insert("x-amz-object-lock-legal-hold", "ON".to_owned());
+            params.insert("x-amz-meta-user-id", "u1".to_owned());
+        }
+        let query = params
+            .into_iter()
+            .map(|(k, v)| format!("{}={}", urlencoding::encode(k), urlencoding::encode(&v)))
+            .collect::<Vec<_>>()
+            .join("&");
+        let path = format!("/{bucket}/{key}");
+        let lock_header = if hoisted { "" } else { "x-amz-object-lock-legal-hold:ON\n" };
+        let canonical = format!("PUT\n{path}\n{query}\nhost:localhost\n{lock_header}\n{signed_headers}\nUNSIGNED-PAYLOAD");
+        let to_sign = format!(
+            "AWS4-HMAC-SHA256\n{stamp}\n{scope}\n{}",
+            hex_simd::encode_to_string(Sha256::digest(canonical), hex_simd::AsciiCase::Lower)
+        );
+        let mut signing_key = b"AWS4review-local-secret".to_vec();
+        for value in [day, "us-east-1", "s3", "aws4_request"] {
+            let mut mac = <Hmac<Sha256> as KeyInit>::new_from_slice(&signing_key).expect("signing key");
+            mac.update(value.as_bytes());
+            signing_key = mac.finalize().into_bytes().to_vec();
+        }
+        let mut mac = <Hmac<Sha256> as KeyInit>::new_from_slice(&signing_key).expect("signing key");
+        mac.update(to_sign.as_bytes());
+        let signature = hex_simd::encode_to_string(mac.finalize().into_bytes(), hex_simd::AsciiCase::Lower);
+        let mut request = http::Request::builder()
+            .method("PUT")
+            .uri(format!("http://localhost{path}?{query}&X-Amz-Signature={signature}"))
+            .header("host", "localhost")
+            .header("content-length", "4")
+            .header("content-md5", "CY9rzUYh03PK3k6DJie09g==");
+        if !hoisted {
+            request = request.header("x-amz-object-lock-legal-hold", "ON");
+        }
+        request
+            .body(s3s::Body::from(bytes::Bytes::from_static(b"test")))
+            .expect("signed request")
+    }
+
+    #[tokio::test]
+    async fn signed_query_metadata_cannot_add_legal_hold_without_permission() {
+        use http_body_util::BodyExt;
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        let iam = rustfs_iam::init_iam_sys(store.clone()).await.expect("test IAM");
+        let context = Arc::new(AppContext::with_default_interfaces(
+            store.clone(),
+            iam,
+            Arc::new(rustfs_kms::KmsServiceManager::new()),
+        ));
+        assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+            access_key: "review-local-access".to_owned(),
+            secret_key: "review-local-secret".to_owned(),
+            status: "on".to_owned(),
+            ..Default::default()
+        }));
+        let slot = ServerContextSlot::new();
+        assert!(slot.install(context));
+        let fs = FS::with_server_ctx(slot);
+        let bucket = "review-lock-permission";
+        store
+            .make_bucket(
+                bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("locked bucket");
+        let policy = format!(
+            r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","Principal":{{"AWS":"*"}},"Action":["s3:PutObject"],"Resource":["arn:aws:s3:::{bucket}/*"]}},{{"Effect":"Deny","Principal":{{"AWS":"*"}},"Action":["s3:PutObjectLegalHold","s3:PutObjectRetention"],"Resource":["arn:aws:s3:::{bucket}/*"]}}]}}"#
+        );
+        let mut metadata = (*crate::storage::get_bucket_metadata(bucket).await.expect("bucket metadata")).clone();
+        metadata.policy_config = Some(serde_json::from_str(&policy).expect("bucket policy"));
+        metadata.policy_config_json = policy.into_bytes();
+        crate::storage::storage_api::set_bucket_metadata(bucket.to_owned(), metadata)
+            .await
+            .expect("set bucket policy");
+        let mut builder = s3s::service::S3ServiceBuilder::new(fs.clone());
+        builder.set_auth(crate::auth::IAMAuth::with_server_context(
+            "review-local-access",
+            "review-local-secret",
+            fs.server_ctx().clone(),
+        ));
+        builder.set_access(fs);
+        let mut config = s3s::config::S3Config::default();
+        config.normalize_forward_slash_path = true;
+        config.enable_sig_v2 = true;
+        config.put_object_max_size = Some(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE);
+        config.sig_v4_allowed_services.push("s3tables".to_owned());
+        builder.set_config(Arc::new(s3s::config::StaticConfigProvider::new(Arc::new(config))));
+        let service = builder.build();
+        let response = service
+            .call(presigned_lock_put(bucket, "header-control", false))
+            .await
+            .expect("header request");
+        let status = response.status();
+        let body = response.into_body().collect().await.expect("header body").to_bytes();
+        assert_eq!(status, http::StatusCode::FORBIDDEN, "header control: {}", String::from_utf8_lossy(&body));
+        assert!(
+            String::from_utf8_lossy(&body).contains("<Code>AccessDenied</Code>"),
+            "signed header must reach the independent permission check: {}",
+            String::from_utf8_lossy(&body)
+        );
+        let response = service
+            .call(presigned_lock_put(bucket, "query-target", true))
+            .await
+            .expect("query request");
+        let status = response.status();
+        let body = response.into_body().collect().await.expect("query body").to_bytes();
+        let stored = store
+            .get_object_info(bucket, "query-target", &crate::storage::ObjectOptions::default())
+            .await;
+        assert_eq!(
+            status,
+            http::StatusCode::OK,
+            "ordinary metadata PUT should succeed: {}",
+            String::from_utf8_lossy(&body)
+        );
+        let stored = stored.expect("ordinary metadata PUT must publish an object");
+        assert!(
+            !stored.user_defined.contains_key("x-amz-object-lock-legal-hold"),
+            "query metadata must not grant a separate Object Lock capability"
+        );
+        assert_eq!(stored.user_defined.get("user-id").map(String::as_str), Some("u1"));
+    }
+}

rustfs/src/storage/ecfs_extend.rs

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

rustfs/src/storage/options.rs

@@ -1072,7 +1072,9 @@ pub fn parse_copy_source_range(range_str: &str) -> S3Result<HTTPRangeSpec> {
 ///
 /// s3s reads the header only. The one case it does not cover is a presigned
 /// request whose digest arrives in the query string alone, which is why that
-/// path still hands the digest to the storage layer.
+/// path still hands the digest to the storage layer. The query-hoist restores
+/// only `x-amz-meta-*` user metadata (rustfs/backlog#3143), so it never
+/// manufactures the header this checks for.
 pub(crate) fn get_content_sha256_with_query(headers: &HeaderMap<HeaderValue>, query: Option<&str>) -> Option<String> {
     match get_request_auth_type_with_query(headers, query) {
         AuthType::Presigned if !headers.contains_key(AMZ_CONTENT_SHA256) => {