chore: merge repaired main for drive diagnostics acceptance

* fix: repair s3s footprint ratchet violation, clippy warnings, and stack overflow

- Replace s3_error! macro with S3Error::with_message in ecfs.rs to keep
  s3s footprint ratchet at baseline (1586) after retention validation refactor
- Remove redundant clone on Copy type in list_objects.rs (clippy::redundant_clone)
- Remove clone_on_copy calls in access.rs test (clippy::clone_on_copy)
- Use manual thread with 8MB stack for heal_concurrent_delete_test to fix
  stack overflow on macOS (consistent with sibling tests in same file)

* test: align CI assertions with version authorization rules

.config/nextest.toml

@@ -92,6 +92,12 @@ setup = 'ecstore-large-stack'
 filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'
 
+# These handler/store fixtures and the relocated-pool GET compose deep storage
+# futures in libtest threads; on Linux they overflow the default 2 MiB stack.
+[[profile.default.scripts]]
+filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
+setup = 'ecstore-base-stack'
+
 [[profile.default.scripts]]
 filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
 setup = 'lifecycle-large-stack'
@@ -337,6 +343,10 @@ setup = 'ecstore-large-stack'
 filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'
 
+[[profile.ci.scripts]]
+filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
+setup = 'ecstore-base-stack'
+
 [[profile.ci.scripts]]
 filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
 setup = 'lifecycle-large-stack'

.github/workflows/build.yml

@@ -440,6 +440,17 @@ jobs:
             fi
           done
 
+      - name: Verify macOS binary has no dynamic liblzma dependency
+        if: matrix.platform == 'macos'
+        shell: bash
+        run: |
+          binary="target/${{ matrix.target }}/release/rustfs"
+          if otool -L "$binary" | grep -E 'liblzma.*\.dylib'; then
+            echo "macOS release binary must not depend on a dynamic liblzma library" >&2
+            exit 1
+          fi
+          "$binary" --version
+
       - name: Generate the CPU symbol catalog
         if: endsWith(matrix.target, '-linux-gnu') && matrix.features == 'pyroscope'
         shell: bash

Cargo.lock

@@ -10594,6 +10594,7 @@ dependencies = [
  "faster-hex",
  "futures",
  "hex-simd",
+ "hmac 0.13.0",
  "hotpath",
  "http 1.5.0",
  "http-body-util",
@@ -11060,6 +11061,7 @@ dependencies = [
  "async-compression",
  "futures",
  "hotpath",
+ "liblzma",
  "rustfs-rio",
  "rustfs-tokio-tar",
  "serde",

crates/e2e_test/src/sts_query_compat_test.rs

@@ -740,7 +740,7 @@ async fn test_bucket_tags_opa_contract() -> TestResult {
         .version_id(version)
         .send()
         .await?;
-    opa.expect_bucket_tags("s3:GetObject", source, Some("finance")).await?;
+    opa.expect_bucket_tags("s3:GetObjectVersion", source, Some("finance")).await?;
     client.list_objects_v2().bucket(source).send().await?;
     opa.expect_bucket_tags("s3:ListBucket", source, Some("finance")).await?;
     client.list_object_versions().bucket(source).send().await?;

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -2939,6 +2939,20 @@ fn resolve_tier_free_version_recovery_enabled(value: Result<String, env::VarErro
 }
 
 fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<()>) -> Option<JoinHandle<()>> {
+    spawn_tier_free_version_recovery_once_with_options(
+        api,
+        started,
+        TIER_FREE_VERSION_RECOVERY_LEADER_LOCK.to_string(),
+        tier_free_version_recovery_leader_retry_delay,
+    )
+}
+
+fn spawn_tier_free_version_recovery_once_with_options(
+    api: Arc<ECStore>,
+    started: &OnceLock<()>,
+    leader_lock_name: String,
+    retry_delay: fn() -> StdDuration,
+) -> Option<JoinHandle<()>> {
     if !tier_free_version_recovery_enabled() {
         warn!(
             event = EVENT_LIFECYCLE_WORKER_STATE,
@@ -2956,15 +2970,15 @@ fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<(
     Some(tokio::spawn(async move {
         let cancel_token = api.ctx.background_cancel_token().unwrap_or_default();
         let acquire_api = Arc::clone(&api);
+        let acquire_lock_name = Arc::<str>::from(leader_lock_name);
         run_tier_free_version_recovery_coordinator(
             cancel_token,
-            tier_free_version_recovery_leader_retry_delay,
+            retry_delay,
             move || {
                 let api = Arc::clone(&acquire_api);
+                let lock_name = Arc::clone(&acquire_lock_name);
                 async move {
-                    let lock = api
-                        .new_ns_lock(RUSTFS_META_BUCKET, TIER_FREE_VERSION_RECOVERY_LEADER_LOCK)
-                        .await?;
+                    let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &lock_name).await?;
                     match lock.get_write_lock_quiet(TIER_FREE_VERSION_RECOVERY_LOCK_TIMEOUT).await {
                         Ok(guard) => Ok(Some(guard)),
                         Err(LockError::Timeout { .. }) => Ok(None),
@@ -14193,6 +14207,10 @@ mod tests {
         .await;
     }
 
+    fn fast_tier_free_version_recovery_leader_retry_delay() -> StdDuration {
+        StdDuration::from_millis(10)
+    }
+
     #[tokio::test]
     #[serial]
     async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() {
@@ -14208,8 +14226,13 @@ mod tests {
         seed_recoverable_free_version(&disk_paths, &bucket, object, None, None).await;
 
         let started = OnceLock::new();
-        let recovery = super::spawn_tier_free_version_recovery_once(Arc::clone(&ecstore), &started)
-            .expect("production recovery entrypoint should start once");
+        let recovery = super::spawn_tier_free_version_recovery_once_with_options(
+            Arc::clone(&ecstore),
+            &started,
+            format!("tier-free-version-recovery/test/{}", Uuid::new_v4()),
+            fast_tier_free_version_recovery_leader_retry_delay,
+        )
+        .expect("production recovery entrypoint should start once");
         let task = tokio::time::timeout(StdDuration::from_secs(30), async { recovery_rx.lock().await.recv().await })
             .await
             .expect("production recovery entrypoint should enqueue the seeded item")

crates/ecstore/src/bucket/metadata_sys.rs

@@ -1292,14 +1292,34 @@ pub(crate) async fn get_object_lock_config_state_in(
 pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_in(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
+) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
+    get_object_lock_config_and_incarnation_from_disk_with_guard_in(ctx, bucket, None).await
+}
+
+/// Bucket deletion already owns the namespace/lifecycle fences. Reuse its
+/// live guard rather than recursively acquiring the bucket namespace lock.
+/// The caller must also hold the metadata transaction read lock.
+pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_with_guard_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    guard: Option<&rustfs_lock::NamespaceLockGuard>,
 ) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
     let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await.clone();
 
-    match bucket_meta_sys
-        .read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
-        .await?
-    {
+    let authority = match guard {
+        Some(guard) => {
+            bucket_meta_sys
+                .read_authoritative_metadata_from_disk_under_guard(bucket, guard)
+                .await?
+        }
+        None => {
+            bucket_meta_sys
+                .read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
+                .await?
+        }
+    };
+    match authority {
         BucketMetadataAuthority::Authoritative(metadata)
             if metadata.bucket_incarnation_sidecar && !metadata.bucket_incarnation_id.is_nil() =>
         {

crates/ecstore/src/disk/disk_store.rs

@@ -2911,7 +2911,7 @@ mod tests {
 
         let timed = wrapper.track_disk_health_with_op(
             "write_all",
-            || std::future::pending::<Result<()>>(),
+            std::future::pending::<Result<()>>,
             SLOW_DISK_OPERATION_THRESHOLD + Duration::from_secs(1),
         );
         tokio::pin!(timed);

crates/ecstore/src/io_support/rio.rs

@@ -442,12 +442,19 @@ impl WritePlan {
         }
 
         if let Some(encryption) = self.encryption {
+            let etag_nonce = match encryption.mode {
+                WriteEncryptionMode::Singlepart { base_nonce } | WriteEncryptionMode::MultipartLegacy { base_nonce, .. } => {
+                    base_nonce
+                }
+                WriteEncryptionMode::SinglepartObjectKey | WriteEncryptionMode::MultipartObjectKey { .. } => [0u8; 12],
+            };
+            let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?;
             reader = match encryption.mode {
                 WriteEncryptionMode::SinglepartObjectKey => HashReader::from_reader(
                     #[cfg(feature = "rio-v2")]
-                    EncryptReader::new_with_object_key(reader, encryption.key_bytes),
+                    EncryptReader::new_with_object_key(protected_reader, encryption.key_bytes),
                     #[cfg(not(feature = "rio-v2"))]
-                    EncryptReader::new(reader, encryption.key_bytes, [0u8; 12]),
+                    EncryptReader::new(protected_reader, encryption.key_bytes, [0u8; 12]),
                     HashReader::SIZE_PRESERVE_LAYER,
                     actual_size,
                     None,
@@ -457,12 +464,12 @@ impl WritePlan {
                 WriteEncryptionMode::Singlepart { base_nonce } => {
                     #[cfg(not(feature = "rio-v2"))]
                     let encrypt_reader = if encryption_frame_v2_enabled() {
-                        EncryptReader::new_v2(reader, encryption.key_bytes, base_nonce)
+                        EncryptReader::new_v2(protected_reader, encryption.key_bytes, base_nonce)
                     } else {
-                        EncryptReader::new(reader, encryption.key_bytes, base_nonce)
+                        EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce)
                     };
                     #[cfg(feature = "rio-v2")]
-                    let encrypt_reader = EncryptReader::new(reader, encryption.key_bytes, base_nonce);
+                    let encrypt_reader = EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce);
                     HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
                 }
                 WriteEncryptionMode::MultipartLegacy {
@@ -471,20 +478,25 @@ impl WritePlan {
                 } => {
                     #[cfg(not(feature = "rio-v2"))]
                     let encrypt_reader = if encryption_frame_v2_enabled() {
-                        EncryptReader::new_multipart_v2(reader, encryption.key_bytes, base_nonce, multipart_part_number)
+                        EncryptReader::new_multipart_v2(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
                     } else {
-                        EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number)
+                        EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
                     };
                     #[cfg(feature = "rio-v2")]
                     let encrypt_reader =
-                        EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number);
+                        EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number);
                     HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
                 }
                 WriteEncryptionMode::MultipartObjectKey { multipart_part_number } => HashReader::from_reader(
                     #[cfg(feature = "rio-v2")]
-                    EncryptReader::new_multipart_with_object_key(reader, encryption.key_bytes, multipart_part_number),
+                    EncryptReader::new_multipart_with_object_key(protected_reader, encryption.key_bytes, multipart_part_number),
                     #[cfg(not(feature = "rio-v2"))]
-                    EncryptReader::new_multipart(reader, encryption.key_bytes, [0u8; 12], multipart_part_number as usize),
+                    EncryptReader::new_multipart(
+                        protected_reader,
+                        encryption.key_bytes,
+                        [0u8; 12],
+                        multipart_part_number as usize,
+                    ),
                     HashReader::SIZE_PRESERVE_LAYER,
                     actual_size,
                     None,
@@ -497,6 +509,7 @@ impl WritePlan {
         // `ignore_value` deliberately avoids a second hasher over compressed or
         // encrypted bytes. The inner reader still validates the plaintext request
         // checksum while this outer reader exposes the request checksum context.
+        reader.encrypted_etag = encrypted;
         reader.add_non_trailing_checksum(checksum, true)?;
         reader.set_trailer(trailer);
 
@@ -513,6 +526,36 @@ mod tests {
     use std::io::Cursor;
     use tokio::io::AsyncReadExt;
 
+    #[tokio::test]
+    async fn write_plan_encryption_persists_protected_etags_for_every_mode() {
+        let data = b"Salary 2026: 85000 EUR";
+        let size = i64::try_from(data.len()).expect("test data size");
+        let mut control =
+            HashReader::from_stream(Cursor::new(data), size, size, None, None, false).expect("create control reader");
+        control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
+        let md5 = control.try_resolve_etag().expect("control MD5");
+        for encryption in [
+            WriteEncryption::singlepart([1; 32], [1; 12]),
+            WriteEncryption::singlepart_object_key([1; 32]),
+            WriteEncryption::multipart([1; 32], [1; 12], 1),
+            WriteEncryption::multipart_object_key([1; 32], 1),
+        ] {
+            let plaintext = HashReader::from_stream(Cursor::new(data), size, size, Some(md5.clone()), None, false)
+                .expect("create checked reader");
+            let mut reader = WritePlan::new()
+                .with_encryption(encryption)
+                .apply(plaintext, size)
+                .expect("apply encryption");
+            let mut ciphertext = Vec::new();
+            reader.read_to_end(&mut ciphertext).await.expect("stream ciphertext");
+            assert_ne!(ciphertext, data);
+            let tag = reader.try_resolve_etag().expect("resolve persisted encrypted ETag");
+            assert_ne!(tag, md5, "stored ETag must not reveal the plaintext MD5");
+            assert_eq!(tag.len(), 32, "multipart ETags require a 16-byte hexadecimal part tag");
+            assert_eq!(reader.try_resolve_etag().as_ref(), Some(&tag));
+        }
+    }
+
     async fn assert_non_trailing_checksum_survives(plan: WritePlan) {
         let plaintext = b"checksum-context-through-write-plan".repeat(256);
         let actual_size = plaintext.len() as i64;

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

@@ -3004,6 +3004,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         // etag
         let etag = resolve_complete_etag(opts, &uploaded_parts);
 
+        // An upload can contain parts from an older writer or a peer. Until
+        // every part's format is attested, keep its ETag out of keyless lists.
+        rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG);
+
         fi.metadata.insert("etag".to_owned(), etag);
 
         let persist_encryption_original_size = should_persist_encryption_original_size(&fi.metadata);

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

@@ -3847,6 +3847,16 @@ impl SetDisks {
                 .map(crate::io_support::rio::compression_index_storage_bytes);
 
             let mut etag = data.stream.try_resolve_etag().unwrap_or_default();
+            if data.stream.encrypted_etag {
+                if opts.preserve_etag.as_ref().is_none_or(|preserved| preserved == &etag) {
+                    insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+                } else {
+                    rustfs_utils::http::metadata_compat::remove_str(
+                        &mut user_defined,
+                        rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+                    );
+                }
+            }
             if let Some(ref tag) = opts.preserve_etag {
                 etag = tag.clone();
             }
@@ -10610,6 +10620,65 @@ mod object_encryption_resolver_wiring_tests {
     use std::sync::Mutex;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
+    #[tokio::test]
+    async fn encrypted_put_attests_only_etags_produced_by_its_encryption_reader() {
+        use super::hermetic_set_disks_support::hermetic_set_disks_isolated;
+        use crate::io_support::rio::{WriteEncryption, WritePlan};
+        use crate::storage_api_contracts::object::ObjectIO as _;
+        use tokio::io::AsyncReadExt;
+
+        let (_temp_dirs, disks, set_disks) = hermetic_set_disks_isolated(4).await;
+        let bucket = "encrypted-etag-proof";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("create bucket volume");
+        }
+        let plaintext = b"Salary 2026: 85000 EUR";
+        let size = plaintext.len() as i64;
+        let mut control = HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("control reader");
+        control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
+        let plaintext_md5 = control.try_resolve_etag().expect("control MD5");
+        for (object, preserved) in [("new", None), ("preserved", Some(plaintext_md5.clone()))] {
+            let reader =
+                HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("plaintext reader");
+            let reader = WritePlan::new()
+                .with_encryption(WriteEncryption::singlepart([1; 32], [2; 12]))
+                .apply(reader, size)
+                .expect("encrypted reader");
+            let mut opts = ObjectOptions {
+                preserve_etag: preserved.clone(),
+                user_defined: HashMap::from([
+                    ("x-amz-server-side-encryption-customer-algorithm".to_owned(), "AES256".to_owned()),
+                    (rustfs_utils::http::SSEC_ORIGINAL_SIZE_HEADER.to_owned(), size.to_string()),
+                ]),
+                ..Default::default()
+            };
+            // A copied producer marker cannot attest to a preserved plaintext tag.
+            insert_str(&mut opts.user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+            set_disks
+                .put_object(bucket, object, &mut PutObjReader::new(reader), &opts)
+                .await
+                .expect("store encrypted bytes");
+            let stored = set_disks
+                .get_object_info(bucket, object, &ObjectOptions::default())
+                .await
+                .expect("read persisted metadata");
+            if let Some(preserved) = preserved {
+                assert_eq!(stored.etag.as_deref(), Some(preserved.as_str()));
+                assert!(!rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
+            } else {
+                assert_ne!(stored.etag.as_ref(), Some(&plaintext_md5));
+                assert_eq!(
+                    rustfs_utils::http::get_consistent_str(
+                        &stored.user_defined,
+                        rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG
+                    ),
+                    Some("v1")
+                );
+                assert!(rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
+            }
+        }
+    }
+
     #[derive(Clone, Default)]
     struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
 

crates/ecstore/src/store/bucket.rs

@@ -901,6 +901,7 @@ impl ECStore {
             }
         };
         let confirmed_missing = existing_bucket_info.is_none();
+        let mut pending_creation_metadata = None;
         if confirmed_missing
             && !opts.no_lock
             && !is_meta_bucketname(bucket)
@@ -921,13 +922,25 @@ impl ECStore {
                     ),
                     other => other,
                 })?;
-            return Err(StorageError::InvalidArgument(
-                "CreateBucket".to_string(),
-                "bucket".to_string(),
-                format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
-            ));
+            let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
+            let metadata_store = metadata_sys::object_store_in(&self.ctx).await?;
+            let pending_intent = persisted
+                && metadata.bucket_incarnation_id == old_incarnation
+                && metadata.needs_bucket_creation_commit()
+                && !crate::bucket::retirement::is_retired(metadata_store, bucket, old_incarnation).await?;
+            if pending_intent {
+                pending_creation_metadata = Some(metadata);
+            } else {
+                return Err(StorageError::InvalidArgument(
+                    "CreateBucket".to_string(),
+                    "bucket".to_string(),
+                    format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
+                ));
+            }
         }
-        let existing_metadata = if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
+        let existing_metadata = if let Some(metadata) = pending_creation_metadata {
+            Some(metadata)
+        } else if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
             let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
             if !persisted {
                 metadata = BucketMetadata::new(bucket);
@@ -1347,6 +1360,23 @@ impl ECStore {
         } else {
             None
         };
+        // Match metadata migration: lifecycle -> metadata transaction ->
+        // bucket namespace. Holding the transaction while waiting for the
+        // namespace prevents a migration writer and force delete from
+        // acquiring these two locks in opposite orders.
+        let object_lock_metadata_transaction = if opts.force && !is_meta_bucketname(bucket) {
+            Some(
+                await_bucket_namespace_operation(
+                    bucket_lifecycle_guard.as_ref(),
+                    bucket,
+                    "Object Lock metadata transaction acquisition",
+                    crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket),
+                )
+                .await?,
+            )
+        } else {
+            None
+        };
         let ns_guard = if !opts.no_lock {
             let ns_lock = self.new_ns_lock(bucket, bucket).await?;
             Some(
@@ -1402,6 +1432,29 @@ impl ECStore {
             }
         }
 
+        // Keep configuration changes fenced until data deletion finishes;
+        // a force delete cannot bypass any Object Lock protection.
+        let object_lock_metadata_guard = if bucket_exists && let Some(guard) = object_lock_metadata_transaction {
+            let (state, incarnation, _) =
+                crate::bucket::metadata_sys::get_object_lock_config_and_incarnation_from_disk_with_guard_in(
+                    &self.ctx,
+                    bucket,
+                    Some(ns_guard.as_ref().unwrap_or(&guard)),
+                )
+                .await?;
+            match state {
+                crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent => Some((guard, incarnation)),
+                crate::bucket::metadata_sys::ObjectLockConfigState::Configured { .. } => {
+                    return Err(StorageError::PrefixAccessDenied(bucket.to_owned(), String::new()));
+                }
+                crate::bucket::metadata_sys::ObjectLockConfigState::Fabricated => {
+                    return Err(StorageError::FileCorrupt);
+                }
+            }
+        } else {
+            None
+        };
+
         if sr_delete && !bucket_exists {
             delete_opts.force_if_empty = true;
         }
@@ -1415,7 +1468,12 @@ impl ECStore {
                 ns_guard.as_ref(),
                 bucket,
                 "bucket delete marker creation",
-                self.mark_bucket_deleted(bucket),
+                await_bucket_namespace_operation(
+                    object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
+                    bucket,
+                    "Object Lock configuration fence during bucket retirement",
+                    self.mark_bucket_deleted(bucket),
+                ),
             )
             .await?;
         }
@@ -1424,9 +1482,13 @@ impl ECStore {
         // Legacy buckets without a stamp cannot produce retirement authority.
         let retirement = if bucket_exists && bucket_lifecycle_guard.is_some() && !is_meta_bucketname(bucket) {
             if let Some(store) = metadata_sys::object_store_if_initialized_in(&self.ctx).await {
-                crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
-                    .await?
-                    .map(|incarnation| (store, incarnation))
+                if let Some((_, incarnation)) = object_lock_metadata_guard.as_ref() {
+                    Some((store, *incarnation))
+                } else {
+                    crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
+                        .await?
+                        .map(|incarnation| (store, incarnation))
+                }
             } else {
                 None
             }
@@ -1438,13 +1500,19 @@ impl ECStore {
             bucket_lifecycle_guard.as_ref(),
             bucket,
             "physical bucket deletion",
-            run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
-                self.delete_bucket_on_sets(bucket, &delete_opts)
-                    .await
-                    .map_err(|err| to_object_err(err, vec![bucket]))
-            }),
+            await_bucket_namespace_operation(
+                object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
+                bucket,
+                "Object Lock configuration fence during bucket deletion",
+                run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
+                    self.delete_bucket_on_sets(bucket, &delete_opts)
+                        .await
+                        .map_err(|err| to_object_err(err, vec![bucket]))
+                }),
+            ),
         )
         .await;
+        drop(object_lock_metadata_guard);
         if let Err(err) = delete_result
             && (!sr_delete || !is_err_strict_volume_not_found(&err))
         {
@@ -3689,6 +3757,116 @@ mod tests {
         assert!(!any_disk_path_exists(&disk_paths, bucket_deleted_marker_volume(&bucket)).await);
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn force_bucket_delete_preserves_object_lock_versions() {
+        let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], Some(2)).await;
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("force-delete-worm-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    versioning_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create Object Lock bucket");
+        let until = (OffsetDateTime::now_utc() + time::Duration::days(365))
+            .format(&time::format_description::well_known::Rfc3339)
+            .expect("format future retention date");
+        let written = store
+            .put_object(
+                &bucket,
+                "retained",
+                &mut PutObjReader::from_vec(b"retained bytes".to_vec()),
+                &ObjectOptions {
+                    versioned: true,
+                    user_defined: HashMap::from([
+                        ("x-amz-object-lock-mode".to_owned(), "COMPLIANCE".to_owned()),
+                        ("x-amz-object-lock-retain-until-date".to_owned(), until),
+                    ]),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("write retained object");
+        for operation in [SRBucketDeleteOp::NoOp, SRBucketDeleteOp::MarkDelete, SRBucketDeleteOp::Purge] {
+            let err = store
+                .delete_bucket(
+                    &bucket,
+                    &DeleteBucketOptions {
+                        force: true,
+                        srdelete_op: operation,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("force deletion must not bypass Object Lock");
+            assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected denial: {err}");
+            let info = store
+                .get_object_info(
+                    &bucket,
+                    "retained",
+                    &ObjectOptions {
+                        version_id: written.version_id.map(|version| version.to_string()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("retained version must survive the rejected bucket deletion");
+            assert_eq!(info.etag, written.etag);
+            assert_eq!(info.size, written.size);
+            assert_eq!(info.version_id, written.version_id);
+        }
+        metadata_sys::inject_object_lock_disk_read_error_in(&store.ctx, &bucket)
+            .await
+            .expect("inject authoritative metadata read failure");
+        let error = store
+            .delete_bucket(
+                &bucket,
+                &DeleteBucketOptions {
+                    force: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("an unreadable Object Lock configuration must fail closed");
+        assert!(error.to_string().contains("injected Object Lock metadata disk read failure"));
+        let info = store
+            .get_object_info(&bucket, "retained", &ObjectOptions::default())
+            .await
+            .expect("metadata read failure must leave the retained object intact");
+        assert_eq!(info.version_id, written.version_id);
+
+        let unlocked = format!("force-delete-unlocked-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&unlocked, &MakeBucketOptions::default())
+            .await
+            .expect("create unlocked control bucket");
+        store
+            .put_object(
+                &unlocked,
+                "ordinary",
+                &mut PutObjReader::from_vec(b"ordinary bytes".to_vec()),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("write ordinary control object");
+        store
+            .delete_bucket(
+                &unlocked,
+                &DeleteBucketOptions {
+                    force: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("force deletion must still work for an unlocked bucket");
+    }
+
     #[tokio::test]
     #[serial]
     async fn bucket_delete_mark_delete_rejects_hidden_object_paths_without_force() {

crates/ecstore/src/store/init.rs

@@ -18318,7 +18318,7 @@ mod tests {
         let bucket = format!("lock-create-intent-retry-{}", uuid::Uuid::new_v4());
         let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket);
         intent.lock_enabled = true;
-        crate::bucket::metadata_sys::set_new_bucket_metadata_in(&ctx, intent)
+        crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, intent)
             .await
             .expect("persist pre-visibility lock intent");
         assert!(

crates/ecstore/src/store/list_objects.rs

@@ -448,8 +448,8 @@ const MAX_LIST_OBJECTS_METADATA_FAST_STALENESS_MS: u64 = 60_000;
 const LIST_OBJECTS_INDEX_PROVIDER_WALKER_KEY_ONLY: &str = "walker_key_only";
 const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY: &str = "persistent_key_only";
 const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY_DEFAULT_GENERATION: &str = "persistent-key-only";
-const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 2;
-const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v2";
+const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 3;
+const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v3";
 const PERSISTENT_KEY_ONLY_INDEX_BUCKET_HEADER: &str = "# bucket=";
 const PERSISTENT_KEY_ONLY_INDEX_GENERATION_HEADER: &str = "# generation=";
 const PERSISTENT_KEY_ONLY_INDEX_CHECKPOINT_HEADER: &str = "# checkpoint_high_water_mark=";
@@ -656,7 +656,10 @@ impl PersistentListMetadataObject {
             name: object.name.clone(),
             size: object.size,
             mod_time: object.mod_time,
-            etag: object.etag.clone(),
+            etag: object
+                .etag
+                .clone()
+                .filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined)),
             storage_class: object.storage_class.clone(),
         }
     }
@@ -2361,24 +2364,43 @@ fn list_objects_paginate(
     (objects, prefixes, is_truncated, next_marker, next_version_idmarker)
 }
 
-fn list_objects_paginate_versions<'a>(
-    mut get_objects: Vec<ObjectInfo>,
-    delimiter: &Option<String>,
+struct VersionedListPagination<'a, I> {
+    delimiter: &'a Option<String>,
     max_keys: i32,
     disk_has_more: bool,
-    cache_id: Option<&str>,
-    marker: Option<&str>,
-    last_scanned_key: Option<&str>,
-    raw_keys: impl Iterator<Item = &'a str>,
-) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>) {
-    filter_versions_common_prefixes_after_marker(&mut get_objects, delimiter.as_deref(), marker);
-    let prefix_at_page_boundary =
-        versioned_prefix_page_boundary_key(&get_objects, raw_keys, delimiter.as_deref(), max_keys, disk_has_more);
+    cache_id: Option<&'a str>,
+    marker: Option<&'a str>,
+    last_scanned_key: Option<&'a str>,
+    raw_keys: I,
+}
 
-    let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) =
-        list_objects_paginate(get_objects, delimiter, max_keys, disk_has_more, cache_id, true, last_scanned_key);
+fn list_objects_paginate_versions<'a, I>(
+    mut get_objects: Vec<ObjectInfo>,
+    pagination: VersionedListPagination<'a, I>,
+) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>)
+where
+    I: Iterator<Item = &'a str>,
+{
+    filter_versions_common_prefixes_after_marker(&mut get_objects, pagination.delimiter.as_deref(), pagination.marker);
+    let prefix_at_page_boundary = versioned_prefix_page_boundary_key(
+        &get_objects,
+        pagination.raw_keys,
+        pagination.delimiter.as_deref(),
+        pagination.max_keys,
+        pagination.disk_has_more,
+    );
+
+    let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) = list_objects_paginate(
+        get_objects,
+        pagination.delimiter,
+        pagination.max_keys,
+        pagination.disk_has_more,
+        pagination.cache_id,
+        true,
+        pagination.last_scanned_key,
+    );
     if is_truncated && let Some(raw_key) = prefix_at_page_boundary {
-        next_marker = Some(append_list_cache_id_to_marker(raw_key, cache_id));
+        next_marker = Some(append_list_cache_id_to_marker(raw_key, pagination.cache_id));
         next_version_idmarker = None;
     }
 
@@ -4326,13 +4348,15 @@ impl ECStore {
 
         let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
             get_objects,
-            &delimiter,
-            max_keys,
-            disk_has_more,
-            next_cache_id.as_deref(),
-            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
         );
 
         Ok(ListObjectVersionsInfo {
@@ -5754,13 +5778,15 @@ impl Sets {
 
         let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
             get_objects,
-            &delimiter,
-            max_keys,
-            disk_has_more,
-            next_cache_id.as_deref(),
-            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
         );
 
         Ok(ListObjectVersionsInfo {
@@ -6568,13 +6594,15 @@ impl SetDisks {
 
         let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
             get_objects,
-            &delimiter,
-            max_keys,
-            disk_has_more,
-            next_cache_id.as_deref(),
-            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
         );
 
         Ok(ListObjectVersionsInfo {
@@ -8848,13 +8876,15 @@ mod test {
             let get_objects = fold_delimiter_page(&window, "", "-");
             let (objects, prefixes, is_truncated, next_marker, _v) = super::list_objects_paginate_versions(
                 get_objects,
-                &delimiter,
-                max_keys,
-                disk_has_more,
-                None,
-                marker.as_deref(),
-                last_scanned.as_deref(),
-                window.iter().map(String::as_str),
+                super::VersionedListPagination {
+                    delimiter: &delimiter,
+                    max_keys,
+                    disk_has_more,
+                    cache_id: None,
+                    marker: marker.as_deref(),
+                    last_scanned_key: last_scanned.as_deref(),
+                    raw_keys: window.iter().map(String::as_str),
+                },
             );
 
             assert!(
@@ -8927,13 +8957,15 @@ mod test {
             let get_objects = fold_delimiter_page(&window, "", "-");
             let (objects, prefixes, is_truncated, next_marker, _) = super::list_objects_paginate_versions(
                 get_objects,
-                &delimiter,
-                max_keys,
-                disk_has_more,
-                None,
-                marker.as_deref(),
-                last_scanned.as_deref(),
-                window.iter().map(String::as_str),
+                super::VersionedListPagination {
+                    delimiter: &delimiter,
+                    max_keys,
+                    disk_has_more,
+                    cache_id: None,
+                    marker: marker.as_deref(),
+                    last_scanned_key: last_scanned.as_deref(),
+                    raw_keys: window.iter().map(String::as_str),
+                },
             );
 
             assert_eq!(objects.len() + prefixes.len(), 1, "a page must respect max-keys");
@@ -9176,6 +9208,38 @@ mod test {
         assert_eq!(parsed, object);
     }
 
+    #[test]
+    fn persistent_listing_snapshots_cannot_restore_legacy_encrypted_fingerprints() {
+        let mut object = ObjectInfo {
+            name: "encrypted".to_owned(),
+            etag: Some("fbfcbcc6d2f035411e6268c8b119593e".to_owned()),
+            user_defined: Arc::new(HashMap::from([(
+                "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+                "AES256".to_owned(),
+            )])),
+            ..Default::default()
+        };
+        let snapshot = PersistentListMetadataObject::from_object_info(&object);
+        assert!(snapshot.to_object_info("bucket").etag.is_none());
+        let mut old_snapshot = snapshot;
+        old_snapshot.etag = object.etag.clone();
+        let old_contents = format!(
+            "# rustfs-listobjects-key-only-v2\n{}\n",
+            encode_persistent_list_metadata_object(&old_snapshot)
+        );
+        let old_index = parse_persistent_key_only_index(&old_contents);
+        assert_ne!(old_index.format_version, PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION);
+
+        rustfs_utils::http::metadata_compat::insert_str(
+            Arc::make_mut(&mut object.user_defined),
+            rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+            "v1".to_owned(),
+        );
+        assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
+        object.user_defined = Arc::new(HashMap::new());
+        assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
+    }
+
     #[test]
     fn persistent_key_only_index_parses_legacy_keys_and_metadata_snapshots() {
         let object = PersistentListMetadataObject {

crates/ecstore/src/store/object.rs

@@ -5351,13 +5351,41 @@ impl ECStore {
         tier_journal_api: Option<Arc<ECStore>>,
     ) -> (Vec<DeletedObject>, Vec<Option<Error>>, Vec<Option<DeleteAccounting>>) {
         Box::pin(async move {
+            let count = objects.len();
+            let mut deleted = vec![DeletedObject::default(); count];
+            let mut errors: Vec<Option<Error>> = (0..count).map(|_| None).collect();
+            let mut accounting = vec![None; count];
+            let mut valid_objects = Vec::with_capacity(count);
+            let mut valid_indices = Vec::with_capacity(count);
+            // Validate the literal key before encoding it or taking locks. A
+            // dot-segment key must never reach filesystem path resolution.
+            for (index, object) in objects.into_iter().enumerate() {
+                match check_del_obj_args(bucket, &object.object_name) {
+                    Ok(()) => {
+                        valid_indices.push(index);
+                        valid_objects.push(object);
+                    }
+                    Err(err) => errors[index] = Some(err),
+                }
+            }
+            if valid_objects.is_empty() {
+                return (deleted, errors, accounting);
+            }
             let mut opts = opts;
             let receipt_sink = install_tier_free_version_receipt_sink(&mut opts);
-            let result = self
-                .handle_delete_objects_with_journal_and_accounting_inner(bucket, objects, opts, tier_journal_api)
+            let (valid_deleted, valid_errors, valid_accounting) = self
+                .handle_delete_objects_with_journal_and_accounting_inner(bucket, valid_objects, opts, tier_journal_api)
                 .await;
             enqueue_recorded_tier_free_versions(self, receipt_sink).await;
-            result
+            for (index, ((object, error), receipt)) in valid_indices
+                .into_iter()
+                .zip(valid_deleted.into_iter().zip(valid_errors).zip(valid_accounting))
+            {
+                deleted[index] = object;
+                errors[index] = error;
+                accounting[index] = receipt;
+            }
+            (deleted, errors, accounting)
         })
         .await
     }
@@ -9094,6 +9122,58 @@ mod tests {
         (dirs, store)
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn batch_delete_rejects_dot_segments_and_keeps_valid_result_positions() {
+        let bucket = "batch-delete-key-validation";
+        let (_dirs, store) = multipool_version_test_store(bucket).await;
+        let options = ObjectOptions {
+            versioned: true,
+            ..Default::default()
+        };
+        let victim = store
+            .put_object(bucket, "victim/object", &mut PutObjReader::from_vec(b"victim".to_vec()), &options)
+            .await
+            .expect("write protected-prefix control object");
+        store
+            .put_object(bucket, "allowed/object", &mut PutObjReader::from_vec(b"allowed".to_vec()), &options)
+            .await
+            .expect("write valid deletion control object");
+        let keys = [
+            "allowed/../victim/object",
+            "allowed/object",
+            "a/./b",
+            "../other-bucket/object",
+        ];
+        let objects = keys
+            .iter()
+            .map(|key| ObjectToDelete {
+                object_name: (*key).to_owned(),
+                ..Default::default()
+            })
+            .collect();
+        let (deleted, errors) = store.handle_delete_objects(bucket, objects, options).await;
+        assert_eq!(deleted.len(), keys.len());
+        assert_eq!(errors.len(), keys.len());
+        for index in [0, 2, 3] {
+            assert!(
+                matches!(&errors[index], Some(StorageError::ObjectNameInvalid(_, _))),
+                "invalid key {}: {:?}",
+                keys[index],
+                errors[index]
+            );
+            assert!(!deleted[index].delete_marker);
+        }
+        assert!(errors[1].is_none(), "valid deletion must still succeed: {:?}", errors[1]);
+        assert!(deleted[1].delete_marker);
+        assert_eq!(deleted[1].object_name, "allowed/object");
+        let after = store
+            .get_object_info(bucket, "victim/object", &ObjectOptions::default())
+            .await
+            .expect("the resolved victim path must remain visible");
+        assert_eq!(after.version_id, victim.version_id);
+        assert_eq!(after.etag, victim.etag);
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn retired_marker_converges_after_bucket_recreation_on_sixteen_disks() {
         let bucket = "retired-marker-c11";

crates/heal/src/heal/mrf_queue/partial_write.rs

@@ -392,7 +392,7 @@ impl PartialWrites {
                     object: anchor.object.clone(),
                     version_id: anchor.version_id,
                     scope: anchor.scope,
-                    delete_marker_purge: anchor.delete_marker_purge.clone(),
+                    delete_marker_purge: anchor.delete_marker_purge,
                     lease: Some(anchor.lease),
                     bucket_incarnation_id: anchor.bucket_incarnation_id,
                     disposition: MrfVerifiedRepairDisposition::AuthoritativelyAbsent,

crates/heal/tests/heal_concurrent_delete_test.rs

@@ -563,9 +563,31 @@ fn admin_root_completes_after_unversioned_delete() {
     run_delete_case(DeleteCase::Unversioned);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[test]
 #[serial]
-async fn admin_root_rejects_recreated_bucket_before_successor_repair() {
+fn admin_root_rejects_recreated_bucket_before_successor_repair() {
+    // Incarnation-bound repairs use a spawned owner; retain the server's stack
+    // budget for the whole scenario. block_on alone would poll that future on
+    // libtest's smaller calling-thread stack.
+    let stack_size = 8 * 1024 * 1024;
+    std::thread::Builder::new()
+        .name("heal-concurrent-delete".to_owned())
+        .stack_size(stack_size)
+        .spawn(move || {
+            let runtime = tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(4)
+                .thread_stack_size(stack_size)
+                .enable_all()
+                .build()
+                .expect("concurrent delete runtime");
+            runtime.block_on(recreated_bucket_repair());
+        })
+        .expect("concurrent delete scenario thread starts")
+        .join()
+        .expect("concurrent delete scenario must not panic");
+}
+
+async fn recreated_bucket_repair() {
     let directory = tempfile::tempdir().expect("recreated bucket fixture");
     let env = TestECStoreEnv::builder().base_dir(directory.path()).build().await;
     env.make_bucket(BUCKET, false).await;

crates/iam/Cargo.toml

@@ -30,6 +30,7 @@ workspace = true
 
 [features]
 default = []
+test-util = []
 hotpath = [
     "hotpath/hotpath",
     "hotpath/tokio",

crates/iam/src/federation/mod.rs

@@ -18,7 +18,6 @@ mod mapper;
 mod model;
 pub mod oidc;
 mod provider;
-mod registry;
 mod transaction;
 
 pub use binding::FederatedSessionBinding;
@@ -27,10 +26,9 @@ pub use mapper::{CoreFederatedAuthorizationMapper, FederatedAuthorizationRules};
 pub(crate) use mapper::{FederatedAuthorizationRule, FederatedAuthorizationRuleRef};
 pub(crate) use model::OpaqueLogoutContinuation;
 pub use model::{
-    FederatedAuthorization, FederatedClaims, FederatedCodeExchange, FederatedLoginSession, FederatedProviderRef,
-    FederatedSession, FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM, VerifiedFederatedIdentity,
+    FederatedAuthorization, FederatedClaims, FederatedLoginSession, FederatedProviderRef, FederatedSession,
+    FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM, VerifiedFederatedIdentity,
 };
-pub use provider::{FederatedIdentityProvider, FederatedProviderView, FederatedRedirectPolicy};
-pub(crate) use provider::{StandardOidcAuthentication, VerifiedFederatedCodeExchange};
-pub use registry::FederatedIdentityRegistry;
+pub(crate) use provider::{FederatedProviderQuery, StandardOidcAuthentication, VerifiedFederatedCodeExchange};
+pub use provider::{FederatedProviderView, FederatedRedirectPolicy};
 pub use transaction::FederatedIdentityService;

crates/iam/src/federation/model.rs

@@ -164,40 +164,6 @@ impl FederatedAuthorization {
     }
 }
 
-pub struct FederatedCodeExchange {
-    pub authorization: FederatedAuthorization,
-    pub redirect_after: Option<String>,
-    pub id_token: String,
-}
-
-impl FederatedCodeExchange {
-    pub(crate) fn into_parts(self) -> (FederatedAuthorization, Option<String>, OpaqueLogoutContinuation) {
-        let Self {
-            authorization,
-            redirect_after,
-            id_token,
-        } = self;
-        let continuation = OpaqueLogoutContinuation {
-            provider: FederatedProviderRef::new(authorization.provider_id.clone()),
-            id_token,
-        };
-        (authorization, redirect_after, continuation)
-    }
-}
-
-impl fmt::Debug for FederatedCodeExchange {
-    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        formatter
-            .debug_struct("FederatedCodeExchange")
-            .field("provider_id", &self.authorization.provider_id)
-            .field("policy_count", &self.authorization.policies.len())
-            .field("group_count", &self.authorization.groups.len())
-            .field("redirect_after_present", &self.redirect_after.is_some())
-            .field("id_token_present", &!self.id_token.is_empty())
-            .finish()
-    }
-}
-
 /// Carries the provider-bound ID token between code exchange and logout token creation.
 pub(crate) struct OpaqueLogoutContinuation {
     provider: FederatedProviderRef,
@@ -397,27 +363,13 @@ mod tests {
 
     #[test]
     fn logout_continuation_binds_provider_and_redacts_token() {
-        let mut authorization = authorization(Vec::new(), Vec::new());
-        authorization
-            .claims
-            .raw
-            .insert("attribute_poison".to_string(), Value::String("secret-attribute".to_string()));
-        let exchange = FederatedCodeExchange {
-            authorization,
-            redirect_after: Some("/console?poison=redirect-secret".to_string()),
-            id_token: "secret-id-token".to_string(),
-        };
-        let debug = format!("{exchange:?}");
-        let (_, _, continuation) = exchange.into_parts();
+        let continuation =
+            OpaqueLogoutContinuation::new(FederatedProviderRef::new("standard_oidc".to_string()), "secret-id-token".to_string());
         let continuation_debug = format!("{continuation:?}");
         let (provider, id_token) = continuation.into_parts();
 
         assert_eq!(provider.as_str(), "standard_oidc");
         assert_eq!(id_token, "secret-id-token");
-        assert!(!debug.contains("secret-id-token"));
-        assert!(!debug.contains("secret-attribute"));
-        assert!(!debug.contains("redirect-secret"));
-        assert!(debug.contains("id_token_present: true"));
         assert!(!continuation_debug.contains("secret-id-token"));
         assert!(continuation_debug.contains("[REDACTED]"));
     }

crates/iam/src/federation/oidc/flow.rs

@@ -16,10 +16,9 @@ use super::config::OidcConfigQuery;
 use super::{claims, config, discovery, http};
 use crate::{
     federation::{
-        CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedAuthorizationRules, FederatedCodeExchange,
-        FederatedIdentityProvider, FederatedIdentityService, FederatedProviderRef, FederatedProviderView,
-        FederatedRedirectPolicy, FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
-        VerifiedFederatedCodeExchange,
+        CoreFederatedAuthorizationMapper, FederatedAuthorizationRules, FederatedIdentityService, FederatedProviderQuery,
+        FederatedProviderRef, FederatedProviderView, FederatedRedirectPolicy, FederationError, OpaqueLogoutContinuation, Result,
+        StandardOidcAuthentication, VerifiedFederatedCodeExchange,
     },
     oidc::{OidcConfigSnapshot, OidcSys},
 };
@@ -45,8 +44,8 @@ impl StandardOidcAdapter {
 
     pub fn into_service(self: Arc<Self>, mapper: CoreFederatedAuthorizationMapper) -> FederatedIdentityService {
         let authentication: Arc<dyn StandardOidcAuthentication> = self.clone();
-        let provider: Arc<dyn FederatedIdentityProvider> = self;
-        FederatedIdentityService::from_standard_oidc_parts(provider, authentication, mapper)
+        let provider_query: Arc<dyn FederatedProviderQuery> = self;
+        FederatedIdentityService::from_standard_oidc_parts(provider_query, authentication, mapper)
     }
 }
 
@@ -58,6 +57,10 @@ impl OidcConfigQuery for StandardOidcAdapter {
 
 #[async_trait::async_trait]
 impl StandardOidcAuthentication for StandardOidcAdapter {
+    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
+        http::authorize_url(&self.oidc, provider_id, redirect_uri, redirect_after).await
+    }
+
     async fn exchange_identity(&self, state: &str, code: &str, redirect_uri: &str) -> Result<VerifiedFederatedCodeExchange> {
         let (oidc_claims, provider_id, session, id_token) = http::exchange_code(&self.oidc, state, code, redirect_uri).await?;
         let provider = FederatedProviderRef::new(provider_id);
@@ -81,10 +84,13 @@ impl StandardOidcAuthentication for StandardOidcAdapter {
         let (provider, id_token) = continuation.into_parts();
         http::create_logout_token(&self.oidc, provider.as_str(), &id_token).await
     }
+
+    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
+        http::build_logout_url(&self.oidc, logout_token, post_logout_redirect_uri).await
+    }
 }
 
-#[async_trait::async_trait]
-impl FederatedIdentityProvider for StandardOidcAdapter {
+impl FederatedProviderQuery for StandardOidcAdapter {
     fn has_providers(&self) -> bool {
         self.oidc.has_providers()
     }
@@ -97,40 +103,7 @@ impl FederatedIdentityProvider for StandardOidcAdapter {
         discovery::list_visible_providers(&self.oidc)
     }
 
-    fn provider_config(&self, provider_id: &str) -> Option<&crate::oidc::OidcProviderConfig> {
-        self.oidc.get_provider_config(provider_id)
-    }
-
     fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
         config::redirect_policy(&self.oidc, provider_id)
     }
-
-    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
-        http::authorize_url(&self.oidc, provider_id, redirect_uri, redirect_after).await
-    }
-
-    async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<FederatedCodeExchange> {
-        let exchange = StandardOidcAuthentication::exchange_identity(self, state, code, redirect_uri).await?;
-        let (identity, redirect_after, continuation) = exchange.into_parts();
-        let authorization = CoreFederatedAuthorizationMapper::new(self.authorization_rules()).map(identity);
-        let (_, id_token) = continuation.into_parts();
-        Ok(FederatedCodeExchange {
-            authorization,
-            redirect_after,
-            id_token,
-        })
-    }
-
-    async fn verify_web_identity_token(&self, jwt: &str) -> Result<FederatedAuthorization> {
-        let identity = StandardOidcAuthentication::verify_identity(self, jwt).await?;
-        Ok(CoreFederatedAuthorizationMapper::new(self.authorization_rules()).map(identity))
-    }
-
-    async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String> {
-        http::create_logout_token(&self.oidc, provider_id, id_token).await
-    }
-
-    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
-        http::build_logout_url(&self.oidc, logout_token, post_logout_redirect_uri).await
-    }
 }

crates/iam/src/federation/provider.rs

@@ -12,8 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::{FederatedAuthorization, FederatedCodeExchange, OpaqueLogoutContinuation, Result, VerifiedFederatedIdentity};
-use crate::oidc::OidcProviderConfig;
+use super::{OpaqueLogoutContinuation, Result, VerifiedFederatedIdentity};
 use serde::{Deserialize, Serialize};
 
 /// Provider data exposed to login discovery consumers.
@@ -56,41 +55,21 @@ impl VerifiedFederatedCodeExchange {
 
 #[async_trait::async_trait]
 pub(crate) trait StandardOidcAuthentication: Send + Sync {
+    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String>;
     async fn exchange_identity(&self, state: &str, code: &str, redirect_uri: &str) -> Result<VerifiedFederatedCodeExchange>;
     async fn verify_identity(&self, jwt: &str) -> Result<VerifiedFederatedIdentity>;
     async fn create_logout_token(&self, continuation: OpaqueLogoutContinuation) -> Result<String>;
+    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>>;
 }
 
-#[async_trait::async_trait]
-pub trait FederatedIdentityProvider: Send + Sync {
+pub(crate) trait FederatedProviderQuery: Send + Sync {
     fn has_providers(&self) -> bool;
 
     fn list_providers(&self) -> Vec<FederatedProviderView>;
 
     fn list_visible_providers(&self) -> Vec<FederatedProviderView>;
 
-    /// Compatibility accessor for existing provider implementations.
-    /// Login redirect consumers should use [`Self::redirect_policy`].
-    fn provider_config(&self, _provider_id: &str) -> Option<&OidcProviderConfig> {
-        None
-    }
-
-    fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
-        self.provider_config(provider_id).map(|config| FederatedRedirectPolicy {
-            redirect_uri: config.redirect_uri.clone(),
-            allow_request_origin: config.redirect_uri_dynamic,
-        })
-    }
-
-    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String>;
-
-    async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<FederatedCodeExchange>;
-
-    async fn verify_web_identity_token(&self, jwt: &str) -> Result<FederatedAuthorization>;
-
-    async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String>;
-
-    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>>;
+    fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy>;
 }
 
 #[cfg(test)]

crates/iam/src/federation/registry.rs

@@ -1,32 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use super::FederatedIdentityProvider;
-use std::sync::Arc;
-
-/// Compatibility facade for integrations built against the original provider API.
-/// Production standard OIDC startup composes the adapter, mapper, and service directly.
-pub struct FederatedIdentityRegistry {
-    standard_oidc: Arc<dyn FederatedIdentityProvider>,
-}
-
-impl FederatedIdentityRegistry {
-    pub fn new(standard_oidc: Arc<dyn FederatedIdentityProvider>) -> Self {
-        Self { standard_oidc }
-    }
-
-    pub(crate) fn standard_oidc_arc(&self) -> Arc<dyn FederatedIdentityProvider> {
-        Arc::clone(&self.standard_oidc)
-    }
-}

crates/iam/src/federation/transaction.rs

@@ -13,11 +13,10 @@
 // limitations under the License.
 
 use super::{
-    CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedIdentityProvider, FederatedIdentityRegistry,
-    FederatedLoginSession, FederatedProviderView, FederatedRedirectPolicy, FederatedSession, FederatedSessionBinding,
-    FederatedSessionTransaction, FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
+    CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedLoginSession, FederatedProviderQuery,
+    FederatedProviderView, FederatedRedirectPolicy, FederatedSession, FederatedSessionBinding, FederatedSessionTransaction,
+    FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
 };
-use crate::oidc::OidcProviderConfig;
 use std::sync::Arc;
 
 const DEFAULT_OIDC_PROVIDER_ID: &str = "default";
@@ -32,11 +31,7 @@ fn sorted_provider_views(mut providers: Vec<FederatedProviderView>) -> Vec<Feder
 }
 
 pub struct FederatedIdentityService {
-    provider: Arc<dyn FederatedIdentityProvider>,
-    standard_oidc: Option<StandardOidcRuntime>,
-}
-
-struct StandardOidcRuntime {
+    provider_query: Arc<dyn FederatedProviderQuery>,
     authentication: Arc<dyn StandardOidcAuthentication>,
     mapper: CoreFederatedAuthorizationMapper,
 }
@@ -49,91 +44,57 @@ struct RuntimeCodeExchange {
 
 impl FederatedIdentityService {
     async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<RuntimeCodeExchange> {
-        match &self.standard_oidc {
-            Some(runtime) => {
-                let exchange = runtime.authentication.exchange_identity(state, code, redirect_uri).await?;
-                let (identity, redirect_after, logout_continuation) = exchange.into_parts();
-                Ok(RuntimeCodeExchange {
-                    authorization: runtime.mapper.map(identity),
-                    redirect_after,
-                    logout_continuation,
-                })
-            }
-            None => {
-                let exchange = self.provider.exchange_code(state, code, redirect_uri).await?;
-                let (authorization, redirect_after, logout_continuation) = exchange.into_parts();
-                Ok(RuntimeCodeExchange {
-                    authorization,
-                    redirect_after,
-                    logout_continuation,
-                })
-            }
-        }
+        let exchange = self.authentication.exchange_identity(state, code, redirect_uri).await?;
+        let (identity, redirect_after, logout_continuation) = exchange.into_parts();
+        Ok(RuntimeCodeExchange {
+            authorization: self.mapper.map(identity),
+            redirect_after,
+            logout_continuation,
+        })
     }
 
     async fn verify_identity(&self, jwt: &str) -> Result<FederatedAuthorization> {
-        match &self.standard_oidc {
-            Some(runtime) => Ok(runtime.mapper.map(runtime.authentication.verify_identity(jwt).await?)),
-            None => self.provider.verify_web_identity_token(jwt).await,
-        }
+        Ok(self.mapper.map(self.authentication.verify_identity(jwt).await?))
     }
 
     async fn create_logout_token(&self, continuation: OpaqueLogoutContinuation) -> Result<String> {
-        match &self.standard_oidc {
-            Some(runtime) => runtime.authentication.create_logout_token(continuation).await,
-            None => {
-                let (provider, id_token) = continuation.into_parts();
-                self.provider.create_logout_token(provider.as_str(), &id_token).await
-            }
-        }
-    }
-
-    /// Build the compatibility service around the published provider interface.
-    pub fn new(registry: FederatedIdentityRegistry) -> Self {
-        Self {
-            provider: registry.standard_oidc_arc(),
-            standard_oidc: None,
-        }
+        self.authentication.create_logout_token(continuation).await
     }
 
     pub(super) fn from_standard_oidc_parts(
-        provider: Arc<dyn FederatedIdentityProvider>,
+        provider_query: Arc<dyn FederatedProviderQuery>,
         authentication: Arc<dyn StandardOidcAuthentication>,
         mapper: CoreFederatedAuthorizationMapper,
     ) -> Self {
         Self {
-            provider,
-            standard_oidc: Some(StandardOidcRuntime { authentication, mapper }),
+            provider_query,
+            authentication,
+            mapper,
         }
     }
 
     pub fn has_providers(&self) -> bool {
-        self.provider.has_providers()
+        self.provider_query.has_providers()
     }
 
     pub fn list_providers(&self) -> Vec<FederatedProviderView> {
-        let providers = self.provider.list_providers();
+        let providers = self.provider_query.list_providers();
         sorted_provider_views(providers)
     }
 
     pub fn list_visible_providers(&self) -> Vec<FederatedProviderView> {
-        let providers = self.provider.list_visible_providers();
+        let providers = self.provider_query.list_visible_providers();
         sorted_provider_views(providers)
     }
 
     pub fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
-        self.provider.redirect_policy(provider_id)
-    }
-
-    /// Compatibility accessor for existing federation consumers.
-    /// Login redirects use [`Self::redirect_policy`], while OIDC-specific
-    /// configuration consumers use the OIDC configuration query.
-    pub fn get_provider_config(&self, provider_id: &str) -> Option<&OidcProviderConfig> {
-        self.provider.provider_config(provider_id)
+        self.provider_query.redirect_policy(provider_id)
     }
 
     pub async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
-        self.provider.authorize_url(provider_id, redirect_uri, redirect_after).await
+        self.authentication
+            .authorize_url(provider_id, redirect_uri, redirect_after)
+            .await
     }
 
     pub async fn complete_authorization_code(
@@ -205,7 +166,9 @@ impl FederatedIdentityService {
     }
 
     pub async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
-        self.provider.build_logout_url(logout_token, post_logout_redirect_uri).await
+        self.authentication
+            .build_logout_url(logout_token, post_logout_redirect_uri)
+            .await
     }
 }
 
@@ -213,9 +176,8 @@ impl FederatedIdentityService {
 mod tests {
     use super::*;
     use crate::federation::{
-        FederatedAuthorization, FederatedAuthorizationRule, FederatedAuthorizationRules, FederatedClaims, FederatedCodeExchange,
-        FederatedIdentityProvider, FederatedProviderRef, FederatedSessionBindingError, VerifiedFederatedCodeExchange,
-        VerifiedFederatedIdentity,
+        FederatedAuthorizationRule, FederatedAuthorizationRules, FederatedClaims, FederatedProviderQuery, FederatedProviderRef,
+        FederatedSessionBindingError, VerifiedFederatedCodeExchange, VerifiedFederatedIdentity,
     };
     use rustfs_credentials::Credentials;
     use std::sync::{Arc, Mutex};
@@ -235,6 +197,8 @@ mod tests {
         web_provider_id: &'static str,
         failure: ProviderFailure,
         events: Arc<Mutex<Vec<&'static str>>>,
+        authorize_calls: Mutex<Vec<(String, String, Option<String>)>>,
+        logout_url_calls: Mutex<Vec<(String, String)>>,
         expected_logout: (&'static str, &'static str),
         listed_provider_ids: Vec<&'static str>,
         visible_provider_ids: Vec<&'static str>,
@@ -249,6 +213,8 @@ mod tests {
                 web_provider_id: DEFAULT_OIDC_PROVIDER_ID,
                 failure: ProviderFailure::None,
                 events,
+                authorize_calls: Mutex::new(Vec::new()),
+                logout_url_calls: Mutex::new(Vec::new()),
                 expected_logout: (DEFAULT_OIDC_PROVIDER_ID, "id-token"),
                 listed_provider_ids: Vec::new(),
                 visible_provider_ids: Vec::new(),
@@ -259,32 +225,6 @@ mod tests {
             self.events.lock().expect("event log should not be poisoned").push(event);
         }
 
-        fn authorization(&self, provider_id: &str) -> FederatedAuthorization {
-            FederatedAuthorization {
-                provider_id: provider_id.to_string(),
-                claims: FederatedClaims {
-                    sub: "subject".to_string(),
-                    email: String::new(),
-                    username: "user".to_string(),
-                    groups: vec!["source-group".to_string()],
-                    raw: Default::default(),
-                },
-                policies: if self.with_policy {
-                    vec!["readwrite".to_string()]
-                } else {
-                    Vec::new()
-                },
-                group_claim_policies: Vec::new(),
-                groups: if self.with_group {
-                    vec!["developers".to_string()]
-                } else {
-                    Vec::new()
-                },
-                roles_claim_key: None,
-                roles: Vec::new(),
-            }
-        }
-
         fn identity(&self, provider_id: &str) -> VerifiedFederatedIdentity {
             VerifiedFederatedIdentity::from_claims(
                 FederatedProviderRef::new(provider_id.to_string()),
@@ -313,8 +253,7 @@ mod tests {
             .collect()
     }
 
-    #[async_trait::async_trait]
-    impl FederatedIdentityProvider for TestProvider {
+    impl FederatedProviderQuery for TestProvider {
         fn has_providers(&self) -> bool {
             true
         }
@@ -333,52 +272,18 @@ mod tests {
                 allow_request_origin: true,
             })
         }
-
-        async fn authorize_url(
-            &self,
-            _provider_id: &str,
-            _redirect_uri: &str,
-            _redirect_after: Option<String>,
-        ) -> Result<String> {
-            Ok("https://identity.example/authorize".to_string())
-        }
-
-        async fn exchange_code(&self, _state: &str, _code: &str, _redirect_uri: &str) -> Result<FederatedCodeExchange> {
-            self.record("exchange");
-            if self.failure == ProviderFailure::Exchange {
-                return Err(FederationError::CodeExchange("exchange failed".to_string()));
-            }
-            Ok(FederatedCodeExchange {
-                authorization: self.authorization(self.browser_provider_id),
-                redirect_after: Some("/browser".to_string()),
-                id_token: "id-token".to_string(),
-            })
-        }
-
-        async fn verify_web_identity_token(&self, _jwt: &str) -> Result<FederatedAuthorization> {
-            self.record("verify");
-            if self.failure == ProviderFailure::Verification {
-                return Err(FederationError::TokenVerification("verification failed".to_string()));
-            }
-            Ok(self.authorization(self.web_provider_id))
-        }
-
-        async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String> {
-            self.record("logout");
-            assert_eq!((provider_id, id_token), self.expected_logout);
-            if self.failure == ProviderFailure::Logout {
-                return Err(FederationError::Logout("logout failed".to_string()));
-            }
-            Ok("logout-token".to_string())
-        }
-
-        async fn build_logout_url(&self, _logout_token: &str, _post_logout_redirect_uri: &str) -> Result<Option<String>> {
-            Ok(Some("https://identity.example/logout".to_string()))
-        }
     }
 
     #[async_trait::async_trait]
     impl StandardOidcAuthentication for TestProvider {
+        async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
+            self.authorize_calls
+                .lock()
+                .expect("authorize call log should not be poisoned")
+                .push((provider_id.to_string(), redirect_uri.to_string(), redirect_after));
+            Ok("https://identity.example/authorize".to_string())
+        }
+
         async fn exchange_identity(
             &self,
             _state: &str,
@@ -414,6 +319,14 @@ mod tests {
             }
             Ok("logout-token".to_string())
         }
+
+        async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
+            self.logout_url_calls
+                .lock()
+                .expect("logout URL call log should not be poisoned")
+                .push((logout_token.to_string(), post_logout_redirect_uri.to_string()));
+            Ok(Some("https://identity.example/logout".to_string()))
+        }
     }
 
     fn standard_service(provider: Arc<TestProvider>) -> FederatedIdentityService {
@@ -432,14 +345,12 @@ mod tests {
             )
         }));
         let authentication: Arc<dyn StandardOidcAuthentication> = provider.clone();
-        let provider: Arc<dyn FederatedIdentityProvider> = provider;
-        FederatedIdentityService {
-            provider,
-            standard_oidc: Some(StandardOidcRuntime {
-                authentication,
-                mapper: CoreFederatedAuthorizationMapper::new(rules),
-            }),
-        }
+        let provider_query: Arc<dyn FederatedProviderQuery> = provider;
+        FederatedIdentityService::from_standard_oidc_parts(
+            provider_query,
+            authentication,
+            CoreFederatedAuthorizationMapper::new(rules),
+        )
     }
 
     struct RecordingBinding {
@@ -492,7 +403,7 @@ mod tests {
         let mut provider = TestProvider::new(events);
         provider.listed_provider_ids = vec!["zeta", "hidden", DEFAULT_OIDC_PROVIDER_ID, "alpha"];
         provider.visible_provider_ids = vec!["zeta", DEFAULT_OIDC_PROVIDER_ID, "alpha"];
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
+        let service = standard_service(Arc::new(provider));
 
         assert_eq!(
             service
@@ -519,6 +430,48 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn authorization_and_logout_urls_preserve_service_arguments() {
+        let provider = Arc::new(TestProvider::new(Arc::new(Mutex::new(Vec::new()))));
+        let service = standard_service(provider.clone());
+
+        let authorize_url = service
+            .authorize_url(
+                "corp",
+                "https://console.example.com/oauth/callback",
+                Some("/browser/path?tab=objects".to_string()),
+            )
+            .await
+            .expect("authorization URL should be returned");
+        let logout_url = service
+            .build_logout_url("opaque-logout-token", "https://console.example.com/signed-out")
+            .await
+            .expect("logout URL should be returned");
+
+        assert_eq!(authorize_url, "https://identity.example/authorize");
+        assert_eq!(logout_url.as_deref(), Some("https://identity.example/logout"));
+        assert_eq!(
+            provider
+                .authorize_calls
+                .lock()
+                .expect("authorize call log should not be poisoned")
+                .as_slice(),
+            [(
+                "corp".to_string(),
+                "https://console.example.com/oauth/callback".to_string(),
+                Some("/browser/path?tab=objects".to_string()),
+            )]
+        );
+        assert_eq!(
+            provider
+                .logout_url_calls
+                .lock()
+                .expect("logout URL call log should not be poisoned")
+                .as_slice(),
+            [("opaque-logout-token".to_string(), "https://console.example.com/signed-out".to_string(),)]
+        );
+    }
+
     #[tokio::test]
     async fn callback_and_web_identity_preserve_provider_and_transaction_boundaries() {
         let events = Arc::new(Mutex::new(Vec::new()));
@@ -526,9 +479,8 @@ mod tests {
         provider.browser_provider_id = "corp";
         provider.web_provider_id = "partner";
         provider.expected_logout = ("corp", "id-token");
-        let provider = Arc::new(provider);
         let binding = Arc::new(RecordingBinding::new(events.clone()));
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
+        let service = standard_service(Arc::new(provider));
 
         let login = service
             .complete_authorization_code("state", "code", "https://console.example/callback", 3600, binding.as_ref())
@@ -604,76 +556,13 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn standard_runtime_checks_authorization_before_binding() {
+    async fn web_identity_without_policy_or_group_is_not_bound() {
         let events = Arc::new(Mutex::new(Vec::new()));
         let mut provider = TestProvider::new(events.clone());
         provider.with_policy = false;
         let binding = Arc::new(RecordingBinding::new(events.clone()));
         let service = standard_service(Arc::new(provider));
 
-        let error = service
-            .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
-            .await
-            .expect_err("standard web identity requires mapped authorization");
-
-        assert!(matches!(error, FederationError::NoAuthorizationContext));
-        assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), ["verify"]);
-    }
-
-    #[tokio::test]
-    async fn standard_runtime_preserves_browser_failure_order() {
-        for (provider_failure, binding_failure, expected_events) in [
-            (ProviderFailure::Exchange, false, vec!["exchange"]),
-            (ProviderFailure::None, true, vec!["exchange", "bind"]),
-            (ProviderFailure::Logout, false, vec!["exchange", "bind", "logout"]),
-        ] {
-            let events = Arc::new(Mutex::new(Vec::new()));
-            let mut provider = TestProvider::new(events.clone());
-            provider.failure = provider_failure;
-            let mut binding = RecordingBinding::new(events.clone());
-            binding.fail = binding_failure;
-            let service = standard_service(Arc::new(provider));
-
-            service
-                .complete_authorization_code("state", "code", "https://console.example/callback", 3600, &binding)
-                .await
-                .expect_err("the configured standard callback failure should be returned");
-
-            assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), expected_events);
-        }
-    }
-
-    #[tokio::test]
-    async fn standard_runtime_preserves_web_identity_failure_order() {
-        for (provider_failure, binding_failure, expected_events) in [
-            (ProviderFailure::Verification, false, vec!["verify"]),
-            (ProviderFailure::None, true, vec!["verify", "bind"]),
-        ] {
-            let events = Arc::new(Mutex::new(Vec::new()));
-            let mut provider = TestProvider::new(events.clone());
-            provider.failure = provider_failure;
-            let mut binding = RecordingBinding::new(events.clone());
-            binding.fail = binding_failure;
-            let service = standard_service(Arc::new(provider));
-
-            service
-                .assume_role_with_web_identity("jwt", 3600, None, &binding)
-                .await
-                .expect_err("the configured standard web identity failure should be returned");
-
-            assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), expected_events);
-        }
-    }
-
-    #[tokio::test]
-    async fn web_identity_without_policy_or_group_is_not_bound() {
-        let events = Arc::new(Mutex::new(Vec::new()));
-        let mut provider = TestProvider::new(events.clone());
-        provider.with_policy = false;
-        let provider = Arc::new(provider);
-        let binding = Arc::new(RecordingBinding::new(events.clone()));
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
-
         let error = service
             .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
             .await
@@ -689,16 +578,15 @@ mod tests {
         let mut provider = TestProvider::new(events.clone());
         provider.with_policy = false;
         provider.with_group = true;
-        let provider = Arc::new(provider);
         let binding = Arc::new(RecordingBinding::new(events.clone()));
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
+        let service = standard_service(Arc::new(provider));
 
         let session = service
             .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
             .await
             .expect("a mapped group is an authorization context");
 
-        assert!(session.authorization.policies.is_empty());
+        assert_eq!(session.authorization.policies, ["developers"]);
         assert_eq!(session.authorization.groups, ["developers"]);
         assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), ["verify", "bind"]);
     }
@@ -716,7 +604,7 @@ mod tests {
             let mut binding = RecordingBinding::new(events.clone());
             binding.fail = binding_failure;
             let binding = Arc::new(binding);
-            let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
+            let service = standard_service(Arc::new(provider));
 
             let error = service
                 .complete_authorization_code("state", "code", "https://console.example/callback", 3600, binding.as_ref())
@@ -755,7 +643,7 @@ mod tests {
             let mut binding = RecordingBinding::new(events.clone());
             binding.fail = binding_failure;
             let binding = Arc::new(binding);
-            let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
+            let service = standard_service(Arc::new(provider));
 
             let error = service
                 .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())

crates/iam/src/oidc.rs

@@ -2354,8 +2354,9 @@ fn extract_canonical_group_values(
     groups
 }
 
-#[cfg(test)]
-pub(crate) fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
+#[cfg(any(test, feature = "test-util"))]
+#[doc(hidden)]
+pub fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
     let configs = configs
         .into_iter()
         .map(test_sourced_config)
@@ -2369,7 +2370,7 @@ pub(crate) fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
     }
 }
 
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 fn test_sourced_config(config: OidcProviderConfig) -> SourcedOidcProviderConfig {
     SourcedOidcProviderConfig {
         config,
@@ -2377,8 +2378,9 @@ fn test_sourced_config(config: OidcProviderConfig) -> SourcedOidcProviderConfig
     }
 }
 
-#[cfg(test)]
-pub(crate) fn test_config(id: &str) -> OidcProviderConfig {
+#[cfg(any(test, feature = "test-util"))]
+#[doc(hidden)]
+pub fn test_config(id: &str) -> OidcProviderConfig {
     OidcProviderConfig {
         id: id.to_string(),
         enabled: true,

crates/policy/src/policy/variables.rs

@@ -252,6 +252,13 @@ impl PolicyVariableResolver for VariableResolver {
 }
 
 pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableResolver) -> Vec<String> {
+    resolve_aws_variables_with_depth(pattern, resolver, 0).await
+}
+
+async fn resolve_aws_variables_with_depth(pattern: &str, resolver: &dyn PolicyVariableResolver, depth: usize) -> Vec<String> {
+    if depth >= 10 {
+        return vec![pattern.to_string()];
+    }
     let mut results = vec![pattern.to_string()];
 
     let mut changed = true;
@@ -264,7 +271,7 @@ pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableR
 
         let mut new_results = Vec::new();
         for result in &results {
-            let resolved = resolve_single_pass(result, resolver).await;
+            let resolved = resolve_single_pass(result, resolver, depth).await;
             if resolved.len() > 1 || (resolved.len() == 1 && &resolved[0] != result) {
                 changed = true;
             }
@@ -288,30 +295,31 @@ pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableR
 fn resolve_aws_variables_boxed<'a>(
     pattern: &'a str,
     resolver: &'a dyn PolicyVariableResolver,
+    depth: usize,
 ) -> std::pin::Pin<Box<dyn Future<Output = Vec<String>> + Send + 'a>> {
-    Box::pin(resolve_aws_variables(pattern, resolver))
+    Box::pin(resolve_aws_variables_with_depth(pattern, resolver, depth))
 }
 
 /// Single pass resolution of variables in a string
-async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolver) -> Vec<String> {
+async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolver, depth: usize) -> Vec<String> {
     // Find all ${...} format variables
-    let mut results = vec![pattern.to_string()];
+    let mut results = vec![(pattern.to_string(), 0)];
 
     // Process each result string
     let mut i = 0;
     while i < results.len() {
-        let mut start = 0;
+        let mut start = results[i].1;
         let mut modified = false;
 
         // Find variables in current string
-        while let Some(pos) = results[i][start..].find("${") {
+        while let Some(pos) = results[i].0[start..].find("${") {
             let actual_pos = start + pos;
 
             // Find the matching closing brace, taking into account nested braces
             let mut brace_count = 1;
             let mut end_pos = actual_pos + 2; // Start after "${"
 
-            let bytes = results[i].as_bytes();
+            let bytes = results[i].0.as_bytes();
             while end_pos < bytes.len() && brace_count > 0 {
                 match bytes[end_pos] {
                     b'{' => brace_count += 1,
@@ -324,19 +332,23 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
             }
 
             if brace_count == 0 {
-                let var_name = &results[i][actual_pos + 2..end_pos];
+                let var_name = &results[i].0[actual_pos + 2..end_pos];
 
                 // Check if this is a nested variable (contains ${...} inside)
                 if var_name.contains("${") {
                     // For nested variables like ${${a}-${b}}, we need to resolve the inner variables first
                     // Then use the resolved result as a new variable to resolve
-                    let resolved_inner = resolve_aws_variables_boxed(var_name, resolver).await;
+                    let resolved_inner = resolve_aws_variables_boxed(var_name, resolver, depth + 1).await;
+                    if resolved_inner.len() == 1 && resolved_inner[0] == var_name {
+                        start = end_pos + 1;
+                        continue;
+                    }
                     let mut new_results = Vec::new();
 
                     for resolved_var_name in resolved_inner {
-                        let prefix = &results[i][..actual_pos];
-                        let suffix = &results[i][end_pos + 1..];
-                        new_results.push(format!("{prefix}{resolved_var_name}{suffix}"));
+                        let prefix = &results[i].0[..actual_pos];
+                        let suffix = &results[i].0[end_pos + 1..];
+                        new_results.push((format!("{prefix}{resolved_var_name}{suffix}"), actual_pos + resolved_var_name.len()));
                     }
 
                     if !new_results.is_empty() {
@@ -354,11 +366,13 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
                         if !values.is_empty() {
                             // If there are multiple values, create a new result for each value
                             let mut new_results = Vec::new();
-                            let prefix = &results[i][..actual_pos];
-                            let suffix = &results[i][end_pos + 1..];
+                            let prefix = &results[i].0[..actual_pos];
+                            let suffix = &results[i].0[end_pos + 1..];
 
                             for value in values {
-                                new_results.push(format!("{prefix}{value}{suffix}"));
+                                // Resume after the substitution; a generated placeholder
+                                // belongs to the next bounded pass, not this one.
+                                new_results.push((format!("{prefix}{value}{suffix}"), actual_pos + value.len()));
                             }
 
                             results.splice(i..i + 1, new_results);
@@ -367,9 +381,9 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
                         } else {
                             // Variable resolved to empty, just remove the variable placeholder
                             let mut new_results = Vec::new();
-                            let prefix = &results[i][..actual_pos];
-                            let suffix = &results[i][end_pos + 1..];
-                            new_results.push(format!("{prefix}{suffix}"));
+                            let prefix = &results[i].0[..actual_pos];
+                            let suffix = &results[i].0[end_pos + 1..];
+                            new_results.push((format!("{prefix}{suffix}"), actual_pos));
 
                             results.splice(i..i + 1, new_results);
                             modified = true;
@@ -391,7 +405,7 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
         }
     }
 
-    results
+    results.into_iter().map(|(value, _)| value).collect()
 }
 
 #[cfg(test)]
@@ -399,6 +413,62 @@ mod tests {
     use super::*;
     use serde_json::Value;
     use std::collections::HashMap;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    struct CycleResolver {
+        calls: AtomicUsize,
+    }
+
+    #[async_trait]
+    impl PolicyVariableResolver for CycleResolver {
+        async fn resolve(&self, name: &str) -> Option<String> {
+            // Make the vulnerable implementation terminate so a regression
+            // reports an assertion failure instead of hanging the test runner.
+            if self.calls.fetch_add(1, Ordering::Relaxed) >= 100 {
+                return None;
+            }
+            match name {
+                "aws:AccountId" => Some("$$".to_string()),
+                "aws:username" => Some("{aws:AccountId}{aws:username}".to_string()),
+                _ => None,
+            }
+        }
+
+        fn is_dynamic(&self, _: &str) -> bool {
+            false
+        }
+    }
+
+    #[tokio::test]
+    async fn generated_policy_variable_cycles_are_bounded() {
+        let resolver = CycleResolver {
+            calls: AtomicUsize::new(0),
+        };
+        let result = resolve_aws_variables("${aws:AccountId}{aws:username}", &resolver).await;
+        assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
+        assert_eq!(result, vec!["${aws:AccountId}{aws:username}".to_string()]);
+    }
+
+    #[tokio::test]
+    async fn a_single_pass_resolves_all_original_variables() {
+        let mut context = VariableContext::new();
+        context.username = Some("alice".to_string());
+        let resolver = VariableResolver::new(context);
+        let pattern = "${aws:username}".repeat(20);
+        assert_eq!(resolve_single_pass(&pattern, &resolver, 0).await, vec!["alice".repeat(20)]);
+    }
+
+    #[tokio::test]
+    async fn nested_policy_variables_have_a_depth_limit() {
+        let resolver = CycleResolver {
+            calls: AtomicUsize::new(0),
+        };
+        let pattern = format!("{}aws:username{}", "${".repeat(100), "}".repeat(100));
+        let result = resolve_aws_variables(&pattern, &resolver).await;
+        assert_eq!(result.len(), 1);
+        assert!(result[0].contains("${"));
+        assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
+    }
 
     #[tokio::test]
     async fn test_resolve_aws_variables_with_username() {

crates/rio/Cargo.toml

@@ -80,6 +80,7 @@ rustfs-tls-runtime.workspace = true
 rustfs-utils = { workspace = true, features = ["io", "hash", "compress"] }
 serde_json = { workspace = true, features = ["raw_value"] }
 md-5 = { workspace = true }
+hmac.workspace = true
 minlz.workspace = true
 tracing.workspace = true
 thiserror.workspace = true

crates/rio/src/etag_reader.rs

@@ -14,8 +14,10 @@
 
 use crate::compress_index::{Index, TryGetIndex};
 use crate::{BadDigest, EtagResolvable, HashReaderDetector, HashReaderMut};
+use hmac::{Hmac, KeyInit, Mac};
 use md5::{Digest, Md5};
 use pin_project_lite::pin_project;
+use sha2::Sha256;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, ReadBuf};
@@ -29,6 +31,7 @@ pin_project! {
         pub finished: bool,
         pub checksum: Option<String>,
         resolved_etag: Option<String>,
+        encrypted_etag_mac: Option<Hmac<Sha256>>,
     }
 }
 
@@ -40,9 +43,21 @@ impl<R> EtagReader<R> {
             finished: false,
             checksum,
             resolved_etag: None,
+            encrypted_etag_mac: None,
         }
     }
 
+    /// Protect the content fingerprint with the object's encryption material.
+    /// Plaintext checksum verification remains the inner reader's responsibility.
+    pub fn new_encrypted(inner: R, key: [u8; 32], nonce: [u8; 12]) -> std::io::Result<Self> {
+        let mut mac = <Hmac<Sha256> as KeyInit>::new_from_slice(&key).map_err(std::io::Error::other)?;
+        mac.update(b"RustFS:encrypted-ETag:v1\0");
+        mac.update(&nonce);
+        let mut reader = Self::new(inner, None);
+        reader.encrypted_etag_mac = Some(mac);
+        Ok(reader)
+    }
+
     /// Get the final md5 value (etag) as a hex string, only compute once.
     /// Can be called multiple times, always returns the same result after finished.
     pub fn get_etag(&mut self) -> String {
@@ -108,6 +123,14 @@ impl<R> EtagResolvable for EtagReader<R> {
         true
     }
     fn try_resolve_etag(&mut self) -> Option<String> {
+        if let Some(mut mac) = self.encrypted_etag_mac.clone() {
+            if !self.finished {
+                return None;
+            }
+            mac.update(self.get_etag().as_bytes());
+            let protected = mac.finalize().into_bytes();
+            return Some(hex_simd::encode_to_string(&protected[..16], hex_simd::AsciiCase::Lower));
+        }
         // EtagReader provides its own etag, not delegating to inner
         if let Some(checksum) = &self.checksum {
             Some(checksum.clone())
@@ -148,6 +171,47 @@ mod tests {
     use std::io::Cursor;
     use tokio::io::{AsyncReadExt, BufReader};
 
+    #[tokio::test]
+    async fn encrypted_etags_hide_plaintext_fingerprints_including_empty_objects() {
+        for data in [b"".as_slice(), b"Salary 2026: 85000 EUR".as_slice()] {
+            let md5 = hex_simd::encode_to_string(Md5::digest(data), hex_simd::AsciiCase::Lower);
+            let mut tags = Vec::new();
+            for (key, nonce) in [([1; 32], [1; 12]), ([2; 32], [1; 12]), ([1; 32], [2; 12])] {
+                let mut reader = EtagReader::new_encrypted(Cursor::new(data), key, nonce).expect("create protected ETag reader");
+                assert!(reader.try_resolve_etag().is_none());
+                let mut output = Vec::new();
+                reader.read_to_end(&mut output).await.expect("read unchanged plaintext");
+                assert_eq!(output, data);
+                let tag = reader.try_resolve_etag().expect("resolve protected ETag after EOF");
+                assert_eq!(tag.len(), 32);
+                assert_ne!(tag, md5);
+                assert_eq!(reader.try_resolve_etag().as_ref(), Some(&tag));
+                tags.push(tag);
+            }
+            assert_ne!(tags[0], tags[1], "different keys must not link identical content");
+            assert_ne!(tags[0], tags[2], "different objects under a legacy key must not link identical content");
+        }
+    }
+
+    #[tokio::test]
+    async fn encrypted_etag_reader_preserves_plaintext_digest_rejection() {
+        let plaintext = crate::HashReader::from_stream(
+            Cursor::new(b"actual plaintext"),
+            16,
+            16,
+            Some("00000000000000000000000000000000".to_owned()),
+            None,
+            false,
+        )
+        .expect("create checked plaintext reader");
+        let mut reader = EtagReader::new_encrypted(plaintext, [1; 32], [2; 12]).expect("create protected ETag reader");
+        let err = reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("wrong plaintext MD5 must still be rejected");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
     #[tokio::test]
     async fn test_etag_reader_basic() {
         let data = b"hello world";

crates/rio/src/hash_reader.rs

@@ -136,6 +136,7 @@ pin_project! {
         content_sha256: Option<String>,
         content_sha256_hasher: Option<Sha256Hasher>,
         checksum_on_finish: bool,
+        pub encrypted_etag: bool,
 
         // Read only at EOF; see the timing contract in `crate::trailer`.
         trailer: Option<SharedTrailerSource>,
@@ -184,6 +185,7 @@ impl HashReader {
             content_sha256: sha256hex.clone(),
             content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
             checksum_on_finish: false,
+            encrypted_etag: false,
             trailer: None,
         })
     }
@@ -224,6 +226,7 @@ impl HashReader {
             content_sha256: sha256hex.clone(),
             content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
             checksum_on_finish: false,
+            encrypted_etag: false,
             trailer: None,
         })
     }
@@ -294,6 +297,7 @@ impl HashReader {
                 content_hash,
                 content_hasher,
                 checksum_on_finish: false,
+                encrypted_etag: false,
                 trailer: existing_hash_reader.get_trailer().cloned(),
             })
         } else {
@@ -322,6 +326,7 @@ impl HashReader {
                 content_sha256: sha256hex.clone(),
                 content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
                 checksum_on_finish: false,
+                encrypted_etag: false,
                 trailer: None,
             })
         }

crates/utils/src/http/object_encryption_keys.rs

@@ -55,11 +55,9 @@ pub const MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER: &str = "X-Minio-Internal
 
 /// Reserved RustFS-branded twin of the MinIO-internal SSE key family.
 ///
-/// No RustFS writer emits these keys today — the SSE writer persists the
-/// MinIO-branded `X-Minio-Internal-Server-Side-Encryption-*` keys verbatim for
-/// interoperability — but redaction (`rustfs_filemeta`) and replication
-/// stripping treat the family as sensitive so that a future or third-party
-/// writer cannot leak sealed material through the reserved names.
+/// The SSE writer persists sealed material under the MinIO-branded names for
+/// interoperability. Redaction and replication stripping also cover this twin
+/// family, including the encrypted ETag format marker.
 pub const RUSTFS_INTERNAL_ENCRYPTION_PREFIX: &str = "x-rustfs-internal-server-side-encryption-";
 
 pub const REPLICATION_SSEC_ALGORITHM_HEADER: &str = "X-Rustfs-Replication-Ssec-Algorithm";
@@ -71,6 +69,15 @@ pub const REPLICATION_SSE_SEAL_ALGORITHM_HEADER: &str = "X-Rustfs-Replication-Se
 pub const REPLICATION_SSE_SEALED_KEY_HEADER: &str = "X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key";
 pub const REPLICATION_ENCRYPTED_MULTIPART_HEADER: &str = "X-Rustfs-Replication-Encrypted-Multipart";
 
+pub const SUFFIX_OPAQUE_ENCRYPTED_ETAG: &str = "server-side-encryption-opaque-etag";
+
+/// Old encrypted objects can carry a plaintext MD5. Keyless listings expose
+/// an encrypted ETag only when its producer recorded the protected format.
+pub fn etag_can_be_listed(metadata: &HashMap<String, String>) -> bool {
+    !metadata.keys().any(|key| is_replication_stripped_encryption_key(key))
+        || super::metadata_compat::get_consistent_str(metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG) == Some("v1")
+}
+
 /// Stored SSE-C metadata keys and the wire names they replicate under.
 ///
 /// Source keys must match what `encryption_material_to_metadata` persists; the
@@ -307,6 +314,28 @@ pub fn stored_managed_encryption_key(metadata: &HashMap<String, String>) -> Opti
 mod tests {
     use super::*;
 
+    #[test]
+    fn keyless_listings_hide_unproven_encrypted_etags() {
+        assert!(etag_can_be_listed(&HashMap::from([("content-type".to_owned(), "text/plain".to_owned())])));
+        for key in [
+            SSEC_ALGORITHM_HEADER,
+            INTERNAL_ENCRYPTION_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER,
+            "X-AMZ-SERVER-SIDE-ENCRYPTION",
+            "x-rustfs-internal-server-side-encryption-sealed-key",
+        ] {
+            let mut metadata = HashMap::from([(key.to_owned(), "encryption-metadata".to_owned())]);
+            assert!(!etag_can_be_listed(&metadata), "legacy encrypted metadata: {key}");
+            super::super::metadata_compat::insert_str(&mut metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG, "unknown".to_owned());
+            assert!(!etag_can_be_listed(&metadata), "unknown producer format: {key}");
+            super::super::metadata_compat::insert_str(&mut metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+            assert!(etag_can_be_listed(&metadata), "protected encrypted metadata: {key}");
+            metadata.insert("x-minio-internal-server-side-encryption-opaque-etag".to_owned(), "unknown".to_owned());
+            assert!(!etag_can_be_listed(&metadata), "conflicting format markers: {key}");
+        }
+    }
+
     #[test]
     fn ssec_transport_projection_retains_only_redacted_reader_headers() {
         let mut headers = http::HeaderMap::new();

crates/zip/Cargo.toml

@@ -49,6 +49,10 @@ rustfs-rio.workspace = true
 tokio = { workspace = true, features = ["io-util", "macros", "rt"] }
 thiserror = { workspace = true }
 
+[target.'cfg(target_os = "macos")'.dependencies]
+# Keep XZ decoding self-contained in macOS binaries instead of linking Homebrew liblzma.
+liblzma = { version = "0.4.8", features = ["static"] }
+
 [dev-dependencies]
 rustfs-tokio-tar = { workspace = true }
 futures = { workspace = true }

docs/README.md

@@ -21,6 +21,9 @@ For bounded local read-backend startup and cancellation, see
 
 For legacy protection assessment and bounded protected copies, see [Object integrity inventory, audit, and migration](operations/shard-integrity-audit.md).
 
+For encrypted listing compatibility and older stored fingerprints, see
+[Encrypted object ETags](operations/encrypted-etags.md).
+
 Operational runbooks live under [`operations/`](operations/). Replication
 operators should start with:
 

docs/operations/encrypted-etags.md

@@ -0,0 +1,44 @@
+# Encrypted object ETags
+
+**Use this when:** upgrading encrypted storage or investigating missing ETags
+in object listings after an upgrade.
+
+New encrypted writes use a keyed fingerprint for the ETag rather than the
+plaintext MD5. Single-part and part ETags are 32-character hexadecimal values;
+completed multipart ETags retain their digest-and-part-count form. They stay
+stable for the stored object. Request MD5 and checksum validation still apply
+to the original plaintext. Encryption and decryption formats do not change.
+
+## Listings and older objects
+
+ListObjects, ListObjectsV2, and ListObjectVersions omit the optional ETag field
+for encrypted objects whose metadata does not attest to the protected format.
+This includes encrypted objects written before the upgrade, preserved source
+ETags without that proof, and completed encrypted multipart uploads. New
+single-part encrypted writes carry the internal format marker and expose their
+protected ETag in listings. Unencrypted listings retain their existing ETags.
+Existing persistent listing snapshots are invalidated; the provider falls
+back to verified storage until a new snapshot is built.
+
+Clients that require an ETag for these older or multipart objects must issue an
+authorized HeadObject request, including the SSE-C key when required. GET,
+HEAD, and conditional requests continue to use the object's stored ETag.
+
+## Existing data and migration
+
+The upgrade does not rewrite existing object metadata or erase fingerprints
+already copied to backups, logs, caches, or replicas. Restrict access to those
+copies separately. Rolling back to an older server also restores its listing
+behavior, so keep the upgraded listing protection on every serving node.
+
+To obtain a protected ETag, read the object with its encryption credentials and
+write it through the upgraded encryption path. Verify the resulting plaintext
+and application references before retiring an old copy. A server-side copy
+that preserves the source ETag is insufficient.
+
+In a versioned bucket, a new write creates a new version; historical versions
+keep their original metadata. Object Lock retention and legal holds still
+apply. Do not bypass them or delete retained versions to migrate an ETag.
+There is no automatic in-place migration of retained metadata in this upgrade.
+Unfinished uploads also keep their original part metadata. Restart an upload
+through the upgraded path when its legacy part fingerprints need to be replaced.

rustfs/Cargo.toml

@@ -389,6 +389,7 @@ symbolic-demangle = { version = "=13.9.0", default-features = false, features =
 
 [dev-dependencies]
 uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
+rustfs-iam = { workspace = true, features = ["test-util"] }
 serial_test = { workspace = true }
 tempfile = { workspace = true }
 aws-config = { workspace = true }

rustfs/src/admin/handlers/bucket_recovery.rs

@@ -17,12 +17,12 @@
 use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_object_store_handle;
+use crate::admin::storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
 use crate::server::ADMIN_PREFIX;
 use crate::site_replication::{site_replication_enabled, with_site_replication_bucket_mutation_lock};
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_policy::policy::action::{Action, AdminAction};
-use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
 use serde::{Deserialize, Serialize};
 use tracing::info;
 use uuid::Uuid;
@@ -63,23 +63,23 @@ impl Operation for RecoverOrphanedBucketHandler {
             .credentials
             .as_ref()
             .map(|credentials| credentials.secret_key.clone())
-            .ok_or_else(|| s3_error!(InvalidRequest, "authentication required"))?;
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "authentication required"))?;
         let bucket = params
             .get("bucket")
             .filter(|bucket| !bucket.is_empty())
-            .ok_or_else(|| s3_error!(InvalidRequest, "bucket name is required"))?;
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "bucket name is required"))?;
         let body = req
             .input
             .store_all_limited(rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE)
             .await
-            .map_err(|_| s3_error!(InvalidRequest, "failed to read recovery request body"))?;
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "failed to read recovery request body"))?;
         let request: RecoverOrphanedBucketRequest = serde_json::from_slice(&body)
-            .map_err(|_| s3_error!(InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?;
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?;
         if request.expected_incarnation_id.is_nil() {
-            return Err(s3_error!(InvalidRequest, "expectedIncarnationId must be non-nil"));
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "expectedIncarnationId must be non-nil"));
         }
         let Some(store) = current_object_store_handle() else {
-            return Err(s3_error!(InternalError, "object store is not initialized"));
+            return Err(s3::error(S3ErrorCode::InternalError, "object store is not initialized"));
         };
 
         let operation_bucket = bucket.to_owned();
@@ -91,21 +91,21 @@ impl Operation for RecoverOrphanedBucketHandler {
                 let lock_bucket = operation_bucket.clone();
                 with_site_replication_bucket_mutation_lock(lock_store, &lock_bucket, move || async move {
                     if site_replication_enabled().await? {
-                        return Err(s3_error!(
-                            OperationAborted,
-                            "orphaned bucket recovery is local; reconcile every site before recreating this bucket"
+                        return Err(s3::error(
+                            S3ErrorCode::OperationAborted,
+                            "orphaned bucket recovery is local; reconcile every site before recreating this bucket",
                         ));
                     }
                     operation_store
                         .recover_orphaned_bucket(&operation_bucket, expected_incarnation)
                         .await
-                        .map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))
+                        .map_err(|error| S3Error::from(crate::error::ApiError::from(error)))
                 })
                 .await??;
-                Ok::<(), s3s::S3Error>(())
+                Ok::<(), S3Error>(())
             })
             .await
-            .map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))?;
+            .map_err(|error| S3Error::from(crate::error::ApiError::from(error)))?;
         completed?;
 
         info!(

rustfs/src/admin/handlers/site_replication.rs

@@ -15,7 +15,7 @@
 use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{
-    current_deployment_id, current_federated_identity_runtime, current_iam_handle, current_object_store_handle, current_region,
+    current_deployment_id, current_iam_handle, current_object_store_handle, current_oidc_config_query, current_region,
     current_replication_pool_handle, current_replication_stats_handle, current_server_config, current_token_signing_key,
     object_store_from_req,
 };
@@ -2086,40 +2086,11 @@ fn open_id_provider_settings(config: &rustfs_iam::oidc::OidcProviderConfig) -> O
     }
 }
 
-fn open_id_settings_from_service(service: &rustfs_iam::federation::FederatedIdentityService, region: String) -> OpenIDSettings {
-    let providers = service
-        .list_providers()
-        .into_iter()
-        .filter_map(|provider| {
-            let config = service.get_provider_config(&provider.provider_id)?;
-            Some((provider.provider_id, open_id_provider_settings(config)))
-        })
-        .collect();
-
-    open_id_settings(providers, region)
-}
-
-fn open_id_settings_from_runtime(
-    oidc_config_query: Option<&dyn rustfs_iam::federation::oidc::OidcConfigQuery>,
-    service: Option<&rustfs_iam::federation::FederatedIdentityService>,
-    region: String,
-) -> OpenIDSettings {
-    if let Some(oidc_config_query) = oidc_config_query {
-        return open_id_settings_from_snapshot(&oidc_config_query.config_snapshot(), region);
-    }
-
-    service
-        .map(|service| open_id_settings_from_service(service, region))
-        .unwrap_or_default()
-}
-
 fn local_idp_settings() -> IDPSettings {
     let mut settings = IDPSettings::default();
     let region = current_region().map(|region| region.to_string()).unwrap_or_default();
-    settings.open_id = current_federated_identity_runtime()
-        .map(|(service, oidc_config_query)| {
-            open_id_settings_from_runtime(oidc_config_query.as_deref(), Some(service.as_ref()), region)
-        })
+    settings.open_id = current_oidc_config_query()
+        .map(|query| open_id_settings_from_snapshot(&query.config_snapshot(), region))
         .unwrap_or_default();
 
     let (ldap, ldap_configs) = load_ldap_idp_settings();
@@ -11444,82 +11415,6 @@ mod tests {
         }
     }
 
-    struct LegacyOidcProvider {
-        configs: Vec<rustfs_iam::oidc::OidcProviderConfig>,
-    }
-
-    struct SnapshotOidcConfigQuery {
-        snapshot: rustfs_iam::oidc::OidcConfigSnapshot,
-    }
-
-    impl rustfs_iam::federation::oidc::OidcConfigQuery for SnapshotOidcConfigQuery {
-        fn config_snapshot(&self) -> rustfs_iam::oidc::OidcConfigSnapshot {
-            self.snapshot.clone()
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl rustfs_iam::federation::FederatedIdentityProvider for LegacyOidcProvider {
-        fn has_providers(&self) -> bool {
-            !self.configs.is_empty()
-        }
-
-        fn list_providers(&self) -> Vec<rustfs_iam::federation::FederatedProviderView> {
-            self.configs
-                .iter()
-                .map(|config| rustfs_iam::federation::FederatedProviderView {
-                    provider_id: config.id.clone(),
-                    display_name: config.display_name.clone(),
-                })
-                .collect()
-        }
-
-        fn list_visible_providers(&self) -> Vec<rustfs_iam::federation::FederatedProviderView> {
-            self.list_providers()
-        }
-
-        fn provider_config(&self, provider_id: &str) -> Option<&rustfs_iam::oidc::OidcProviderConfig> {
-            self.configs.iter().find(|config| config.id == provider_id)
-        }
-
-        async fn authorize_url(
-            &self,
-            _provider_id: &str,
-            _redirect_uri: &str,
-            _redirect_after: Option<String>,
-        ) -> rustfs_iam::federation::Result<String> {
-            unreachable!("site replication compatibility test does not authorize")
-        }
-
-        async fn exchange_code(
-            &self,
-            _state: &str,
-            _code: &str,
-            _redirect_uri: &str,
-        ) -> rustfs_iam::federation::Result<rustfs_iam::federation::FederatedCodeExchange> {
-            unreachable!("site replication compatibility test does not exchange codes")
-        }
-
-        async fn verify_web_identity_token(
-            &self,
-            _jwt: &str,
-        ) -> rustfs_iam::federation::Result<rustfs_iam::federation::FederatedAuthorization> {
-            unreachable!("site replication compatibility test does not verify tokens")
-        }
-
-        async fn create_logout_token(&self, _provider_id: &str, _id_token: &str) -> rustfs_iam::federation::Result<String> {
-            unreachable!("site replication compatibility test does not create logout tokens")
-        }
-
-        async fn build_logout_url(
-            &self,
-            _logout_token: &str,
-            _post_logout_redirect_uri: &str,
-        ) -> rustfs_iam::federation::Result<Option<String>> {
-            unreachable!("site replication compatibility test does not build logout URLs")
-        }
-    }
-
     fn ldap_settings_kvs() -> rustfs_config::server_config::KVS {
         rustfs_config::server_config::KVS(vec![
             rustfs_config::server_config::KV {
@@ -11692,43 +11587,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn service_only_oidc_runtime_preserves_site_replication_settings() {
-        let default = oidc_snapshot_provider("default", "client-a");
-        let corp = oidc_snapshot_provider("corp", "client-b");
-        let expected = open_id_settings_from_snapshot(
-            &rustfs_iam::oidc::OidcConfigSnapshot::new(vec![corp.clone(), default.clone()]),
-            "eu-site-1".to_string(),
-        );
-        let service = rustfs_iam::federation::FederatedIdentityService::new(
-            rustfs_iam::federation::FederatedIdentityRegistry::new(Arc::new(LegacyOidcProvider {
-                configs: vec![corp.config, default.config],
-            })),
-        );
-
-        assert_eq!(
-            serde_json::to_value(open_id_settings_from_runtime(None, Some(&service), "eu-site-1".to_string(),))
-                .expect("serialize legacy OpenID settings"),
-            serde_json::to_value(expected).expect("serialize snapshot OpenID settings"),
-        );
-    }
-
-    #[test]
-    fn oidc_query_takes_precedence_over_service_only_compatibility() {
-        let query = SnapshotOidcConfigQuery {
-            snapshot: rustfs_iam::oidc::OidcConfigSnapshot::new(vec![oidc_snapshot_provider("default", "query-client")]),
-        };
-        let service = rustfs_iam::federation::FederatedIdentityService::new(
-            rustfs_iam::federation::FederatedIdentityRegistry::new(Arc::new(LegacyOidcProvider {
-                configs: vec![oidc_snapshot_provider("default", "service-client").config],
-            })),
-        );
-
-        let settings = open_id_settings_from_runtime(Some(&query), Some(&service), "eu-site-1".to_string());
-
-        assert_eq!(settings.claim_provider.client_id, "query-client");
-    }
-
     // A whole provider object present on one side only must not spill its
     // credential-derived leaves into the rejection message.
     #[test]

rustfs/src/admin/runtime_sources.rs

@@ -23,12 +23,11 @@ use crate::app::object_usecase::DefaultObjectUsecase;
 use crate::runtime_sources as root_runtime_sources;
 pub(crate) use crate::runtime_sources::{
     AppContext, ServerContextSlot, current_action_credentials, current_boot_time, current_bucket_metadata_handle,
-    current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_runtime,
-    current_federated_identity_service, current_iam_handle, current_kms_runtime_service_manager,
-    current_notification_system_for_context, current_object_data_cache_handle_for_context,
-    current_object_store_handle_for_context, current_ready_iam_handle, current_region, current_replication_pool_handle,
-    current_replication_stats_handle, current_replication_stats_handle_for_context, current_server_config_for_context,
-    current_token_signing_key,
+    current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_service,
+    current_iam_handle, current_kms_runtime_service_manager, current_notification_system_for_context,
+    current_object_data_cache_handle_for_context, current_object_store_handle_for_context, current_oidc_config_query,
+    current_ready_iam_handle, current_region, current_replication_pool_handle, current_replication_stats_handle,
+    current_replication_stats_handle_for_context, current_server_config_for_context, current_token_signing_key,
 };
 #[cfg(test)]
 pub(crate) use crate::runtime_sources::{

rustfs/src/app/bucket_usecase.rs

@@ -1086,7 +1086,11 @@ fn build_list_object_versions_metadata_output(
                         size: Some(object.size),
                         version_id: Some(version_id),
                         is_latest: Some(object.is_latest),
-                        e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)),
+                        e_tag: object
+                            .etag
+                            .as_ref()
+                            .filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined))
+                            .map(|etag| to_s3s_etag(etag)),
                         storage_class: Some(ObjectVersionStorageClass::from(
                             object
                                 .storage_class
@@ -1173,7 +1177,11 @@ fn build_list_objects_v2_metadata_output(
                     key: Some(encode_list_objects_v2_value(&object.name, encoding_type)),
                     last_modified: object.mod_time.map(Timestamp::from),
                     size: Some(object.get_actual_size_or_physical()),
-                    e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)),
+                    e_tag: object
+                        .etag
+                        .as_ref()
+                        .filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined))
+                        .map(|etag| to_s3s_etag(etag)),
                     storage_class: Some(ObjectStorageClass::from(
                         object
                             .storage_class
@@ -4548,6 +4556,89 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
     }
 
+    #[test]
+    fn keyless_object_and_version_listings_do_not_expose_legacy_encrypted_fingerprints() {
+        let legacy_md5 = "fbfcbcc6d2f035411e6268c8b119593e";
+        let mut protected_metadata =
+            HashMap::from([("x-amz-server-side-encryption-customer-algorithm".to_owned(), "AES256".to_owned())]);
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut protected_metadata,
+            rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+            "v1".to_owned(),
+        );
+        let objects = vec![
+            ObjectInfo {
+                name: "legacy-encrypted".to_owned(),
+                etag: Some(legacy_md5.to_owned()),
+                user_defined: Arc::new(HashMap::from([(
+                    "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+                    "AES256".to_owned(),
+                )])),
+                ..Default::default()
+            },
+            ObjectInfo {
+                name: "protected-encrypted".to_owned(),
+                etag: Some("0123456789abcdef0123456789abcdef".to_owned()),
+                user_defined: Arc::new(protected_metadata),
+                ..Default::default()
+            },
+            ObjectInfo {
+                name: "plain".to_owned(),
+                etag: Some("9876543210abcdef9876543210abcdef".to_owned()),
+                ..Default::default()
+            },
+        ];
+        let listings = build_list_objects_v2_metadata_output(
+            ListObjectsV2Info {
+                objects: objects.clone(),
+                ..Default::default()
+            },
+            "demo-bucket",
+            &ListObjectsV2Params {
+                prefix: String::new(),
+                max_keys: 1000,
+                delimiter: None,
+                response_start_after: None,
+                start_after_for_query: None,
+                response_continuation_token: None,
+                decoded_continuation_token: None,
+            },
+            None,
+            false,
+            &HashMap::new(),
+        );
+        assert!(listings.contents[0].object.e_tag.is_none());
+        assert!(listings.contents[1].object.e_tag.is_some());
+        assert!(listings.contents[2].object.e_tag.is_some());
+        let xml = String::from_utf8(serialize_config(&listings).expect("serialize objects")).expect("UTF-8 objects");
+        assert!(!xml.contains(legacy_md5));
+
+        let versions = build_list_object_versions_metadata_output(
+            ListObjectVersionsInfo {
+                objects,
+                ..Default::default()
+            },
+            "demo-bucket",
+            &ListObjectVersionsParams {
+                prefix: String::new(),
+                max_keys: 1000,
+                delimiter: None,
+                key_marker: None,
+                version_id_marker: None,
+            },
+            None,
+            &HashMap::new(),
+        );
+        for (index, entry) in versions.entries.iter().enumerate() {
+            let ListObjectVersionMetadataEntry::Version(version, _) = entry else {
+                panic!("expected object version");
+            };
+            assert_eq!(version.e_tag.is_some(), index != 0);
+        }
+        let xml = String::from_utf8(serialize_config(&versions).expect("serialize versions")).expect("UTF-8 versions");
+        assert!(!xml.contains(legacy_md5));
+    }
+
     #[test]
     fn build_list_object_versions_metadata_output_maps_metadata_and_preserves_entry_order() {
         use time::macros::datetime;

rustfs/src/app/context.rs

@@ -103,11 +103,6 @@ pub fn resolve_oidc_config_query() -> Option<Arc<dyn OidcConfigQuery>> {
     resolve_oidc_config_query_with(get_global_app_context())
 }
 
-/// Resolve the paired federation service and OIDC configuration query from one runtime observation.
-pub(crate) fn resolve_federated_identity_runtime() -> Option<FederatedIdentityRuntimeSnapshot> {
-    resolve_federated_identity_runtime_with(get_global_app_context())
-}
-
 /// Publish the initialized federation service and OIDC configuration query.
 pub fn publish_federated_identity_runtime(
     service: Arc<FederatedIdentityService>,
@@ -116,11 +111,6 @@ pub fn publish_federated_identity_runtime(
     handles::publish_default_federated_identity_runtime(service, oidc_config_query)
 }
 
-/// Publish a service-only federation runtime for compatibility integrations.
-pub fn publish_federated_identity_service(service: Arc<FederatedIdentityService>) -> bool {
-    handles::publish_default_federated_identity_service(service)
-}
-
 /// Resolve a ready IAM system handle using AppContext-first precedence.
 pub fn resolve_ready_iam_handle() -> rustfs_iam::error::Result<Arc<IamSys<ObjectStore>>> {
     if let Some(context) = get_global_app_context() {
@@ -360,6 +350,7 @@ fn resolve_oidc_config_query_with(context: Option<Arc<AppContext>>) -> Option<Ar
     context.and_then(|context| context.federated_identity().oidc_config_query())
 }
 
+#[cfg(test)]
 fn resolve_federated_identity_runtime_with(context: Option<Arc<AppContext>>) -> Option<FederatedIdentityRuntimeSnapshot> {
     context.and_then(|context| context.federated_identity().runtime_snapshot())
 }
@@ -554,7 +545,7 @@ mod tests {
     use async_trait::async_trait;
     use rustfs_iam::{
         federation::{
-            FederatedIdentityRegistry, FederatedIdentityService,
+            CoreFederatedAuthorizationMapper, FederatedIdentityService,
             oidc::{OidcConfigQuery, StandardOidcAdapter},
         },
         oidc::OidcSys,
@@ -617,25 +608,11 @@ mod tests {
     }
 
     impl FederatedIdentityInterface for TestFederatedIdentityInterface {
-        fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
-            self.runtime
-                .read()
-                .ok()
-                .and_then(|runtime| runtime.as_ref().map(|runtime| Arc::clone(&runtime.service)))
-        }
-
-        fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
-            self.runtime
-                .read()
-                .ok()
-                .and_then(|runtime| runtime.as_ref().map(|runtime| Arc::clone(&runtime.oidc_config_query)))
-        }
-
         fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
             self.runtime.read().ok().and_then(|runtime| {
                 runtime
                     .as_ref()
-                    .map(|runtime| (Arc::clone(&runtime.service), Some(Arc::clone(&runtime.oidc_config_query))))
+                    .map(|runtime| (Arc::clone(&runtime.service), Arc::clone(&runtime.oidc_config_query)))
             })
         }
 
@@ -654,7 +631,8 @@ mod tests {
     fn test_federated_identity_runtime(oidc: OidcSys) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
         let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
         let oidc_config_query = adapter.clone();
-        let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(adapter)));
+        let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+        let service = Arc::new(adapter.into_service(mapper));
         (service, oidc_config_query)
     }
 
@@ -1206,11 +1184,7 @@ mod tests {
         let (resolved_oidc, resolved_oidc_query) =
             resolve_federated_identity_runtime_with(Some(context.clone())).expect("context federated identity runtime");
         assert!(Arc::ptr_eq(&resolved_oidc, &context_oidc));
-        assert!(
-            resolved_oidc_query
-                .as_ref()
-                .is_some_and(|query| Arc::ptr_eq(query, &context_oidc_query))
-        );
+        assert!(Arc::ptr_eq(&resolved_oidc_query, &context_oidc_query));
         assert!(publish_federated_identity_runtime_with(
             Some(context.clone()),
             fallback_oidc.clone(),

rustfs/src/app/context/global.rs

@@ -169,11 +169,6 @@ impl AppContext {
         self.federated_identity.publish_runtime(service, oidc_config_query)
     }
 
-    /// Publish a service-only federation runtime for compatibility integrations.
-    pub fn publish_federated_identity_service(&self, service: Arc<FederatedIdentityService>) -> bool {
-        self.federated_identity.publish_handle(service)
-    }
-
     pub fn kms(&self) -> Arc<dyn KmsInterface> {
         self.kms.clone()
     }

rustfs/src/app/context/handles.rs

@@ -87,24 +87,15 @@ pub struct FederatedIdentityHandle {
 
 struct FederatedIdentityRuntime {
     service: Arc<FederatedIdentityService>,
-    oidc_config_query: Option<Arc<dyn OidcConfigQuery>>,
+    oidc_config_query: Arc<dyn OidcConfigQuery>,
 }
 
 impl FederatedIdentityHandle {
-    pub fn new(service: Option<Arc<FederatedIdentityService>>) -> Self {
-        Self {
-            runtime: StdRwLock::new(service.map(|service| FederatedIdentityRuntime {
-                service,
-                oidc_config_query: None,
-            })),
-        }
-    }
-
     pub fn with_runtime(runtime: Option<(Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>)>) -> Self {
         Self {
             runtime: StdRwLock::new(runtime.map(|(service, oidc_config_query)| FederatedIdentityRuntime {
                 service,
-                oidc_config_query: Some(oidc_config_query),
+                oidc_config_query,
             })),
         }
     }
@@ -112,45 +103,26 @@ impl FederatedIdentityHandle {
 
 impl Default for FederatedIdentityHandle {
     fn default() -> Self {
-        Self::new(None)
+        Self::with_runtime(None)
     }
 }
 
 impl FederatedIdentityInterface for FederatedIdentityHandle {
-    fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
-        self.runtime_snapshot().map(|(service, _)| service)
-    }
-
-    fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
-        self.runtime_snapshot().and_then(|(_, oidc_config_query)| oidc_config_query)
-    }
-
     fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
         self.runtime.read().ok().and_then(|runtime| {
             runtime
                 .as_ref()
-                .map(|runtime| (Arc::clone(&runtime.service), runtime.oidc_config_query.as_ref().map(Arc::clone)))
+                .map(|runtime| (Arc::clone(&runtime.service), Arc::clone(&runtime.oidc_config_query)))
         })
     }
 
-    fn publish_handle(&self, service: Arc<FederatedIdentityService>) -> bool {
-        let Ok(mut published_runtime) = self.runtime.write() else {
-            return false;
-        };
-        *published_runtime = Some(FederatedIdentityRuntime {
-            service,
-            oidc_config_query: None,
-        });
-        true
-    }
-
     fn publish_runtime(&self, service: Arc<FederatedIdentityService>, oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
         let Ok(mut published_runtime) = self.runtime.write() else {
             return false;
         };
         *published_runtime = Some(FederatedIdentityRuntime {
             service,
-            oidc_config_query: Some(oidc_config_query),
+            oidc_config_query,
         });
         true
     }
@@ -639,15 +611,6 @@ pub fn publish_default_federated_identity_runtime(
     default_federated_identity_handle().publish_runtime(service, oidc_config_query)
 }
 
-/// Publish a service-only federation runtime for compatibility integrations.
-pub fn publish_default_federated_identity_service(service: Arc<FederatedIdentityService>) -> bool {
-    default_federated_identity_handle().publish_handle(service)
-}
-
-pub fn federated_identity_interface(service: Option<Arc<FederatedIdentityService>>) -> Arc<dyn FederatedIdentityInterface> {
-    Arc::new(FederatedIdentityHandle::new(service))
-}
-
 pub fn federated_identity_runtime_interface(
     runtime: Option<(Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>)>,
 ) -> Arc<dyn FederatedIdentityInterface> {
@@ -686,14 +649,13 @@ pub fn default_buffer_config_interface() -> Arc<dyn BufferConfigInterface> {
 mod tests {
     use super::{
         KmsRuntimeHandle, KmsServiceManager, ServerConfigHandle, default_federated_identity_interface,
-        federated_identity_interface, federated_identity_runtime_interface, publish_default_federated_identity_runtime,
-        runtime_sources,
+        federated_identity_runtime_interface, publish_default_federated_identity_runtime, runtime_sources,
     };
     use crate::app::context::interfaces::{KmsRuntimeInterface, ServerConfigInterface};
     use rustfs_config::server_config::Config;
     use rustfs_iam::{
         federation::{
-            FederatedIdentityRegistry, FederatedIdentityService,
+            CoreFederatedAuthorizationMapper, FederatedIdentityService,
             oidc::{OidcConfigQuery, StandardOidcAdapter},
         },
         oidc::OidcSys,
@@ -705,7 +667,8 @@ mod tests {
         let oidc = OidcSys::empty().expect("empty OIDC configuration should be valid");
         let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
         let oidc_config_query = adapter.clone();
-        let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(adapter)));
+        let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+        let service = Arc::new(adapter.into_service(mapper));
         (service, oidc_config_query)
     }
 
@@ -726,7 +689,7 @@ mod tests {
         assert!(Arc::ptr_eq(&resolved_query, &first_query));
         let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("initial federated identity runtime");
         assert!(Arc::ptr_eq(&resolved_service, &first));
-        assert!(resolved_query.as_ref().is_some_and(|query| Arc::ptr_eq(query, &first_query)));
+        assert!(Arc::ptr_eq(&resolved_query, &first_query));
 
         assert!(interface.publish_runtime(second.clone(), second_query.clone()));
         assert!(
@@ -738,36 +701,11 @@ mod tests {
         assert!(interface.oidc_config_query().is_some());
         let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("replacement federated identity runtime");
         assert!(Arc::ptr_eq(&resolved_service, &second));
-        assert!(resolved_query.as_ref().is_some_and(|query| Arc::ptr_eq(query, &second_query)));
-    }
-
-    #[test]
-    fn federated_identity_handle_preserves_service_only_compatibility() {
-        let (service, _) = test_federated_identity_runtime();
-        let (replacement, _) = test_federated_identity_runtime();
-        let interface = federated_identity_interface(Some(service.clone()));
-
-        assert!(
-            interface
-                .handle()
-                .as_ref()
-                .is_some_and(|resolved| Arc::ptr_eq(resolved, &service))
-        );
-        assert!(interface.oidc_config_query().is_none());
-        assert!(interface.publish_handle(replacement.clone()));
-        assert!(
-            interface
-                .handle()
-                .as_ref()
-                .is_some_and(|resolved| Arc::ptr_eq(resolved, &replacement))
-        );
-        assert!(interface.oidc_config_query().is_none());
-        let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("service-only federated identity runtime");
-        assert!(Arc::ptr_eq(&resolved_service, &replacement));
-        assert!(resolved_query.is_none());
+        assert!(Arc::ptr_eq(&resolved_query, &second_query));
     }
 
     #[test]
+    #[serial_test::serial(federated_identity_runtime)]
     fn default_federated_identity_handle_shares_early_publish_and_replacement() {
         let (first, first_query) = test_federated_identity_runtime();
         let (second, second_query) = test_federated_identity_runtime();

rustfs/src/app/context/interfaces.rs

@@ -48,25 +48,22 @@ pub trait IamInterface: Send + Sync {
 }
 
 /// Published federation runtime handles for authentication and OIDC configuration consumers.
-pub type FederatedIdentityRuntimeSnapshot = (Arc<FederatedIdentityService>, Option<Arc<dyn OidcConfigQuery>>);
+pub type FederatedIdentityRuntimeSnapshot = (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>);
 
 pub trait FederatedIdentityInterface: Send + Sync {
-    fn handle(&self) -> Option<Arc<FederatedIdentityService>>;
+    fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot>;
+
+    fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
+        self.runtime_snapshot().map(|(service, _)| service)
+    }
+
     fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
-        None
+        self.runtime_snapshot().map(|(_, oidc_config_query)| oidc_config_query)
     }
-    fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
-        self.handle().map(|service| (service, self.oidc_config_query()))
-    }
-    /// Compatibility publication hook for service-only implementations.
-    /// Standard OIDC runtimes publish the paired query through
-    /// [`Self::publish_runtime`].
-    fn publish_handle(&self, _service: Arc<FederatedIdentityService>) -> bool {
+
+    fn publish_runtime(&self, _service: Arc<FederatedIdentityService>, _oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
         false
     }
-    fn publish_runtime(&self, service: Arc<FederatedIdentityService>, _oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
-        self.publish_handle(service)
-    }
 }
 
 /// KMS interface for application-layer use-cases.

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -964,7 +964,11 @@ async fn copy_object_allows_new_version_for_locked_destination_but_blocks_explic
         .await
         .expect_err("explicit CopyObject overwrite of a locked version should be blocked");
 
-    assert_eq!(err.code(), &s3s::S3ErrorCode::AccessDenied);
+    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);
+    assert_eq!(
+        err.message(),
+        Some("Invalid argument: A destination versionId requires replication authorization")
+    );
     assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), dst_object).await, source_payload);
     assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), dst_object).await, 2);
 }

rustfs/src/app/object/delete.rs

@@ -511,6 +511,15 @@ impl DefaultObjectUsecase {
         // cannot flood the log.
         let mut bulk_denial_logged = false;
         for (idx, obj_id) in delete.objects.iter().enumerate() {
+            if let Err(err) = validate_object_key(&obj_id.key, "DeleteObjects") {
+                delete_results[idx].error = Some(s3s::dto::Error {
+                    code: Some(err.code().as_str().to_string()),
+                    key: Some(obj_id.key.clone()),
+                    message: Some(err.to_string()),
+                    version_id: obj_id.version_id.clone(),
+                });
+                continue;
+            }
             let raw_version_id = obj_id.version_id.clone();
             let (version_id, version_uuid) = match normalize_delete_objects_version_id(raw_version_id.clone()) {
                 Ok(parsed) => parsed,

rustfs/src/connect/diagnostics/perf_drive.rs

@@ -27,6 +27,8 @@ use std::io::{Cursor, Write as _};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+#[cfg(test)]
+use std::{cell::RefCell, mem};
 
 use base64_simd::URL_SAFE_NO_PAD;
 use p256::ecdsa::{Signature, SigningKey, signature::Signer as _};
@@ -70,6 +72,140 @@ const SCRATCH_PREFIX: &str = ".rustfs-connect-drive-";
 
 static DRIVE_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false);
 
+#[cfg(test)]
+#[derive(Clone, Copy, Debug)]
+enum DriveStage {
+    ScratchCreate,
+    Open,
+    Write,
+    Sync,
+    ReadbackVerify,
+    Cleanup,
+}
+
+#[cfg(test)]
+const MAX_DRIVE_STAGE_TIMINGS: usize = 6;
+
+#[cfg(test)]
+#[derive(Debug)]
+pub struct DriveStageTiming {
+    phase: DriveStage,
+    elapsed_micros: u64,
+    succeeded: bool,
+}
+
+#[cfg(test)]
+// Keep parallel test workers isolated; one measurement emits at most six fixed phases.
+thread_local! {
+    static DRIVE_STAGE_TIMINGS: RefCell<Vec<DriveStageTiming>> = const { RefCell::new(Vec::new()) };
+}
+
+#[cfg(test)]
+pub fn clear_drive_stage_timings() {
+    DRIVE_STAGE_TIMINGS.with(|timings| timings.borrow_mut().clear());
+}
+
+#[cfg(test)]
+pub fn take_drive_stage_timings() -> Vec<DriveStageTiming> {
+    DRIVE_STAGE_TIMINGS.with(|timings| mem::take(&mut *timings.borrow_mut()))
+}
+
+#[cfg(test)]
+struct DriveStageTimer {
+    phase: DriveStage,
+    started: Instant,
+    succeeded: bool,
+}
+
+#[cfg(test)]
+impl DriveStageTimer {
+    fn new(phase: DriveStage) -> Self {
+        Self {
+            phase,
+            started: Instant::now(),
+            succeeded: false,
+        }
+    }
+
+    fn finish(mut self, succeeded: bool) {
+        self.succeeded = succeeded;
+    }
+}
+
+#[cfg(test)]
+impl Drop for DriveStageTimer {
+    fn drop(&mut self) {
+        let timing = DriveStageTiming {
+            phase: self.phase,
+            elapsed_micros: u64::try_from(self.started.elapsed().as_micros()).unwrap_or(u64::MAX),
+            succeeded: self.succeeded,
+        };
+        DRIVE_STAGE_TIMINGS.with(|timings| {
+            let mut timings = timings.borrow_mut();
+            if timings.len() < MAX_DRIVE_STAGE_TIMINGS {
+                timings.push(timing);
+            }
+        });
+    }
+}
+
+#[cfg(test)]
+mod drive_stage_timing_tests {
+    use super::*;
+
+    #[test]
+    fn unfinished_stage_is_recorded_as_failed_and_take_clears() {
+        clear_drive_stage_timings();
+        drop(DriveStageTimer::new(DriveStage::ReadbackVerify));
+
+        let timings = take_drive_stage_timings();
+        assert_eq!(timings.len(), 1);
+        assert!(matches!(timings[0].phase, DriveStage::ReadbackVerify));
+        assert!(!timings[0].succeeded);
+        assert!(timings[0].elapsed_micros < u64::MAX);
+        assert!(take_drive_stage_timings().is_empty());
+    }
+
+    #[test]
+    fn stage_records_are_bounded_and_clear_discards_them() {
+        clear_drive_stage_timings();
+        for _ in 0..(MAX_DRIVE_STAGE_TIMINGS + 2) {
+            DriveStageTimer::new(DriveStage::Write).finish(true);
+        }
+
+        let timings = take_drive_stage_timings();
+        assert_eq!(timings.len(), MAX_DRIVE_STAGE_TIMINGS);
+        assert!(timings.iter().all(|timing| timing.succeeded));
+
+        DriveStageTimer::new(DriveStage::Cleanup).finish(true);
+        clear_drive_stage_timings();
+        assert!(take_drive_stage_timings().is_empty());
+    }
+
+    #[test]
+    fn stage_records_are_isolated_between_threads() {
+        clear_drive_stage_timings();
+        DriveStageTimer::new(DriveStage::Open).finish(true);
+
+        let child_timings = std::thread::spawn(|| {
+            assert!(take_drive_stage_timings().is_empty());
+            DriveStageTimer::new(DriveStage::Sync).finish(true);
+            take_drive_stage_timings()
+        })
+        .join()
+        .expect("stage timing thread");
+
+        assert_eq!(child_timings.len(), 1);
+        assert!(matches!(child_timings[0].phase, DriveStage::Sync));
+        assert!(child_timings[0].succeeded);
+
+        let parent_timings = take_drive_stage_timings();
+        assert_eq!(parent_timings.len(), 1);
+        assert!(matches!(parent_timings[0].phase, DriveStage::Open));
+        assert!(parent_timings[0].succeeded);
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub enum DriveOutcome {
@@ -389,13 +525,22 @@ pub async fn measure_drive(
     let _lease = CollectorLease::acquire()?;
     let started = Instant::now();
     let deadline = tokio::time::Instant::now() + request.duration;
-    let mut scratch = match ScratchGuard::create(request) {
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::ScratchCreate);
+    let created = ScratchGuard::create(request);
+    #[cfg(test)]
+    stage.finish(created.is_ok());
+    let mut scratch = match created {
         Ok(scratch) => scratch,
         Err(error) => return Ok(failed_measurement(request, started.elapsed(), classify_io(&error), 0, 0, 0)),
     };
 
     let measured = run_benchmark(request, &scratch, cancel, deadline).await;
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Cleanup);
     let cleanup = scratch.cleanup();
+    #[cfg(test)]
+    stage.finish(cleanup.is_ok());
     let elapsed = started.elapsed().min(request.duration);
     if cleanup.is_err() {
         let (read_bytes, write_bytes, io_count) = match &measured {
@@ -661,7 +806,11 @@ async fn run_benchmark(
     cancel: &CancellationToken,
     deadline: tokio::time::Instant,
 ) -> Result<BenchmarkSample, BenchmarkFailure> {
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Open);
     let std_file = scratch.open_file().map_err(BenchmarkFailure::Io)?;
+    #[cfg(test)]
+    stage.finish(true);
     let mut file = tokio::fs::File::from_std(std_file);
     let block_size = usize::try_from(request.block_bytes).map_err(|_| BenchmarkFailure::Verification)?;
     let write_buffer = vec![0xa5; block_size];
@@ -669,6 +818,8 @@ async fn run_benchmark(
     let mut write_bytes = 0_u64;
     let mut io_count = 0_u64;
     let write_started = Instant::now();
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Write);
     while remaining > 0 {
         let length = usize::try_from(remaining.min(request.block_bytes)).map_err(|_| BenchmarkFailure::Verification)?;
         cancellable(file.write_all(&write_buffer[..length]), cancel, deadline).await?;
@@ -677,8 +828,16 @@ async fn run_benchmark(
         remaining -= length as u64;
         tokio::task::yield_now().await;
     }
+    #[cfg(test)]
+    stage.finish(true);
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Sync);
     cancellable(file.sync_all(), cancel, deadline).await?;
+    #[cfg(test)]
+    stage.finish(true);
     let write_latency = write_started.elapsed();
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::ReadbackVerify);
     cancellable(file.seek(std::io::SeekFrom::Start(0)), cancel, deadline).await?;
 
     let mut read_buffer = vec![0_u8; block_size];
@@ -701,6 +860,8 @@ async fn run_benchmark(
     if read_bytes != request.scratch_bytes || write_bytes != request.scratch_bytes || io_count > MAX_OPERATIONS {
         return Err(BenchmarkFailure::Verification);
     }
+    #[cfg(test)]
+    stage.finish(true);
     Ok(BenchmarkSample {
         read_bytes,
         write_bytes,

rustfs/src/runtime_sources.rs

@@ -35,7 +35,6 @@ pub(crate) use context::{
     resolve_bucket_monitor_handle as current_bucket_monitor_handle, resolve_buffer_config as current_buffer_config,
     resolve_daily_tier_stats as current_daily_tier_stats, resolve_deployment_id as current_deployment_id,
     resolve_encryption_service as current_encryption_service, resolve_endpoints_handle as current_endpoints_handle,
-    resolve_federated_identity_runtime as current_federated_identity_runtime,
     resolve_federated_identity_service as current_federated_identity_service, resolve_iam_handle as current_iam_handle,
     resolve_iam_ready as current_iam_ready, resolve_internode_metrics as current_internode_metrics,
     resolve_kms_runtime_service_manager as current_kms_runtime_service_manager,
@@ -47,6 +46,7 @@ pub(crate) use context::{
     resolve_object_data_cache_handle_for_context as current_object_data_cache_handle_for_context,
     resolve_object_store_handle as current_object_store_handle,
     resolve_object_store_handle_for_context as current_object_store_handle_for_context,
+    resolve_oidc_config_query as current_oidc_config_query,
     resolve_or_init_kms_runtime_service_manager as current_or_init_kms_runtime_service_manager,
     resolve_outbound_tls_generation as current_outbound_tls_generation, resolve_outbound_tls_state as current_outbound_tls_state,
     resolve_performance_metrics as current_performance_metrics, resolve_ready_iam_handle as current_ready_iam_handle,
@@ -63,13 +63,13 @@ pub(crate) fn publish_federated_identity_runtime(
     service: Arc<FederatedIdentityService>,
     oidc_config_query: Arc<dyn OidcConfigQuery>,
 ) -> bool {
-    warn_on_browser_redirect_fallback(&service);
+    warn_on_browser_redirect_fallback(service.has_providers());
     context::publish_federated_identity_runtime(service, oidc_config_query)
 }
 
-fn warn_on_browser_redirect_fallback(service: &FederatedIdentityService) {
+fn warn_on_browser_redirect_fallback(has_providers: bool) {
     let browser_redirect_url = rustfs_utils::get_env_opt_str(ENV_RUSTFS_BROWSER_REDIRECT_URL);
-    if service.has_providers() && browser_redirect_url.as_deref().is_none_or(|value| value.trim().is_empty()) {
+    if has_providers && browser_redirect_url.as_deref().is_none_or(|value| value.trim().is_empty()) {
         warn!(
             event = EVENT_OIDC_BROWSER_REDIRECT_FALLBACK,
             component = LOG_COMPONENT_MAIN,
@@ -103,76 +103,19 @@ pub(crate) fn current_app_context() -> Option<Arc<AppContext>> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use rustfs_iam::federation::oidc::OidcConfigQuery;
-    use rustfs_iam::federation::{
-        FederatedAuthorization, FederatedCodeExchange, FederatedIdentityProvider, FederatedIdentityRegistry,
-        FederatedProviderView, FederatedRedirectPolicy, Result as FederationResult,
+    use rustfs_iam::{
+        federation::{
+            CoreFederatedAuthorizationMapper,
+            oidc::{OidcConfigQuery, StandardOidcAdapter},
+        },
+        oidc::{OidcSys, make_test_sys, test_config},
     };
-    use rustfs_iam::oidc::OidcConfigSnapshot;
     use std::{
         io::{self, Write},
         sync::Mutex,
     };
     use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
 
-    struct TestProvider {
-        has_providers: bool,
-    }
-
-    #[async_trait::async_trait]
-    impl FederatedIdentityProvider for TestProvider {
-        fn has_providers(&self) -> bool {
-            self.has_providers
-        }
-
-        fn list_providers(&self) -> Vec<FederatedProviderView> {
-            Vec::new()
-        }
-
-        fn list_visible_providers(&self) -> Vec<FederatedProviderView> {
-            Vec::new()
-        }
-
-        fn redirect_policy(&self, _provider_id: &str) -> Option<FederatedRedirectPolicy> {
-            None
-        }
-
-        async fn authorize_url(
-            &self,
-            _provider_id: &str,
-            _redirect_uri: &str,
-            _redirect_after: Option<String>,
-        ) -> FederationResult<String> {
-            unreachable!("startup publication test does not authorize")
-        }
-
-        async fn exchange_code(&self, _state: &str, _code: &str, _redirect_uri: &str) -> FederationResult<FederatedCodeExchange> {
-            unreachable!("startup publication test does not exchange codes")
-        }
-
-        async fn verify_web_identity_token(&self, _jwt: &str) -> FederationResult<FederatedAuthorization> {
-            unreachable!("startup publication test does not verify tokens")
-        }
-
-        async fn create_logout_token(&self, _provider_id: &str, _id_token: &str) -> FederationResult<String> {
-            unreachable!("startup publication test does not create logout tokens")
-        }
-
-        async fn build_logout_url(
-            &self,
-            _logout_token: &str,
-            _post_logout_redirect_uri: &str,
-        ) -> FederationResult<Option<String>> {
-            unreachable!("startup publication test does not build logout URLs")
-        }
-    }
-
-    impl OidcConfigQuery for TestProvider {
-        fn config_snapshot(&self) -> OidcConfigSnapshot {
-            OidcConfigSnapshot::new(Vec::new())
-        }
-    }
-
     #[derive(Clone, Default)]
     struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
 
@@ -197,14 +140,14 @@ mod tests {
         }
     }
 
-    fn test_runtime(has_providers: bool) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
-        let provider = Arc::new(TestProvider { has_providers });
-        let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(provider.clone())));
-        (service, provider)
-    }
-
-    fn capture_startup_publication(has_providers: bool, browser_redirect_url: Option<&str>) -> String {
+    fn capture_startup_publication(oidc: OidcSys, browser_redirect_url: Option<&str>) -> String {
         temp_env::with_var(ENV_RUSTFS_BROWSER_REDIRECT_URL, browser_redirect_url, || {
+            let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
+            let oidc_config_query: Arc<dyn OidcConfigQuery> = adapter.clone();
+            let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+            let service = Arc::new(adapter.into_service(mapper));
+            let expected_service = service.clone();
+            let expected_oidc_config_query = oidc_config_query.clone();
             let logs = CapturedLogs::default();
             let captured = logs.0.clone();
             let subscriber = Registry::default().with(
@@ -217,8 +160,12 @@ mod tests {
             );
 
             tracing::subscriber::with_default(subscriber, || {
-                let (service, oidc_config_query) = test_runtime(has_providers);
                 assert!(publish_federated_identity_runtime(service, oidc_config_query));
+                let (published_service, published_oidc_config_query) = context::default_federated_identity_interface()
+                    .runtime_snapshot()
+                    .expect("published federated identity runtime should resolve");
+                assert!(Arc::ptr_eq(&published_service, &expected_service));
+                assert!(Arc::ptr_eq(&published_oidc_config_query, &expected_oidc_config_query));
             });
 
             String::from_utf8(captured.lock().expect("captured log lock").clone()).expect("captured logs must be UTF-8")
@@ -226,16 +173,20 @@ mod tests {
     }
 
     #[test]
+    #[serial_test::serial(federated_identity_runtime)]
     fn oidc_startup_publication_warns_only_for_request_header_fallback() {
-        let output = capture_startup_publication(true, None);
+        let output = capture_startup_publication(make_test_sys(vec![test_config("default")]), None);
 
         assert!(output.contains("WARN"), "{output}");
         assert!(output.contains(EVENT_OIDC_BROWSER_REDIRECT_FALLBACK), "{output}");
         assert!(output.contains(ENV_RUSTFS_BROWSER_REDIRECT_URL), "{output}");
         assert!(output.contains("request_host_and_forwarded_proto"), "{output}");
         assert!(output.contains("trusted_console_origin_not_configured"), "{output}");
-        assert!(capture_startup_publication(false, None).is_empty());
-        assert!(capture_startup_publication(true, Some("https://console.example.com")).is_empty());
-        assert!(!capture_startup_publication(true, Some("  ")).is_empty());
+        assert!(capture_startup_publication(OidcSys::empty().expect("empty OIDC system should initialize"), None).is_empty());
+        assert!(
+            capture_startup_publication(make_test_sys(vec![test_config("default")]), Some("https://console.example.com"),)
+                .is_empty()
+        );
+        assert!(!capture_startup_publication(make_test_sys(vec![test_config("default")]), Some("  ")).is_empty());
     }
 }

rustfs/src/storage/access.rs

@@ -140,6 +140,9 @@ struct InternalObjectAuthorization;
 #[derive(Clone, Debug)]
 struct RequestedObjectLockRetainUntil(OffsetDateTime);
 
+#[derive(Clone, Debug)]
+struct RequestedObjectLockRetention(ObjectLockRetention);
+
 #[derive(Clone, Debug)]
 struct StagedMultipartPartAuthorization;
 
@@ -885,6 +888,21 @@ fn authorization_conditions<T>(
     }
     merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions);
     merge_request_object_tag_conditions(action, &req.headers, &mut conditions)?;
+    if let Some(RequestedObjectLockRetention(retention)) = req.extensions.get::<RequestedObjectLockRetention>() {
+        // Standalone retention writes apply the XML body, so headers must not
+        // supply a different value to the policy decision.
+        conditions.remove("object-lock-mode");
+        conditions.remove("object-lock-retain-until-date");
+        if let Some(mode) = &retention.mode {
+            conditions.insert("object-lock-mode".to_string(), vec![mode.as_str().to_string()]);
+        }
+        if let Some(date) = &retention.retain_until_date {
+            let date = OffsetDateTime::from(date.clone())
+                .format(&time::format_description::well_known::Rfc3339)
+                .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid retain-until date"))?;
+            conditions.insert("object-lock-retain-until-date".to_string(), vec![date]);
+        }
+    }
     merge_object_lock_remaining_retention_days_condition(
         req.extensions.get::<RequestedObjectLockRetainUntil>(),
         OffsetDateTime::now_utc(),
@@ -908,6 +926,12 @@ pub(crate) fn set_requested_object_lock_retain_until<T>(req: &mut S3Request<T>,
     }
 }
 
+fn set_requested_object_lock_retention(req: &mut S3Request<PutObjectRetentionInput>) {
+    let retention = req.input.retention.clone().unwrap_or_default();
+    set_requested_object_lock_retain_until(req, retention.retain_until_date.as_ref());
+    req.extensions.insert(RequestedObjectLockRetention(retention));
+}
+
 fn merge_object_lock_remaining_retention_days_condition(
     requested: Option<&RequestedObjectLockRetainUntil>,
     now: OffsetDateTime,
@@ -1123,11 +1147,8 @@ pub(crate) fn delete_object_authorize_action(version_id: Option<&str>) -> Action
 /// conflated at the authorization boundary.
 ///
 /// Note: `ActionSet::is_match` still maps a `s3:GetObjectVersion` grant onto a
-/// `s3:GetObject` request. That mapping is intentionally left in place until the
-/// remaining version-aware read paths (HeadObject, GetObjectAcl, tagging) get the
-/// same treatment — see the GHSA-3ppv follow-up audit. It does not re-open this
-/// disclosure: it only broadens a Version grant toward current reads, never the
-/// reverse.
+/// `s3:GetObject` request. It only broadens a Version grant toward current
+/// reads, never the reverse.
 fn versioned_read_action(version_id: Option<&str>) -> Action {
     if version_id.is_some() {
         Action::S3Action(S3Action::GetObjectVersionAction)
@@ -1136,6 +1157,20 @@ fn versioned_read_action(version_id: Option<&str>) -> Action {
     }
 }
 
+fn versioned_tagging_action(action: S3Action, version_id: Option<&str>) -> Action {
+    let action = if version_id.is_some() {
+        match action {
+            S3Action::GetObjectTaggingAction => S3Action::GetObjectVersionTaggingAction,
+            S3Action::PutObjectTaggingAction => S3Action::PutObjectVersionTaggingAction,
+            S3Action::DeleteObjectTaggingAction => S3Action::DeleteObjectVersionTaggingAction,
+            action => action,
+        }
+    } else {
+        action
+    };
+    Action::S3Action(action)
+}
+
 async fn get_or_fetch_object_tag_conditions<T>(
     req: &mut S3Request<T>,
     store: &ECStore,
@@ -2560,7 +2595,11 @@ impl S3Access for FS {
         req_info.object = Some(req.input.key.clone());
         req_info.version_id = req.input.version_id.clone();
 
-        authorize_request(req, Action::S3Action(S3Action::DeleteObjectTaggingAction)).await
+        authorize_request(
+            req,
+            versioned_tagging_action(S3Action::DeleteObjectTaggingAction, req.input.version_id.as_deref()),
+        )
+        .await
     }
 
     /// Checks whether the DeleteObjects request has accesses to the resources.
@@ -2881,7 +2920,11 @@ impl S3Access for FS {
         req_info.object = Some(req.input.key.clone());
         req_info.version_id = req.input.version_id.clone();
 
-        authorize_request(req, Action::S3Action(S3Action::GetObjectTaggingAction)).await
+        authorize_request(
+            req,
+            versioned_tagging_action(S3Action::GetObjectTaggingAction, req.input.version_id.as_deref()),
+        )
+        .await
     }
 
     /// Checks whether the GetObjectTorrent request has accesses to the resources.
@@ -2933,7 +2976,7 @@ impl S3Access for FS {
             return Ok(());
         }
 
-        authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await?;
+        authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await?;
         req.extensions.insert(source_generation);
         Ok(())
     }
@@ -3336,12 +3379,7 @@ impl S3Access for FS {
 
         let bucket = req.input.bucket.clone();
         let bucket_generation = load_bucket_generation(self, req, &bucket).await;
-        let requested_retain_until = req
-            .input
-            .retention
-            .as_ref()
-            .and_then(|retention| retention.retain_until_date.clone());
-        set_requested_object_lock_retain_until(req, requested_retain_until.as_ref());
+        set_requested_object_lock_retention(req);
         authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
 
         // S3 Standard: When bypass_governance header is set, must have s3:BypassGovernanceRetention permission
@@ -3362,7 +3400,11 @@ impl S3Access for FS {
         req_info.object = Some(req.input.key.clone());
         req_info.version_id = req.input.version_id.clone();
 
-        authorize_request(req, Action::S3Action(S3Action::PutObjectTaggingAction)).await
+        authorize_request(
+            req,
+            versioned_tagging_action(S3Action::PutObjectTaggingAction, req.input.version_id.as_deref()),
+        )
+        .await
     }
 
     /// Checks whether the PutPublicAccessBlock request has accesses to the resources.
@@ -3481,9 +3523,9 @@ mod tests {
         owner_can_bypass_policy_deny, post_object_authorize_action, put_bucket_policy_authorize_action, request_context_from_req,
         request_object_store, require_owned_reserved_table_object, retention_write_requested, table_data_plane_admin_action,
         table_data_plane_content_mutation, table_data_plane_resource_for_request, table_publication_guard_error,
-        validate_post_object_success_controls, versioned_read_action,
+        validate_post_object_success_controls, versioned_read_action, versioned_tagging_action,
     };
-    use super::{remaining_retention_days, set_requested_object_lock_retain_until};
+    use super::{remaining_retention_days, set_requested_object_lock_retain_until, set_requested_object_lock_retention};
     use crate::auth::OBJECT_LOCK_REMAINING_RETENTION_DAYS_CONDITION;
     use crate::error::ApiError;
     use crate::storage::storage_api::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
@@ -4285,11 +4327,109 @@ mod tests {
         );
     }
 
-    fn remaining_retention_days_conditions(req: &S3Request<()>, action: Action) -> HashMap<String, Vec<String>> {
+    #[tokio::test]
+    async fn versioned_tagging_requires_version_permissions() {
+        use rustfs_policy::policy::{Args, Policy};
+        let policy = Policy::parse_config(br#"{
+            "Version":"2012-10-17",
+            "Statement":[{"Effect":"Allow","Action":["s3:GetObjectTagging","s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":"arn:aws:s3:::bucket/*"}]
+        }"#).expect("current-object tagging policy should parse");
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let groups = None;
+        for (current, versioned) in [
+            (S3Action::GetObjectTaggingAction, S3Action::GetObjectVersionTaggingAction),
+            (S3Action::PutObjectTaggingAction, S3Action::PutObjectVersionTaggingAction),
+            (S3Action::DeleteObjectTaggingAction, S3Action::DeleteObjectVersionTaggingAction),
+        ] {
+            for version in [None, Some("null"), Some("0194e0f1-0000-7000-8000-000000000000")] {
+                let action = versioned_tagging_action(current, version);
+                assert_eq!(action, Action::S3Action(if version.is_some() { versioned } else { current }));
+                let args = Args {
+                    account: "writer",
+                    groups: &groups,
+                    action,
+                    bucket: "bucket",
+                    conditions: &conditions,
+                    is_owner: false,
+                    object: "object",
+                    claims: &claims,
+                    deny_only: false,
+                };
+                assert_eq!(policy.is_allowed(&args).await, version.is_none(), "version selector {version:?}");
+            }
+        }
+    }
+
+    fn remaining_retention_days_conditions<T>(req: &S3Request<T>, action: Action) -> HashMap<String, Vec<String>> {
         let credentials = rustfs_credentials::Credentials::default();
         authorization_conditions(req, &credentials, None, None, None, None, action).expect("conditions should build")
     }
 
+    #[tokio::test]
+    async fn retention_policy_checks_body_instead_of_conflicting_headers() {
+        use rustfs_policy::policy::{Args, Policy};
+        let mut req = build_request(
+            PutObjectRetentionInput {
+                retention: Some(ObjectLockRetention {
+                    mode: Some(ObjectLockRetentionMode::from_static("COMPLIANCE")),
+                    retain_until_date: Some(
+                        Timestamp::parse(TimestampFormat::DateTime, "2035-01-02T00:00:00+23:59").expect("body date should parse"),
+                    ),
+                }),
+                ..Default::default()
+            },
+            Method::PUT,
+        );
+        req.headers
+            .insert("x-amz-object-lock-mode", HeaderValue::from_static("GOVERNANCE"));
+        req.headers
+            .insert("x-amz-object-lock-retain-until-date", HeaderValue::from_static("2029-01-01T00:00:00Z"));
+        set_requested_object_lock_retention(&mut req);
+        let action = Action::S3Action(S3Action::PutObjectRetentionAction);
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-mode"), Some(&vec!["COMPLIANCE".to_string()]));
+        let applied = OffsetDateTime::from(
+            req.input
+                .retention
+                .as_ref()
+                .expect("body retention")
+                .retain_until_date
+                .clone()
+                .expect("body date"),
+        );
+        let authorized = OffsetDateTime::parse(
+            &conditions["object-lock-retain-until-date"][0],
+            &time::format_description::well_known::Rfc3339,
+        )
+        .expect("condition date should parse");
+        assert_eq!(authorized.unix_timestamp(), applied.unix_timestamp());
+        let policy = Policy::parse_config(br#"{
+            "Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:PutObjectRetention","Resource":"arn:aws:s3:::bucket/*",
+            "Condition":{"DateLessThan":{"s3:object-lock-retain-until-date":"2030-01-01T00:00:00Z"}}}]
+        }"#).expect("bounded retention policy should parse");
+        let args = Args {
+            account: "writer",
+            groups: &None,
+            action,
+            bucket: "bucket",
+            conditions: &conditions,
+            is_owner: false,
+            object: "object",
+            claims: &HashMap::new(),
+            deny_only: false,
+        };
+        assert!(
+            !policy.is_allowed(&args).await,
+            "an in-range header cannot authorize an out-of-range body"
+        );
+        req.input.retention = None;
+        set_requested_object_lock_retention(&mut req);
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert!(!conditions.contains_key("object-lock-mode"));
+        assert!(!conditions.contains_key("object-lock-retain-until-date"));
+    }
+
     fn request_retaining_for(retain_for: time::Duration) -> S3Request<()> {
         let mut req = build_request((), Method::PUT);
         let retain_until = OffsetDateTime::now_utc() + retain_for;

rustfs/src/storage/ecfs.rs

@@ -402,6 +402,30 @@ async fn validate_table_catalog_object_mutation(bucket: &str, key: &str) -> S3Re
 const MAXIMUM_RETENTION_DAYS: i32 = 36_500;
 const MAXIMUM_RETENTION_YEARS: i32 = 100;
 
+pub(crate) fn object_lock_retention_from_metadata(metadata: &HashMap<String, String>) -> S3Result<ObjectLockRetention> {
+    let mode = metadata
+        .get(AMZ_OBJECT_LOCK_MODE_LOWER)
+        .map(|value| ObjectLockRetentionMode::from(value.as_str().to_string()));
+    let retain_until_date = metadata
+        .get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)
+        .and_then(|value| OffsetDateTime::parse(value, &Rfc3339).ok())
+        // The S3 timestamp serializer emits a literal Z, so its wall clock
+        // must be UTC before it is serialized.
+        .map(|date| {
+            date.checked_to_offset(time::UtcOffset::UTC)
+                .filter(|date| (0..=9999).contains(&date.year()))
+                .map(Timestamp::from)
+                .ok_or_else(|| {
+                    S3Error::with_message(
+                        S3ErrorCode::InternalError,
+                        "Stored retention date is outside the supported UTC range".to_string(),
+                    )
+                })
+        })
+        .transpose()?;
+    Ok(ObjectLockRetention { mode, retain_until_date })
+}
+
 fn invalid_object_lock_configuration(message: impl Into<String>) -> S3Error {
     S3Error::with_message(S3ErrorCode::MalformedXML, message.into())
 }
@@ -1112,19 +1136,8 @@ impl S3 for FS {
             s3_error!(InternalError, "{}", e.to_string())
         })?;
 
-        let mode = object_info
-            .user_defined
-            .get(AMZ_OBJECT_LOCK_MODE_LOWER)
-            .map(|v| ObjectLockRetentionMode::from(v.as_str().to_string()));
-
-        let retain_until_date = object_info
-            .user_defined
-            .get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)
-            .and_then(|v| OffsetDateTime::parse(v.as_str(), &Rfc3339).ok())
-            .map(Timestamp::from);
-
         let output = GetObjectRetentionOutput {
-            retention: Some(ObjectLockRetention { mode, retain_until_date }),
+            retention: Some(object_lock_retention_from_metadata(&object_info.user_defined)?),
         };
         let version_id = req.input.version_id.clone().unwrap_or_default();
         helper = helper.object(object_info).version_id(version_id);

rustfs/src/storage/ecfs_extend.rs

@@ -340,6 +340,10 @@ pub(crate) fn validate_object_key(key: &str, operation: &str) -> S3Result<()> {
         ));
     }
 
+    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"));
+    }
+
     // Log debug info for keys with special characters to help diagnose encoding issues
     if key.contains([' ', '+', '%']) {
         debug!(

rustfs/src/storage/ecfs_test.rs

@@ -18,7 +18,8 @@ mod tests {
     use crate::server::cors;
     use crate::storage::StorageError;
     use crate::storage::ecfs::{
-        FS, parse_object_version_id, propagate_object_lock_peer_reload, validate_object_lock_configuration_input,
+        FS, object_lock_retention_from_metadata, parse_object_version_id, propagate_object_lock_peer_reload,
+        validate_object_lock_configuration_input,
     };
     use crate::storage::ecfs_extend::{apply_bucket_default_lock_retention, map_bucket_object_lock_config_state};
     use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner};
@@ -988,6 +989,44 @@ mod tests {
         assert!(validate_list_object_unordered_with_delimiter(delimiter_some, complex_query_without_unordered).is_ok());
     }
 
+    #[test]
+    fn retention_response_serialization_preserves_offset_metadata_instant() {
+        for value in [
+            "2035-01-02T00:00:00+23:59",
+            "2035-01-02T00:00:00-23:59",
+            "2035-01-02T00:00:00Z",
+        ] {
+            let metadata = HashMap::from([
+                (AMZ_OBJECT_LOCK_MODE_LOWER.to_owned(), "COMPLIANCE".to_owned()),
+                (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_owned(), value.to_owned()),
+            ]);
+            let retention = object_lock_retention_from_metadata(&metadata).expect("representable retention instant");
+            assert_eq!(retention.mode.as_ref().expect("mode").as_str(), "COMPLIANCE");
+            let mut serialized = Vec::new();
+            retention
+                .retain_until_date
+                .expect("stored retention date")
+                .format(s3s::dto::TimestampFormat::DateTime, &mut serialized)
+                .expect("serialize the response timestamp");
+            let serialized = String::from_utf8(serialized).expect("UTF-8 timestamp");
+            let returned = OffsetDateTime::parse(&serialized, &Rfc3339).expect("parse wire timestamp");
+            let stored = OffsetDateTime::parse(value, &Rfc3339).expect("parse stored timestamp");
+            assert_eq!(returned.unix_timestamp(), stored.unix_timestamp(), "stored date {value}");
+        }
+    }
+
+    #[test]
+    fn retention_response_rejects_instants_outside_the_utc_wire_range() {
+        for value in ["9999-12-31T23:59:59-23:59", "0000-01-01T00:00:00+23:59"] {
+            let metadata = HashMap::from([
+                (AMZ_OBJECT_LOCK_MODE_LOWER.to_owned(), "COMPLIANCE".to_owned()),
+                (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_owned(), value.to_owned()),
+            ]);
+            let error = object_lock_retention_from_metadata(&metadata).expect_err("unrepresentable UTC timestamp");
+            assert_eq!(error.code(), &S3ErrorCode::InternalError);
+        }
+    }
+
     #[test]
     fn test_parse_object_lock_retention() {
         use time::macros::datetime;

rustfs/src/storage/options.rs

@@ -368,6 +368,13 @@ pub async fn put_opts_with_replication_authorization(
     metadata: HashMap<String, String>,
     replication_request_authorized: bool,
 ) -> Result<ObjectOptions> {
+    if vid.is_some() && !replication_request_authorized {
+        return Err(StorageError::InvalidArgument(
+            bucket.to_owned(),
+            object.to_owned(),
+            "A destination versionId requires replication authorization".to_owned(),
+        ));
+    }
     let versioning_cfg = bucket_versioning_config_for_write(bucket).await?;
     let versioned = versioning_cfg.prefix_enabled(object);
     let version_suspended = versioning_cfg.prefix_suspended(object);
@@ -1532,7 +1539,15 @@ mod tests {
         let headers = create_test_headers();
         let invalid_uuid = "invalid-uuid".to_string();
 
-        let result = put_opts("test-bucket", "test-object", Some(invalid_uuid), &headers, HashMap::new()).await;
+        let result = put_opts_with_replication_authorization(
+            "test-bucket",
+            "test-object",
+            Some(invalid_uuid),
+            &headers,
+            HashMap::new(),
+            true,
+        )
+        .await;
 
         assert!(result.is_err());
         if let Err(err) = result {
@@ -1556,13 +1571,35 @@ mod tests {
         // get_opts / del_opts already do.
         let headers = create_test_headers();
 
-        let opts = put_opts("test-bucket", "test-object", Some("null".to_string()), &headers, HashMap::new())
-            .await
-            .expect("PUT with versionId=null must be accepted as the null version");
+        let opts = put_opts_with_replication_authorization(
+            "test-bucket",
+            "test-object",
+            Some("null".to_string()),
+            &headers,
+            HashMap::new(),
+            true,
+        )
+        .await
+        .expect("PUT with versionId=null must be accepted as the null version");
 
         assert_eq!(opts.version_id, Some(Uuid::nil().to_string()));
     }
 
+    #[tokio::test]
+    async fn ordinary_writes_cannot_choose_a_destination_version() {
+        let headers = create_test_headers();
+        for version in ["null", "", "0194e0f1-0000-7000-8000-000000000000"] {
+            let err = put_opts("test-bucket", "test-object", Some(version.to_owned()), &headers, HashMap::new())
+                .await
+                .expect_err("ordinary PUT must not overwrite a selected version");
+            assert!(matches!(err, StorageError::InvalidArgument(_, _, _)));
+            let err = copy_dst_opts("test-bucket", "test-object", Some(version.to_owned()), &headers, HashMap::new())
+                .await
+                .expect_err("ordinary COPY must not overwrite a selected version");
+            assert!(matches!(err, StorageError::InvalidArgument(_, _, _)));
+        }
+    }
+
     #[tokio::test]
     async fn test_copy_dst_opts() {
         let headers = create_test_headers();

rustfs/src/storage/s3_api/bucket.rs

@@ -226,7 +226,11 @@ pub(crate) fn build_list_object_versions_output(
             size: Some(v.size),
             version_id: Some(list_versions_response_version_id(v.version_id)),
             is_latest: Some(v.is_latest),
-            e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
+            e_tag: v
+                .etag
+                .as_ref()
+                .filter(|_| rustfs_utils::http::etag_can_be_listed(&v.user_defined))
+                .map(|etag| to_s3s_etag(etag)),
             storage_class: v.storage_class.clone().map(ObjectVersionStorageClass::from),
             ..Default::default()
         })
@@ -311,7 +315,11 @@ pub(crate) fn build_list_objects_v2_output(
                 // logical-size sentinel; never expose that internal value in
                 // an S3 response.
                 size: Some(v.get_actual_size_or_physical()),
-                e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
+                e_tag: v
+                    .etag
+                    .as_ref()
+                    .filter(|_| rustfs_utils::http::etag_can_be_listed(&v.user_defined))
+                    .map(|etag| to_s3s_etag(etag)),
                 storage_class: v.storage_class.clone().map(ObjectStorageClass::from),
                 ..Default::default()
             };
@@ -709,6 +717,83 @@ mod tests {
         );
     }
 
+    #[test]
+    fn standard_list_responses_hide_legacy_encrypted_fingerprints() {
+        let mut protected_metadata = std::collections::HashMap::from([(
+            "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+            "AES256".to_owned(),
+        )]);
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut protected_metadata,
+            rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+            "v1".to_owned(),
+        );
+        let objects: Vec<ObjectInfo> = [
+            std::collections::HashMap::from([(
+                "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+                "AES256".to_owned(),
+            )]),
+            std::collections::HashMap::from([("x-amz-server-side-encryption".to_owned(), "aws:kms".to_owned())]),
+            protected_metadata,
+            std::collections::HashMap::new(),
+        ]
+        .into_iter()
+        .enumerate()
+        .map(|(index, metadata)| ObjectInfo {
+            name: format!("object-{index}"),
+            etag: Some("0123456789abcdef0123456789abcdef".to_owned()),
+            user_defined: std::sync::Arc::new(metadata),
+            size: 42,
+            ..Default::default()
+        })
+        .collect();
+        let v2 = build_list_objects_v2_output(
+            ListObjectsV2Info {
+                objects: objects.clone(),
+                ..Default::default()
+            },
+            false,
+            1000,
+            "bucket".to_owned(),
+            String::new(),
+            None,
+            None,
+            None,
+            None,
+        );
+        let v1 = build_list_objects_output(v2.clone(), None);
+        for listing in [&v1.contents, &v2.contents] {
+            let entries = listing.as_ref().expect("listed objects");
+            assert_eq!(entries.len(), 4);
+            for (index, entry) in entries.iter().enumerate() {
+                assert_eq!(entry.e_tag.is_some(), index >= 2);
+                assert_eq!(entry.key.as_deref(), Some(format!("object-{index}").as_str()));
+                assert_eq!(entry.size, Some(42));
+            }
+        }
+        let versions = build_list_object_versions_output(
+            ListObjectVersionsInfo {
+                objects,
+                ..Default::default()
+            },
+            "bucket".to_owned(),
+            &ListObjectVersionsParams {
+                prefix: String::new(),
+                delimiter: None,
+                key_marker: None,
+                version_id_marker: None,
+                max_keys: 1000,
+            },
+            None,
+        );
+        let versions = versions.versions.expect("listed versions");
+        assert_eq!(versions.len(), 4);
+        for (index, entry) in versions.iter().enumerate() {
+            assert_eq!(entry.e_tag.is_some(), index >= 2);
+            assert_eq!(entry.size, Some(42));
+        }
+    }
+
     #[test]
     fn list_responses_report_standard_for_legacy_label_only_file_metadata() {
         let version_id = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("fixture version ID should be valid");

rustfs/src/storage/sse.rs

@@ -4127,6 +4127,7 @@ pub fn strip_managed_encryption_metadata(metadata: &mut HashMap<String, String>)
     for key in KEYS.iter() {
         metadata.remove(*key);
     }
+    rustfs_utils::http::metadata_compat::remove_str(metadata, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG);
 }
 
 pub fn mark_encrypted_multipart_metadata(metadata: &mut HashMap<String, String>) {

rustfs/tests/connect_perf_drive.rs

@@ -81,14 +81,16 @@ async fn real_local_write_read_is_measured_and_scratch_is_removed() {
     let _guard = TEST_LOCK.lock().await;
     let root = tempfile::tempdir().expect("scratch root");
     let request = request(root.path());
+    perf_drive::clear_drive_stage_timings();
     let measurement = measure_drive(&request, &CancellationToken::new())
         .await
         .expect("real drive benchmark");
+    let stage_timings = perf_drive::take_drive_stage_timings();
 
     assert_eq!(
         measurement.result.outcome(),
         DriveOutcome::Succeeded,
-        "drive measurement: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}",
+        "drive measurement: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}, stage_timings={stage_timings:?}",
         measurement.result.reason_code(),
         measurement.target.reason_code,
         measurement.target.duration_millis,
@@ -366,13 +368,15 @@ async fn successful_result_has_signed_bounded_private_offline_export() {
     let _guard = TEST_LOCK.lock().await;
     let root = tempfile::tempdir().expect("scratch root");
     let request = request(root.path());
+    perf_drive::clear_drive_stage_timings();
     let measurement = measure_drive(&request, &CancellationToken::new())
         .await
         .expect("drive measurement");
+    let stage_timings = perf_drive::take_drive_stage_timings();
     assert_eq!(
         measurement.result.outcome(),
         DriveOutcome::Succeeded,
-        "drive measurement before signed export: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}",
+        "drive measurement before signed export: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}, stage_timings={stage_timings:?}",
         measurement.result.reason_code(),
         measurement.target.reason_code,
         measurement.target.duration_millis,