fix(kms): version local key records safely

A same-name CopyObject marks the operation `metadata_only`, which lets the
store layer rewrite `xl.meta` in place and leave the data blocks untouched.
The handler independently strips the source encryption metadata and calls
`sse_encryption`, which mints a *fresh* DEK. On an unversioned bucket both
happen at once, so the object ends up with a new DEK sitting beside ciphertext
sealed under the old one, and can never be decrypted again.

The mirror case is silent: an encrypted source copied without any destination
SSE keeps its ciphertext while losing the key metadata, so GET returns raw
ciphertext as if it were plaintext, with HTTP 200 and no error anywhere.

Keep `metadata_only` off whenever either side of the copy is encrypted, so the
store layer performs a full read/write rewrite through `put_object`. This is
the same resolution the versioned historical-restore path already uses for
this risk (issue #4238), and it matches MinIO's
`isSourceEncrypted || isTargetEncrypted -> metadataOnly = false` guard in
CopyObjectHandler.

The target half of the predicate deliberately tests `effective_sse` rather
than the request headers MinIO inspects: `effective_sse` also resolves the
bucket default-encryption rule, and `sse_encryption` mints a DEK from that
resolved value. A header-only check would miss a same-key copy performed under
a bucket default rule. The source half reuses `ObjectInfo::is_encrypted` so a
future encryption flavour is covered here as soon as it is recognised there.

Versioned buckets were already safe: that path falls through to `put_object`
regardless of `metadata_only`. RestoreObject also sets `metadata_only` but
only appends restore keys and never re-derives a DEK, so it is unaffected.

.config/make/lint-fmt.mak

@@ -60,6 +60,11 @@ body-cache-whitelist-check: ## Check the body-cache eligibility gate stays a fai
 	@echo "🧱 Checking body-cache whitelist guard..."
 	./scripts/check_body_cache_whitelist.sh
 
+.PHONY: fips-wording-check
+fips-wording-check: ## Check outward docs do not make unsupported FIPS claims
+	@echo "📣 Checking FIPS wording guard..."
+	./scripts/check_fips_wording.sh
+
 .PHONY: log-analyzer-rules-check
 log-analyzer-rules-check: core-deps ## Check log-analyzer rule anchors still exist verbatim in source
 	@echo "🩺 Checking log-analyzer rule anchors..."

.config/make/pre-commit.mak

@@ -19,13 +19,13 @@ planning-docs-check: ## Check that no planning-type documents are committed
 	./scripts/check_no_planning_docs.sh
 
 .PHONY: pre-commit
-pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
+pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check fips-wording-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
 	@echo "✅ All pre-commit checks passed!"
 
 .PHONY: pre-pr
-pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check doc-paths-check planning-docs-check log-analyzer-rules-check clippy-check test ## Run full pre-PR checks with clippy and tests
+pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check fips-wording-check doc-paths-check planning-docs-check log-analyzer-rules-check clippy-check test ## Run full pre-PR checks with clippy and tests
 	@echo "✅ All pre-PR checks passed!"
 
 .PHONY: dev-check
-dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check doc-paths-check planning-docs-check quick-check ## Run fast local development checks
+dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check fips-wording-check doc-paths-check planning-docs-check quick-check ## Run fast local development checks
 	@echo "✅ Fast development checks passed!"

AGENTS.md

@@ -347,6 +347,11 @@ cargo run -p rustfs-filemeta --example dump_fileinfo -- "/path/to/file/xl.meta"
   absent, empty, and nil all mean "no value", never `Uuid::nil()`.
 - A remote-tier version of `None`/`""` means the tier bucket is unversioned:
   send **no** `versionId` on tier GET/DELETE.
+- Structs persisted in the scanner data-usage cache (`DataUsageCacheInfo`,
+  `DataUsageEntry`) carry a hand-written map-encoded `Serialize`. MessagePack
+  encodes derived structs as arrays, where an appended field makes the whole
+  cache a decode error for older readers — keep new fields `#[serde(default)]`
+  and keep the map encoding rather than reverting to `derive(Serialize)`.
 
 ## Naming Conventions
 

Cargo.lock

@@ -9642,6 +9642,7 @@ dependencies = [
  "tokio",
  "tokio-util",
  "tracing",
+ "tracing-subscriber",
  "url",
  "uuid",
  "vaultrs",

crates/data-usage/src/data_usage.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize, ser::SerializeMap as _};
 use std::{
     collections::{HashMap, HashSet},
     hash::{DefaultHasher, Hash, Hasher},
@@ -51,24 +51,36 @@ pub fn usage_last_update_is_untrusted_future(existing_last_update: SystemTime, n
     existing_last_update > now + USAGE_LAST_UPDATE_FUTURE_TOLERANCE
 }
 
-#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct TierStats {
     pub total_size: u64,
-    pub num_versions: i32,
-    pub num_objects: i32,
+    pub num_versions: u64,
+    pub num_objects: u64,
 }
 
 impl TierStats {
     pub fn add(&self, u: &TierStats) -> TierStats {
         TierStats {
-            total_size: self.total_size + u.total_size,
-            num_versions: self.num_versions + u.num_versions,
-            num_objects: self.num_objects + u.num_objects,
+            total_size: self.total_size.saturating_add(u.total_size),
+            num_versions: self.num_versions.saturating_add(u.num_versions),
+            num_objects: self.num_objects.saturating_add(u.num_objects),
         }
     }
+
+    /// True when [`TierStats::add`] would report the exact sum instead of saturating.
+    pub fn fits_add(&self, u: &TierStats) -> bool {
+        self.total_size.checked_add(u.total_size).is_some()
+            && self.num_versions.checked_add(u.num_versions).is_some()
+            && self.num_objects.checked_add(u.num_objects).is_some()
+    }
+
+    /// True when this tier contributed nothing, i.e. merging it is a no-op.
+    pub fn is_empty(&self) -> bool {
+        self.total_size == 0 && self.num_versions == 0 && self.num_objects == 0
+    }
 }
 
-#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 pub struct AllTierStats {
     pub tiers: HashMap<String, TierStats>,
 }
@@ -78,31 +90,35 @@ impl AllTierStats {
         Self { tiers: HashMap::new() }
     }
 
-    pub fn add_sizes(&mut self, tiers: HashMap<String, TierStats>) {
+    pub fn is_empty(&self) -> bool {
+        self.tiers.is_empty()
+    }
+
+    /// Folds a scan summary's per-tier map in.
+    ///
+    /// Scanners seed the map with a zeroed entry for every configured tier, so
+    /// empty contributions are skipped to keep the persisted cache from growing
+    /// one key per tier on every folder that never held tiered data.
+    pub fn add_sizes(&mut self, tiers: &HashMap<String, TierStats>) {
         for (tier, st) in tiers {
-            self.tiers
-                .insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
+            if st.is_empty() {
+                continue;
+            }
+            let entry = self.tiers.entry(tier.clone()).or_default();
+            *entry = entry.add(st);
         }
     }
 
-    pub fn merge(&mut self, other: AllTierStats) {
-        for (tier, st) in other.tiers {
-            self.tiers
-                .insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
-        }
+    pub fn merge(&mut self, other: &AllTierStats) {
+        self.add_sizes(&other.tiers);
     }
 
-    pub fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) {
-        for (tier, st) in &self.tiers {
-            stats.insert(
-                tier.clone(),
-                TierStats {
-                    total_size: st.total_size,
-                    num_versions: st.num_versions,
-                    num_objects: st.num_objects,
-                },
-            );
-        }
+    /// True when [`AllTierStats::merge`] would report exact sums for every tier.
+    pub fn fits_merge(&self, other: &AllTierStats) -> bool {
+        other
+            .tiers
+            .iter()
+            .all(|(tier, right)| self.tiers.get(tier).is_none_or(|left| left.fits_add(right)))
     }
 }
 
@@ -183,6 +199,14 @@ pub struct DataUsageInfo {
     pub objects_total_size: u64,
     /// Replication info across all buckets
     pub replication_info: HashMap<String, BucketTargetUsageInfo>,
+    /// Usage per storage class and remote tier across all buckets.
+    ///
+    /// Absent on snapshots written before per-tier accounting was published,
+    /// and on clusters with no remote tier configured: the scanner classifies
+    /// objects by tier (including `STANDARD`/`REDUCED_REDUNDANCY`) only once a
+    /// tier exists, so an absent value means "not accounted", never "zero".
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub tier_stats: Option<AllTierStats>,
 
     /// Total number of buckets in this cluster
     pub buckets_count: u64,
@@ -562,7 +586,7 @@ impl ReplicationAllStats {
 }
 
 /// Data usage cache entry
-#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+#[derive(Clone, Debug, Default, Deserialize)]
 pub struct DataUsageEntry {
     pub children: DataUsageHashMap,
     // These fields do not include any children.
@@ -577,6 +601,34 @@ pub struct DataUsageEntry {
     /// Number of objects that failed to scan (e.g., IO errors)
     #[serde(default)]
     pub failed_objects: usize,
+    /// Per-tier usage contributed by this entry, present only once a scan
+    /// observed tier-classified objects.
+    #[serde(default)]
+    pub all_tier_stats: Option<AllTierStats>,
+}
+
+impl Serialize for DataUsageEntry {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        // Keep entries map-encoded so older readers can ignore fields appended
+        // by newer scanner versions during rolling upgrades. The derived
+        // (array) encoding made any appended field a decode error for them.
+        let mut state = serializer.serialize_map(Some(11))?;
+        state.serialize_entry("children", &self.children)?;
+        state.serialize_entry("size", &self.size)?;
+        state.serialize_entry("objects", &self.objects)?;
+        state.serialize_entry("versions", &self.versions)?;
+        state.serialize_entry("delete_markers", &self.delete_markers)?;
+        state.serialize_entry("obj_sizes", &self.obj_sizes)?;
+        state.serialize_entry("obj_versions", &self.obj_versions)?;
+        state.serialize_entry("replication_stats", &self.replication_stats)?;
+        state.serialize_entry("compacted", &self.compacted)?;
+        state.serialize_entry("failed_objects", &self.failed_objects)?;
+        state.serialize_entry("all_tier_stats", &self.all_tier_stats)?;
+        state.end()
+    }
 }
 
 impl DataUsageEntry {
@@ -635,10 +687,22 @@ impl DataUsageEntry {
             }
         }
 
+        if let Some(o_tiers) = other.all_tier_stats.as_ref().filter(|tiers| !tiers.is_empty()) {
+            self.all_tier_stats.get_or_insert_with(AllTierStats::new).merge(o_tiers);
+        }
+
         self.obj_sizes.merge_from(&other.obj_sizes);
         self.obj_versions.merge_from(&other.obj_versions);
     }
 
+    /// Folds a scan summary's per-tier map into this entry.
+    pub fn add_tier_sizes(&mut self, tiers: &HashMap<String, TierStats>) {
+        if tiers.values().all(TierStats::is_empty) {
+            return;
+        }
+        self.all_tier_stats.get_or_insert_with(AllTierStats::new).add_sizes(tiers);
+    }
+
     pub fn checked_merge(&mut self, other: &DataUsageEntry) -> bool {
         let scalar_counts_fit = self.objects.checked_add(other.objects).is_some()
             && self.versions.checked_add(other.versions).is_some()
@@ -698,7 +762,12 @@ impl DataUsageEntry {
             }
         };
 
-        if !scalar_counts_fit || !histograms_fit || !replication_fits {
+        let tier_stats_fit = match (&self.all_tier_stats, &other.all_tier_stats) {
+            (_, None) | (None, Some(_)) => true,
+            (Some(left), Some(right)) => left.fits_merge(right),
+        };
+
+        if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit {
             return false;
         }
         self.merge(other);
@@ -1038,6 +1107,7 @@ impl DataUsageCache {
             versions_total_count: flat.versions as u64,
             delete_markers_total_count: flat.delete_markers as u64,
             objects_total_size: flat.size as u64,
+            tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
             buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
             buckets_usage,
             usage_snapshot_complete: self.info.snapshot_complete,
@@ -1525,6 +1595,172 @@ mod tests {
         buckets_count: u64,
     }
 
+    fn tier_entry(tier: &str, stats: TierStats) -> DataUsageEntry {
+        let mut entry = DataUsageEntry::default();
+        entry.add_tier_sizes(&HashMap::from([(tier.to_string(), stats)]));
+        entry
+    }
+
+    #[test]
+    fn tier_stats_survive_entry_merge() {
+        let mut left = tier_entry(
+            "WARM",
+            TierStats {
+                total_size: 10,
+                num_versions: 2,
+                num_objects: 1,
+            },
+        );
+        let mut right = tier_entry(
+            "WARM",
+            TierStats {
+                total_size: 5,
+                num_versions: 1,
+                num_objects: 1,
+            },
+        );
+        right.add_tier_sizes(&HashMap::from([(
+            "COLD".to_string(),
+            TierStats {
+                total_size: 7,
+                num_versions: 1,
+                num_objects: 0,
+            },
+        )]));
+
+        assert!(left.checked_merge(&right), "merging exact tier totals must be accepted");
+
+        let tiers = &left.all_tier_stats.expect("merged entry keeps tier stats").tiers;
+        assert_eq!(
+            tiers.get("WARM"),
+            Some(&TierStats {
+                total_size: 15,
+                num_versions: 3,
+                num_objects: 2,
+            })
+        );
+        assert_eq!(
+            tiers.get("COLD"),
+            Some(&TierStats {
+                total_size: 7,
+                num_versions: 1,
+                num_objects: 0,
+            })
+        );
+    }
+
+    #[test]
+    fn tier_stats_merge_into_an_untiered_entry() {
+        let mut left = DataUsageEntry::default();
+        let right = tier_entry(
+            "WARM",
+            TierStats {
+                total_size: 10,
+                num_versions: 1,
+                num_objects: 1,
+            },
+        );
+
+        assert!(left.checked_merge(&right));
+
+        assert_eq!(
+            left.all_tier_stats.expect("tier stats adopted from the merged entry").tiers["WARM"],
+            TierStats {
+                total_size: 10,
+                num_versions: 1,
+                num_objects: 1,
+            }
+        );
+    }
+
+    #[test]
+    fn checked_merge_rejects_overflowing_tier_totals() {
+        let mut left = tier_entry(
+            "WARM",
+            TierStats {
+                total_size: u64::MAX,
+                num_versions: 1,
+                num_objects: 1,
+            },
+        );
+        let right = tier_entry(
+            "WARM",
+            TierStats {
+                total_size: 1,
+                num_versions: 1,
+                num_objects: 1,
+            },
+        );
+
+        assert!(!left.checked_merge(&right), "saturating tier totals must not be published");
+        assert_eq!(left.all_tier_stats.expect("left is untouched").tiers["WARM"].total_size, u64::MAX);
+    }
+
+    /// Entry shape released before per-tier accounting, using the derived
+    /// (array) encoding those writers produced.
+    #[derive(Serialize, Deserialize)]
+    struct LegacyEntry {
+        children: DataUsageHashMap,
+        size: usize,
+        objects: usize,
+        versions: usize,
+        delete_markers: usize,
+        obj_sizes: SizeHistogram,
+        obj_versions: VersionsHistogram,
+        replication_stats: Option<ReplicationAllStats>,
+        compacted: bool,
+        #[serde(default)]
+        failed_objects: usize,
+    }
+
+    #[test]
+    fn entries_are_map_encoded_so_appended_fields_stay_readable() {
+        // A derived (array) encoding turns every appended field into a decode
+        // error for readers built before it existed, which would cost a mixed
+        // -version cluster its whole scan cache. Entries must stay map-encoded.
+        let current = tier_entry(
+            "WARM",
+            TierStats {
+                total_size: 3,
+                num_versions: 1,
+                num_objects: 1,
+            },
+        );
+        let mut encoded = Vec::new();
+        current
+            .serialize(&mut rmp_serde::Serializer::new(&mut encoded))
+            .expect("encode current entry");
+
+        let legacy: LegacyEntry = rmp_serde::from_slice(&encoded).expect("legacy reader should ignore the appended field");
+        assert_eq!(legacy.objects, 0);
+    }
+
+    #[test]
+    fn legacy_array_encoded_entries_still_load() {
+        let legacy = LegacyEntry {
+            children: DataUsageHashMap::default(),
+            size: 12,
+            objects: 3,
+            versions: 4,
+            delete_markers: 1,
+            obj_sizes: SizeHistogram::default(),
+            obj_versions: VersionsHistogram::default(),
+            replication_stats: None,
+            compacted: false,
+            failed_objects: 2,
+        };
+        let mut encoded = Vec::new();
+        legacy
+            .serialize(&mut rmp_serde::Serializer::new(&mut encoded))
+            .expect("encode legacy entry");
+
+        let decoded: DataUsageEntry = rmp_serde::from_slice(&encoded).expect("current reader should default the missing field");
+
+        assert_eq!(decoded.size, 12);
+        assert_eq!(decoded.failed_objects, 2);
+        assert!(decoded.all_tier_stats.is_none());
+    }
+
     #[test]
     fn hash_path_uses_portable_slash_semantics() {
         for (input, expected) in [
@@ -1901,6 +2137,44 @@ mod tests {
         assert_eq!(info.buckets_count, 2);
         assert!(info.buckets_usage.is_empty());
         assert_eq!(info.objects_total_count, 3);
+        assert!(info.tier_stats.is_none());
+    }
+
+    #[test]
+    fn test_dui_reports_tier_usage_from_the_flattened_tree() {
+        let root_hash = hash_path("root");
+        let bucket_hash = hash_path("bucket-a");
+        let mut cache = DataUsageCache {
+            info: DataUsageCacheInfo {
+                name: "root".to_string(),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        cache.replace_hashed(&root_hash, &None, &DataUsageEntry::default());
+        cache.replace_hashed(
+            &bucket_hash,
+            &Some(root_hash),
+            &tier_entry(
+                "WARM",
+                TierStats {
+                    total_size: 40,
+                    num_versions: 2,
+                    num_objects: 2,
+                },
+            ),
+        );
+
+        let info = cache.dui("root", &["bucket-a".to_string()]);
+
+        assert_eq!(
+            info.tier_stats.expect("child tier usage should roll up to the root").tiers["WARM"],
+            TierStats {
+                total_size: 40,
+                num_versions: 2,
+                num_objects: 2,
+            }
+        );
     }
 
     #[test]

crates/e2e_test/src/kms/copy_object_self_copy_sse_test.rs

@@ -0,0 +1,351 @@
+// 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.
+
+//! Regression test: a same-key CopyObject that only rewrites metadata must never re-key a
+//! managed-SSE (SSE-S3 / SSE-KMS) object.
+//!
+//! On an **unversioned** bucket the handler marks a same-name copy `metadata_only`, and the
+//! store layer then updates `xl.meta` in place without touching the data blocks. The handler
+//! nevertheless strips the source encryption metadata and generates a *fresh* DEK for the
+//! destination. Combining the two writes "new DEK + old ciphertext": the object is permanently
+//! undecryptable. The fix forces a full data rewrite whenever the copy re-derives managed
+//! encryption material, so the stored bytes always match the key metadata beside them.
+//!
+//! Companion to `copy_object_version_restore_sse_test` (issue #4238), which pins the same
+//! invariant for the versioned historical-restore path.
+
+use super::common::{LocalKMSTestEnvironment, create_key_with_specific_id};
+use crate::common::init_logging;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{
+    MetadataDirective, ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration,
+    ServerSideEncryptionRule,
+};
+use serial_test::serial;
+use tracing::info;
+
+#[tokio::test]
+#[serial]
+async fn test_metadata_replace_self_copy_of_sse_object_stays_decryptable() {
+    init_logging();
+    info!("same-key CopyObject with REPLACE metadata must not re-key an SSE-S3 object");
+
+    let mut kms_env = LocalKMSTestEnvironment::new().await.expect("failed to create local KMS env");
+    let default_key_id = "rustfs-e2e-test-default-key";
+    let keys_dir = kms_env.kms_keys_dir.clone();
+    create_key_with_specific_id(&keys_dir, default_key_id)
+        .await
+        .expect("failed to create local KMS key");
+    kms_env
+        .base_env
+        .start_rustfs_server_with_env(
+            vec![
+                "--kms-enable",
+                "--kms-backend",
+                "local",
+                "--kms-key-dir",
+                &keys_dir,
+                "--kms-default-key-id",
+                default_key_id,
+            ],
+            &[("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true")],
+        )
+        .await
+        .expect("failed to start RustFS with local KMS");
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+
+    let client = kms_env.base_env.create_s3_client();
+    // Deliberately an UNVERSIONED bucket: that is the branch where the store layer can service
+    // the self-copy as a pure metadata update.
+    let bucket = "copy-object-self-copy-sse-test";
+    let key = "secrets/report.txt";
+
+    client
+        .create_bucket()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to create bucket");
+
+    // Content long enough that a truncated/garbled decrypt cannot coincidentally match.
+    let content = b"encrypted payload that must survive a metadata-only self copy -- 0123456789";
+    let put = client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .content_type("text/plain; charset=utf-8")
+        .metadata("stage", "before")
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .body(ByteStream::from_static(content))
+        .send()
+        .await
+        .expect("PUT failed");
+    assert_eq!(put.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+
+    // Copy the object onto itself, replacing user metadata. This is the `mc cp --attr` /
+    // "edit metadata in place" shape that AWS supports on an existing object.
+    let copy_out = client
+        .copy_object()
+        .bucket(bucket)
+        .key(key)
+        .copy_source(format!("{bucket}/{key}"))
+        .metadata_directive(MetadataDirective::Replace)
+        .content_type("text/plain; charset=utf-8")
+        .metadata("stage", "after")
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .send()
+        .await
+        .expect("same-key CopyObject with REPLACE metadata must succeed");
+    assert_eq!(copy_out.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+
+    // The object must still decrypt to the original plaintext. Before the fix the stored
+    // ciphertext was left untouched while the metadata carried a brand-new DEK, so this GET
+    // either failed outright or returned garbage.
+    let get = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("GET after self-copy failed: the object was re-keyed without rewriting the ciphertext");
+    assert_eq!(get.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    assert_eq!(
+        get.metadata().and_then(|m| m.get("stage")),
+        Some(&"after".to_string()),
+        "REPLACE metadata must take effect"
+    );
+    let body = get.body.collect().await.expect("collect body").into_bytes();
+    assert_eq!(
+        body.as_ref(),
+        content,
+        "object must still decrypt to the original plaintext after a metadata-only self copy"
+    );
+
+    kms_env.base_env.stop_server();
+}
+
+#[tokio::test]
+#[serial]
+async fn test_metadata_replace_self_copy_dropping_sse_rewrites_plaintext() {
+    init_logging();
+    info!("same-key CopyObject that drops SSE must rewrite the data, not orphan the ciphertext");
+
+    let mut kms_env = LocalKMSTestEnvironment::new().await.expect("failed to create local KMS env");
+    let default_key_id = "rustfs-e2e-test-default-key";
+    let keys_dir = kms_env.kms_keys_dir.clone();
+    create_key_with_specific_id(&keys_dir, default_key_id)
+        .await
+        .expect("failed to create local KMS key");
+    kms_env
+        .base_env
+        .start_rustfs_server_with_env(
+            vec![
+                "--kms-enable",
+                "--kms-backend",
+                "local",
+                "--kms-key-dir",
+                &keys_dir,
+                "--kms-default-key-id",
+                default_key_id,
+            ],
+            &[("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true")],
+        )
+        .await
+        .expect("failed to start RustFS with local KMS");
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+
+    let client = kms_env.base_env.create_s3_client();
+    // Unversioned, and deliberately WITHOUT a bucket default-encryption rule, so the copy below
+    // resolves to "no destination encryption".
+    let bucket = "copy-object-self-copy-drop-sse-test";
+    let key = "secrets/report.txt";
+
+    client
+        .create_bucket()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to create bucket");
+
+    let content = b"encrypted payload whose ciphertext must not survive as bogus plaintext -- 0123456789";
+    client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .metadata("stage", "before")
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .body(ByteStream::from_static(content))
+        .send()
+        .await
+        .expect("PUT failed");
+
+    // Self-copy with REPLACE and no SSE header. Per AWS semantics the destination ends up
+    // unencrypted. The dangerous outcome is the silent one: the handler strips the source key
+    // metadata while a metadata-only copy leaves the ciphertext in place, so a later GET would
+    // hand back raw ciphertext as if it were plaintext — corruption with no error anywhere.
+    client
+        .copy_object()
+        .bucket(bucket)
+        .key(key)
+        .copy_source(format!("{bucket}/{key}"))
+        .metadata_directive(MetadataDirective::Replace)
+        .metadata("stage", "after")
+        .send()
+        .await
+        .expect("same-key CopyObject dropping SSE must succeed");
+
+    let get = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("GET after self-copy failed");
+    assert_eq!(
+        get.server_side_encryption(),
+        None,
+        "destination must be unencrypted once the copy drops SSE"
+    );
+    assert_eq!(
+        get.metadata().and_then(|m| m.get("stage")),
+        Some(&"after".to_string()),
+        "REPLACE metadata must take effect"
+    );
+    let body = get.body.collect().await.expect("collect body").into_bytes();
+    assert_eq!(
+        body.as_ref(),
+        content,
+        "object must read back as the original plaintext, not the orphaned ciphertext"
+    );
+
+    kms_env.base_env.stop_server();
+}
+
+#[tokio::test]
+#[serial]
+async fn test_metadata_replace_self_copy_under_bucket_default_sse_stays_decryptable() {
+    init_logging();
+    info!("bucket default encryption must also keep a same-key copy off the metadata-only path");
+
+    let mut kms_env = LocalKMSTestEnvironment::new().await.expect("failed to create local KMS env");
+    let default_key_id = "rustfs-e2e-test-default-key";
+    let keys_dir = kms_env.kms_keys_dir.clone();
+    create_key_with_specific_id(&keys_dir, default_key_id)
+        .await
+        .expect("failed to create local KMS key");
+    kms_env
+        .base_env
+        .start_rustfs_server_with_env(
+            vec![
+                "--kms-enable",
+                "--kms-backend",
+                "local",
+                "--kms-key-dir",
+                &keys_dir,
+                "--kms-default-key-id",
+                default_key_id,
+            ],
+            &[("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true")],
+        )
+        .await
+        .expect("failed to start RustFS with local KMS");
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+
+    let client = kms_env.base_env.create_s3_client();
+    let bucket = "copy-object-self-copy-bucket-default-sse-test";
+    let key = "secrets/report.txt";
+
+    client
+        .create_bucket()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to create bucket");
+
+    // Store the object as PLAINTEXT first: no SSE header and no bucket default rule yet. This is
+    // what makes the case sharp — at copy time the source metadata carries no encryption markers,
+    // so the source-side half of the guard cannot fire.
+    let content = b"plaintext payload that must not be orphaned under a new DEK -- 0123456789";
+    let put = client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .metadata("stage", "before")
+        .body(ByteStream::from_static(content))
+        .send()
+        .await
+        .expect("PUT failed");
+    assert_eq!(put.server_side_encryption(), None, "the object must start out unencrypted");
+
+    // Only NOW enable bucket default encryption. The destination's encryption therefore comes
+    // from the bucket rule and from nowhere else: the source is unencrypted and the copy request
+    // carries no SSE header. A guard that only inspects request headers (MinIO decides
+    // `isTargetEncrypted` from `crypto.S3.IsRequested(r.Header)`) would let this through, yet
+    // `sse_encryption` still mints a fresh DEK from the resolved bucket default — which is why
+    // the guard keys off the *effective* encryption rather than the requested one.
+    let encryption_config = ServerSideEncryptionConfiguration::builder()
+        .rules(
+            ServerSideEncryptionRule::builder()
+                .apply_server_side_encryption_by_default(
+                    ServerSideEncryptionByDefault::builder()
+                        .sse_algorithm(ServerSideEncryption::Aes256)
+                        .build()
+                        .unwrap(),
+                )
+                .build(),
+        )
+        .build()
+        .unwrap();
+    client
+        .put_bucket_encryption()
+        .bucket(bucket)
+        .server_side_encryption_configuration(encryption_config)
+        .send()
+        .await
+        .expect("failed to set bucket default encryption");
+
+    // No SSE header on the copy — the bucket default alone drives the destination encryption.
+    client
+        .copy_object()
+        .bucket(bucket)
+        .key(key)
+        .copy_source(format!("{bucket}/{key}"))
+        .metadata_directive(MetadataDirective::Replace)
+        .metadata("stage", "after")
+        .send()
+        .await
+        .expect("same-key CopyObject under bucket default encryption must succeed");
+
+    let get = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("GET after self-copy failed: the object was re-keyed without rewriting the ciphertext");
+    assert_eq!(get.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    assert_eq!(
+        get.metadata().and_then(|m| m.get("stage")),
+        Some(&"after".to_string()),
+        "REPLACE metadata must take effect"
+    );
+    let body = get.body.collect().await.expect("collect body").into_bytes();
+    assert_eq!(
+        body.as_ref(),
+        content,
+        "object must still decrypt to the original plaintext after a metadata-only self copy"
+    );
+
+    kms_env.base_env.stop_server();
+}

crates/e2e_test/src/kms/mod.rs

@@ -48,6 +48,9 @@ mod bucket_default_encryption_test;
 #[cfg(test)]
 mod encryption_metadata_test;
 
+#[cfg(test)]
+mod copy_object_self_copy_sse_test;
+
 #[cfg(test)]
 mod copy_object_version_restore_sse_test;
 

crates/ecstore/src/store/object.rs

@@ -969,6 +969,15 @@ impl ECStore {
 
             if !dst_opts.versioned && src_opts.version_id.is_none() {
                 if src_info.metadata_only {
+                    // Zero-copy update: only xl.meta is rewritten, the data blocks stay as they
+                    // are. The caller must therefore guarantee that the destination metadata
+                    // still describes the stored bytes. In particular a copy that re-derives
+                    // encryption material may NOT set metadata_only — that would leave a fresh
+                    // DEK beside ciphertext sealed under the old one, permanently destroying the
+                    // object. The S3 handler enforces this before calling in (see the
+                    // metadata_only decision in rustfs/src/app/object_usecase.rs); the sibling
+                    // versioned branch below resolves the same risk by rewriting through
+                    // put_object (issue #4238).
                     return self.pools[pool_idx]
                         .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &dst_opts)
                         .await;

crates/kms/Cargo.toml

@@ -99,6 +99,9 @@ tokio = { workspace = true, features = ["net", "test-util"] }
 # Replays canned AWS KMS HTTP exchanges so the AWS backend tests stay offline.
 aws-smithy-http-client = { workspace = true, default-features = false, features = ["test-util"] }
 http = { workspace = true }
+# Captures warning events in format-compatibility tests without installing a
+# process-wide subscriber.
+tracing-subscriber = { workspace = true, features = ["fmt"] }
 
 [features]
 default = []

crates/kms/src/backends/local.rs

@@ -32,6 +32,7 @@ use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use jiff::Zoned;
 use rand::RngExt;
+use serde::de::IgnoredAny;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
@@ -511,8 +512,12 @@ pub(crate) fn unknown_protection_marker(record: &[u8]) -> serde_json::Result<Opt
 }
 
 /// Serializable representation of a master key stored on disk
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize)]
 struct StoredMasterKey {
+    /// Persisted record schema version. Records written before this field was
+    /// introduced default to version 1 during deserialization.
+    #[serde(default = "default_stored_master_key_format_version")]
+    format_version: u32,
     key_id: String,
     version: u32,
     algorithm: String,
@@ -537,6 +542,81 @@ struct StoredMasterKey {
     at_rest_protection: StoredKeyProtection,
 }
 
+pub(crate) const STORED_MASTER_KEY_FORMAT_VERSION: u32 = 1;
+
+fn default_stored_master_key_format_version() -> u32 {
+    STORED_MASTER_KEY_FORMAT_VERSION
+}
+
+/// Read only the schema marker before attempting the complete key-record
+/// decode. A future record may add or remove required fields, but its version
+/// still needs to be reported as unsupported rather than as generic corruption.
+pub(crate) fn stored_master_key_format_version(record: &[u8]) -> serde_json::Result<u32> {
+    #[derive(Deserialize)]
+    struct FormatProbe {
+        #[serde(default = "default_stored_master_key_format_version")]
+        format_version: u32,
+    }
+
+    Ok(serde_json::from_slice::<FormatProbe>(record)?.format_version)
+}
+
+impl<'de> Deserialize<'de> for StoredMasterKey {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct Wire {
+            #[serde(default = "default_stored_master_key_format_version")]
+            format_version: u32,
+            key_id: String,
+            version: u32,
+            algorithm: String,
+            usage: KeyUsage,
+            status: KeyStatus,
+            description: Option<String>,
+            metadata: HashMap<String, String>,
+            #[serde(with = "crate::time_serde::zoned")]
+            created_at: Zoned,
+            #[serde(with = "crate::time_serde::option_zoned")]
+            rotated_at: Option<Zoned>,
+            created_by: Option<String>,
+            #[serde(default, with = "crate::time_serde::option_zoned")]
+            deletion_date: Option<Zoned>,
+            encrypted_key_material: String,
+            nonce: Vec<u8>,
+            #[serde(default)]
+            at_rest_protection: StoredKeyProtection,
+            #[serde(flatten)]
+            unknown_fields: HashMap<String, IgnoredAny>,
+        }
+
+        let wire = Wire::deserialize(deserializer)?;
+        for field in wire.unknown_fields.keys() {
+            tracing::warn!(key_id = %wire.key_id, field = %field, "Local KMS key record contains an unknown field");
+        }
+
+        Ok(Self {
+            format_version: wire.format_version,
+            key_id: wire.key_id,
+            version: wire.version,
+            algorithm: wire.algorithm,
+            usage: wire.usage,
+            status: wire.status,
+            description: wire.description,
+            metadata: wire.metadata,
+            created_at: wire.created_at,
+            rotated_at: wire.rotated_at,
+            created_by: wire.created_by,
+            deletion_date: wire.deletion_date,
+            encrypted_key_material: wire.encrypted_key_material,
+            nonce: wire.nonce,
+            at_rest_protection: wire.at_rest_protection,
+        })
+    }
+}
+
 impl LocalKmsClient {
     /// Create a new local KMS client
     pub async fn new(config: LocalConfig) -> Result<Self> {
@@ -851,6 +931,12 @@ impl LocalKmsClient {
 
         let content = fs::read(&key_path).await?;
 
+        let format_version = stored_master_key_format_version(&content)
+            .map_err(|e| KmsError::material_corrupt(key_id, format!("stored key record is not a readable JSON object: {e}")))?;
+        if format_version > STORED_MASTER_KEY_FORMAT_VERSION {
+            return Err(KmsError::unsupported_format_version(key_id, format_version.to_string()));
+        }
+
         // Two-stage parse so an unrecognised protection marker is reported as an
         // unsupported format (a newer build may still read the key) instead of being
         // folded into generic corruption with every other malformed record.
@@ -1023,6 +1109,7 @@ impl LocalKmsClient {
         };
 
         let stored_key = StoredMasterKey {
+            format_version: STORED_MASTER_KEY_FORMAT_VERSION,
             key_id: master_key.key_id.clone(),
             version: master_key.version,
             algorithm: master_key.algorithm.clone(),
@@ -2428,6 +2515,121 @@ mod tests {
         assert_eq!(key_info.created_at.time_zone().iana_name(), Some("UTC"));
     }
 
+    #[tokio::test]
+    async fn stored_master_key_format_version_is_explicit_and_legacy_defaults_to_v1() {
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client.create_key("format-key", "AES_256", None).await.expect("create key");
+
+        let key_path = client.master_key_path("format-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        assert_eq!(record.get("format_version"), Some(&serde_json::json!(STORED_MASTER_KEY_FORMAT_VERSION)));
+
+        // A record from before the explicit field was added remains readable.
+        record
+            .as_object_mut()
+            .expect("key record is an object")
+            .remove("format_version")
+            .expect("current records carry format_version");
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode legacy key record"))
+            .await
+            .expect("write legacy key record");
+        let info = client
+            .describe_key("format-key", None)
+            .await
+            .expect("legacy key record should load");
+        assert_eq!(info.key_id, "format-key");
+    }
+
+    #[tokio::test]
+    async fn stored_master_key_accepts_an_older_numeric_format_version() {
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client
+            .create_key("older-format-key", "AES_256", None)
+            .await
+            .expect("create key");
+
+        let key_path = client.master_key_path("older-format-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        record["format_version"] = serde_json::json!(0);
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode older key record"))
+            .await
+            .expect("write older key record");
+
+        let info = client
+            .describe_key("older-format-key", None)
+            .await
+            .expect("older format version should remain readable");
+        assert_eq!(info.key_id, "older-format-key");
+    }
+
+    #[tokio::test]
+    async fn stored_master_key_rejects_a_newer_format_version_before_decrypting() {
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client
+            .create_key("future-format-key", "AES_256", None)
+            .await
+            .expect("create key");
+
+        let key_path = client.master_key_path("future-format-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        record["format_version"] = serde_json::json!(99);
+        record.as_object_mut().expect("key record is an object").remove("usage");
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode future key record"))
+            .await
+            .expect("write future key record");
+
+        let error = client
+            .describe_key("future-format-key", None)
+            .await
+            .expect_err("a newer key format must fail closed");
+        assert!(matches!(
+            error,
+            KmsError::UnsupportedFormatVersion { key_id, version }
+                if key_id == "future-format-key" && version == "99"
+        ));
+    }
+
+    #[tokio::test]
+    async fn stored_master_key_unknown_fields_remain_readable() {
+        const UNKNOWN_FIELD_VALUE: &str = "field value must not be logged";
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client
+            .create_key("unknown-field-key", "AES_256", None)
+            .await
+            .expect("create key");
+
+        let key_path = client.master_key_path("unknown-field-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        record["future_field"] = serde_json::json!("field value must not be logged");
+        fs::write(
+            &key_path,
+            serde_json::to_vec_pretty(&record).expect("encode key record with unknown field"),
+        )
+        .await
+        .expect("write key record with unknown field");
+
+        let logs = crate::test_support::CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(logs.clone())
+            .finish();
+        let record = fs::read(&key_path).await.expect("read key record");
+        let stored: StoredMasterKey = tracing::subscriber::with_default(subscriber, || {
+            serde_json::from_slice(&record).expect("unknown fields must remain forward-compatible")
+        });
+        assert_eq!(stored.key_id, "unknown-field-key");
+
+        let output = logs.output();
+        assert!(output.contains("Local KMS key record contains an unknown field"));
+        assert!(output.contains("future_field"));
+        assert!(!output.contains(UNKNOWN_FIELD_VALUE));
+    }
+
     #[tokio::test]
     async fn test_load_master_key_accepts_legacy_encrypted_record_without_protection_field() {
         let (client, temp_dir) = create_test_client().await;

crates/kms/src/backup/dry_run.rs

@@ -72,6 +72,7 @@ impl From<&BackupError> for RestoreBlocker {
             BackupError::Corrupted { .. } => RestoreBlockerCode::BundleCorrupted,
             BackupError::Truncated { .. } => RestoreBlockerCode::BundleTruncated,
             BackupError::UnknownVersion { .. } => RestoreBlockerCode::UnknownFormatVersion,
+            BackupError::UnsupportedFormatVersion { .. } => RestoreBlockerCode::UnknownFormatVersion,
             BackupError::WrongKek { .. } => RestoreBlockerCode::WrongBackupKek,
             BackupError::MissingArtifact { .. } => RestoreBlockerCode::MissingArtifact,
             BackupError::IncompleteBundle { .. } => RestoreBlockerCode::IncompleteBundle,

crates/kms/src/backup/error.rs

@@ -51,6 +51,11 @@ pub enum BackupError {
         supplied_kek_version: u32,
     },
 
+    /// A bundled key record declares a format version this build does not
+    /// understand.
+    #[error("bundled key record '{key_id}' declares unsupported format version {version}; this build cannot restore it")]
+    UnsupportedFormatVersion { key_id: String, version: String },
+
     /// Manifest requires an artifact that is not present in the bundle.
     #[error("backup bundle is missing a required artifact: {artifact}")]
     MissingArtifact { artifact: String },
@@ -128,6 +133,15 @@ mod tests {
             "unknown backup manifest format version 9 (this build supports version 1)"
         );
 
+        assert_eq!(
+            BackupError::UnsupportedFormatVersion {
+                key_id: "alpha".to_string(),
+                version: "9".to_string(),
+            }
+            .to_string(),
+            "bundled key record 'alpha' declares unsupported format version 9; this build cannot restore it"
+        );
+
         assert_eq!(
             BackupError::missing_artifact("key-material").to_string(),
             "backup bundle is missing a required artifact: key-material"

crates/kms/src/backup/local_export.rs

@@ -48,7 +48,10 @@
 //! published. A crash at any earlier point leaves a bundle without a
 //! manifest, which decodes as an incomplete bundle and can never be restored.
 
-use crate::backends::local::{LocalKmsClient, StoredKeyProtection, unknown_protection_marker};
+use crate::backends::local::{
+    LocalKmsClient, STORED_MASTER_KEY_FORMAT_VERSION, StoredKeyProtection, stored_master_key_format_version,
+    unknown_protection_marker,
+};
 use crate::backup::capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
 use crate::backup::error::BackupError;
 use crate::backup::manifest::{
@@ -380,6 +383,12 @@ async fn collect_snapshot(client: &LocalKmsClient) -> Result<CollectedSnapshot>
         // classified first so a record from a newer build keeps its own
         // verdict — an operator who reads "material corrupt" starts a
         // disaster recovery for what is only a version mismatch.
+        let format_version = stored_master_key_format_version(&raw).map_err(|error| {
+            KmsError::material_corrupt(&stem, format!("stored key record is not a readable JSON object: {error}"))
+        })?;
+        if format_version > STORED_MASTER_KEY_FORMAT_VERSION {
+            return Err(KmsError::unsupported_format_version(&stem, format_version.to_string()));
+        }
         let unknown_marker = unknown_protection_marker(&raw).map_err(|error| {
             KmsError::material_corrupt(&stem, format!("stored key record is not a readable JSON object: {error}"))
         })?;
@@ -1150,4 +1159,26 @@ mod tests {
             "got {error:?}"
         );
     }
+
+    #[tokio::test]
+    async fn numeric_record_format_version_from_a_newer_build_aborts_export_as_unsupported_format() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("alpha", "AES_256", None).await.expect("create key");
+
+        let record_path = client.key_directory().join("alpha.key");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&record_path).expect("read record")).expect("decode record");
+        record["format_version"] = serde_json::json!(99);
+        std::fs::write(&record_path, serde_json::to_vec_pretty(&record).expect("encode record")).expect("write record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("a newer record format must abort the export");
+        assert!(
+            matches!(&error, KmsError::UnsupportedFormatVersion { key_id, version }
+                if key_id == "alpha" && version == "99"),
+            "got {error:?}"
+        );
+    }
 }

crates/kms/src/backup/local_restore.rs

@@ -54,7 +54,8 @@
 
 use crate::backends::local::{
     LOCAL_KMS_MASTER_KEY_SALT_FILE, LOCAL_KMS_MASTER_KEY_SALT_LEN, LOCAL_RESTORE_COMMIT_MARKER_FILE, LocalKmsClient,
-    StoredKeyProtection, durable_file, is_orphan_commit_temp_name, unknown_protection_marker, validate_key_id,
+    STORED_MASTER_KEY_FORMAT_VERSION, StoredKeyProtection, durable_file, is_orphan_commit_temp_name,
+    stored_master_key_format_version, unknown_protection_marker, validate_key_id,
 };
 use crate::backup::capability::AtRestProtection;
 use crate::backup::dry_run::{
@@ -608,6 +609,15 @@ fn decode_key_record(
     // Classify the protection marker before the schema parse: a record from a
     // newer build is not a damaged bundle, and reporting it as corruption
     // sends the operator into disaster recovery instead of a version change.
+    let format_version = stored_master_key_format_version(&plaintext)
+        .map_err(|error| BackupError::corrupted(format!("bundled key record '{stem}' is not a readable JSON object: {error}")))?;
+    if format_version > STORED_MASTER_KEY_FORMAT_VERSION {
+        return Err(BackupError::UnsupportedFormatVersion {
+            key_id: stem,
+            version: format_version.to_string(),
+        }
+        .into());
+    }
     let unknown_marker = unknown_protection_marker(&plaintext)
         .map_err(|error| BackupError::corrupted(format!("bundled key record '{stem}' is not a readable JSON object: {error}")))?;
     if let Some(version) = unknown_marker {
@@ -2102,6 +2112,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn bundled_record_format_version_from_a_newer_build_is_not_reported_as_corruption() {
+        let record = serde_json::json!({"format_version": 99});
+        let error = match decode_key_record(
+            "artifacts/keys/alpha.key.enc",
+            Zeroizing::new(serde_json::to_vec(&record).expect("encode record")),
+            &[AtRestProtection::EncryptedMasterKey],
+        ) {
+            Ok(_) => panic!("a record this build cannot interpret must be rejected"),
+            Err(error) => error,
+        };
+
+        let KmsError::Backup(inner) = &error else {
+            panic!("expected a backup error, got {error:?}");
+        };
+        assert!(
+            matches!(inner, BackupError::UnsupportedFormatVersion { key_id, version }
+                if key_id == "alpha" && version == "99"),
+            "got {inner:?}"
+        );
+        assert_eq!(
+            RestoreBlocker::from(inner).code,
+            RestoreBlockerCode::UnknownFormatVersion,
+            "a dry run must report a version blocker, not a corruption blocker"
+        );
+    }
+
     /// The commit marker's format-version branch, symmetric with the Vault
     /// restore marker's: an unknown version is refused outright everywhere the
     /// marker is read, and the backend refuses to start while it is present.

crates/kms/src/encryption/dek.rs

@@ -36,7 +36,7 @@ use std::collections::HashMap;
 /// material. Envelopes written before versioning carry `None`; backends must resolve
 /// `None` to a deterministic baseline version recorded in key metadata, never
 /// implicitly to whatever version is current.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct DataKeyEnvelope {
     pub key_id: String,
     pub master_key_id: String,
@@ -54,6 +54,45 @@ pub struct DataKeyEnvelope {
     pub master_key_version: Option<u32>,
 }
 
+impl<'de> Deserialize<'de> for DataKeyEnvelope {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct Wire {
+            key_id: String,
+            master_key_id: String,
+            key_spec: String,
+            encrypted_key: Vec<u8>,
+            nonce: Vec<u8>,
+            encryption_context: HashMap<String, String>,
+            #[serde(with = "crate::time_serde::zoned")]
+            created_at: Zoned,
+            #[serde(default)]
+            master_key_version: Option<u32>,
+            #[serde(flatten)]
+            unknown_fields: HashMap<String, IgnoredAny>,
+        }
+
+        let wire = Wire::deserialize(deserializer)?;
+        for field in wire.unknown_fields.keys() {
+            tracing::warn!(field = %field, "KMS data-key envelope contains an unknown field");
+        }
+
+        Ok(Self {
+            key_id: wire.key_id,
+            master_key_id: wire.master_key_id,
+            key_spec: wire.key_spec,
+            encrypted_key: wire.encrypted_key,
+            nonce: wire.nonce,
+            encryption_context: wire.encryption_context,
+            created_at: wire.created_at,
+            master_key_version: wire.master_key_version,
+        })
+    }
+}
+
 #[derive(Deserialize)]
 struct DataKeyEnvelopeMarker {
     #[serde(rename = "key_id")]
@@ -368,6 +407,38 @@ mod tests {
         assert_eq!(deserialized.master_key_version, None);
     }
 
+    #[test]
+    fn test_data_key_envelope_unknown_fields_remain_readable() {
+        const UNKNOWN_FIELD_VALUE: &str = "field value must not be logged";
+        let envelope_json = r#"{
+            "key_id": "test-key-id",
+            "master_key_id": "master-key-id",
+            "key_spec": "AES_256",
+            "encrypted_key": [1, 2, 3, 4],
+            "nonce": [5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16],
+            "encryption_context": {"bucket": "test-bucket"},
+            "created_at": "2024-01-01T00:00:00+00:00[UTC]",
+            "future_field": "field value must not be logged"
+        }"#;
+
+        let logs = crate::test_support::CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(logs.clone())
+            .finish();
+        let deserialized: DataKeyEnvelope = tracing::subscriber::with_default(subscriber, || {
+            serde_json::from_str(envelope_json).expect("unknown fields must remain readable")
+        });
+        assert_eq!(deserialized.key_id, "test-key-id");
+        assert_eq!(deserialized.master_key_version, None);
+
+        let output = logs.output();
+        assert!(output.contains("KMS data-key envelope contains an unknown field"));
+        assert!(output.contains("future_field"));
+        assert!(!output.contains(UNKNOWN_FIELD_VALUE));
+    }
+
     #[test]
     fn test_data_key_envelope_none_version_serializes_without_field() {
         // A `None` version must keep the serialized envelope on the historical

crates/kms/src/lib.rs

@@ -82,6 +82,45 @@ pub mod service_manager;
 mod time_serde;
 pub mod types;
 
+#[cfg(test)]
+pub(crate) mod test_support {
+    use std::io::{self, Write};
+    use std::sync::{Arc, Mutex};
+
+    #[derive(Clone, Default)]
+    pub(crate) struct CapturedLogs {
+        output: Arc<Mutex<Vec<u8>>>,
+    }
+
+    pub(crate) struct CapturedWriter(CapturedLogs);
+
+    impl Write for CapturedWriter {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            self.0.output.lock().expect("log buffer lock poisoned").extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs {
+        type Writer = CapturedWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            CapturedWriter(self.clone())
+        }
+    }
+
+    impl CapturedLogs {
+        pub(crate) fn output(&self) -> String {
+            String::from_utf8(self.output.lock().expect("log buffer lock poisoned").clone())
+                .expect("captured logs should be UTF-8")
+        }
+    }
+}
+
 // Re-export public API
 pub use api_types::{
     CacheSummary, ConfigureAwsKmsRequest, ConfigureKmsRequest, ConfigureKmsResponse, ConfigureLocalKmsRequest,

crates/scanner/src/data_usage_define.rs

@@ -27,8 +27,8 @@ use rustfs_common::heal_channel::HealScanMode;
 #[cfg(test)]
 use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
-    BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DataUsageEntry, DataUsageHash, DataUsageHashMap,
-    DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, hash_path,
+    AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DataUsageEntry, DataUsageHash,
+    DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, TierStats, hash_path,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -184,69 +184,6 @@ pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
 
 const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
 
-#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq)]
-pub struct TierStats {
-    pub total_size: u64,
-    pub num_versions: i32,
-    pub num_objects: i32,
-}
-
-impl TierStats {
-    pub fn add(&self, u: &TierStats) -> TierStats {
-        TierStats {
-            total_size: self.total_size + u.total_size,
-            num_versions: self.num_versions + u.num_versions,
-            num_objects: self.num_objects + u.num_objects,
-        }
-    }
-
-    pub fn from_object_info(oi: &ObjectInfo) -> Self {
-        TierStats {
-            total_size: oi.size as u64,
-            num_versions: 1,
-            num_objects: if oi.is_latest { 1 } else { 0 },
-        }
-    }
-}
-
-#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
-pub struct AllTierStats {
-    pub tiers: HashMap<String, TierStats>,
-}
-
-impl AllTierStats {
-    pub fn new() -> Self {
-        Self { tiers: HashMap::new() }
-    }
-
-    pub fn add_sizes(&mut self, tiers: HashMap<String, TierStats>) {
-        for (tier, st) in tiers {
-            self.tiers
-                .insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
-        }
-    }
-
-    pub fn merge(&mut self, other: AllTierStats) {
-        for (tier, st) in other.tiers {
-            self.tiers
-                .insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
-        }
-    }
-
-    pub fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) {
-        for (tier, st) in &self.tiers {
-            stats.insert(
-                tier.clone(),
-                TierStats {
-                    total_size: st.total_size,
-                    num_versions: st.num_versions,
-                    num_objects: st.num_objects,
-                },
-            );
-        }
-    }
-}
-
 /// Size summary for a single object or group of objects
 #[derive(Debug, Default, Clone)]
 pub struct SizeSummary {
@@ -301,7 +238,11 @@ impl SizeSummary {
         }
 
         if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
-            *tier_stats = tier_stats.add(&TierStats::from_object_info(oi));
+            *tier_stats = tier_stats.add(&TierStats {
+                total_size: u64::try_from(oi.size).unwrap_or(0),
+                num_versions: 1,
+                num_objects: u64::from(oi.is_latest),
+            });
         }
     }
 }
@@ -963,6 +904,7 @@ impl DataUsageCache {
             versions_total_count: flat.versions as u64,
             delete_markers_total_count: flat.delete_markers as u64,
             objects_total_size: flat.size as u64,
+            tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
             buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
             buckets_usage,
             ..Default::default()

crates/scanner/src/scanner_folder.rs

@@ -461,6 +461,8 @@ fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary)
             .pending_count
             .saturating_add(u64::try_from(st.pending_count).unwrap_or(u64::MAX));
     }
+
+    into.add_tier_sizes(&summary.tier_stats);
 }
 
 /// Cached folder information for scanning
@@ -3049,7 +3051,7 @@ mod tests {
 
     use super::*;
     use crate::storage_api::VersionPurgeStatusType;
-    use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, new_disk};
+    use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
     use rustfs_filemeta::{FileInfo, FileMeta};
     use serial_test::serial;
     #[cfg(unix)]
@@ -3128,6 +3130,56 @@ mod tests {
         assert_eq!(target_stats.pending_count, u64::MAX);
     }
 
+    #[test]
+    fn scanner_size_summary_application_accumulates_tier_stats() {
+        let mut entry = DataUsageEntry::default();
+        let mut summary = SizeSummary::default();
+        summary.tier_stats.insert(
+            "WARM".to_string(),
+            TierStats {
+                total_size: 100,
+                num_versions: 2,
+                num_objects: 1,
+            },
+        );
+        // Scanners seed a zeroed entry for every configured tier; those must not
+        // reach the cache as empty keys.
+        summary
+            .tier_stats
+            .insert(storageclass::STANDARD.to_string(), TierStats::default());
+
+        apply_scanner_size_summary(&mut entry, &summary);
+        apply_scanner_size_summary(&mut entry, &summary);
+
+        let tiers = entry.all_tier_stats.as_ref().expect("tier stats should be recorded");
+        assert_eq!(
+            tiers.tiers.get("WARM"),
+            Some(&TierStats {
+                total_size: 200,
+                num_versions: 4,
+                num_objects: 2,
+            })
+        );
+        assert!(!tiers.tiers.contains_key(storageclass::STANDARD));
+    }
+
+    #[test]
+    fn scanner_size_summary_application_skips_untiered_summaries() {
+        let mut entry = DataUsageEntry::default();
+        let mut summary = SizeSummary {
+            total_size: 10,
+            ..Default::default()
+        };
+        summary
+            .tier_stats
+            .insert(storageclass::STANDARD.to_string(), TierStats::default());
+        summary.tier_stats.insert(storageclass::RRS.to_string(), TierStats::default());
+
+        apply_scanner_size_summary(&mut entry, &summary);
+
+        assert!(entry.all_tier_stats.is_none(), "zero-only tier maps must not allocate cache state");
+    }
+
     async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-test-{}", Uuid::new_v4()));
         tokio::fs::create_dir_all(&temp_dir)

crates/scanner/src/scanner_io.rs

@@ -1146,6 +1146,7 @@ fn completed_data_usage_info(
         versions_total_count: u64::try_from(total.versions).ok()?,
         delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
         objects_total_size: u64::try_from(total.size).ok()?,
+        tier_stats: total.all_tier_stats.filter(|tiers| !tiers.is_empty()),
         buckets_count: u64::try_from(all_buckets.len()).ok()?,
         bucket_sizes,
         buckets_usage,
@@ -1250,6 +1251,57 @@ mod publish_gate_tests {
         completed_data_usage_info(results, &expected_sources, all_buckets, true, budget_elapsed, cancelled)
     }
 
+    #[test]
+    fn completed_data_usage_info_publishes_tier_stats_across_sets() {
+        let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string()];
+        let warm = |total_size, num_versions, num_objects| {
+            HashMap::from([(
+                "WARM".to_string(),
+                TierStats {
+                    total_size,
+                    num_versions,
+                    num_objects,
+                },
+            )])
+        };
+
+        let mut first_set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
+        let mut tiered = DataUsageEntry::default();
+        tiered.add_tier_sizes(&warm(100, 2, 1));
+        first_set.replace("bucket-b", DATA_USAGE_ROOT, tiered);
+
+        let mut second_set = completed_root_cache("bucket-b", 2, 20, DataUsageCacheSource::new(1, 0));
+        let mut tiered = DataUsageEntry::default();
+        tiered.add_tier_sizes(&warm(50, 1, 1));
+        second_set.replace("bucket-a", DATA_USAGE_ROOT, tiered);
+
+        let (data_usage_info, _) = completed_data_usage_info_for_test(&[first_set, second_set], &all_buckets, false, false)
+            .expect("completed sets should publish a snapshot");
+
+        assert_eq!(
+            data_usage_info
+                .tier_stats
+                .expect("tier usage should reach the snapshot")
+                .tiers["WARM"],
+            TierStats {
+                total_size: 150,
+                num_versions: 3,
+                num_objects: 2,
+            }
+        );
+    }
+
+    #[test]
+    fn completed_data_usage_info_omits_tier_stats_without_tiered_objects() {
+        let all_buckets = vec!["bucket-a".to_string()];
+        let set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
+
+        let (data_usage_info, _) = completed_data_usage_info_for_test(&[set], &all_buckets, false, false)
+            .expect("completed set should publish a snapshot");
+
+        assert!(data_usage_info.tier_stats.is_none());
+    }
+
     #[test]
     fn completed_data_usage_info_requires_every_set_before_publish() {
         let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string(), "bucket-empty".to_string()];

docs/operations/kms-backend-security.md

@@ -117,6 +117,7 @@ This section states only what is true of the current implementation. It is writt
 Nothing in this list requires a coordinated format cutover. The compatibility is deliberate and is covered by decode tests.
 
 - **DEK envelopes.** `DataKeyEnvelope::master_key_version` is optional and omitted when absent, so envelopes written by non-rotating backends stay byte-identical to the historical seven-field JSON shape. An upgraded node reading a pre-versioning envelope resolves `None` to the key's recorded baseline version, or — for a key that was never rotated, and so has no baseline — to the current version, which is exactly the pre-versioning behavior.
+- **Local key records.** Each `<key_id>.key` record carries `format_version: 1`; records written before that field existed default to version 1 when read. A reader accepts a record whose version is at most the version it understands, and rejects a newer version with `UnsupportedFormatVersion` before it attempts to decrypt key material. Unknown fields remain accepted for rollback compatibility, but their names are emitted at `warn` level without values.
 - **KV2 key records.** `baseline_version` is read with a serde default, so records written by older builds deserialize unchanged, and `None` correctly means "never rotated".
 - **Transit metadata records.** Metadata persisted in KV v2 by either build decodes on the other.
 

docs/operations/kms-cryptographic-compliance.md

@@ -51,7 +51,7 @@ Suggested boilerplate when the topic cannot be avoided:
 
 ### Guard
 
-There is currently no FIPS-related wording anywhere in the repository's Markdown; that clean baseline is what a grep anchor test protects. Any future occurrence of the banned strings in shipped documentation should be treated as a defect and either removed or brought under the qualifier rule above.
+`README.md` and `CHANGELOG.md` currently contain no FIPS-related wording; `scripts/check_fips_wording.sh` is the grep guard for that public baseline. Any future occurrence of the banned strings in either file should be treated as a defect and either removed or brought under the qualifier rule above. This document intentionally contains the terminology needed to define the policy and is not part of that narrow outward-material scan.
 
 ## The `rustfs-crypto` `fips` feature: what it actually does
 

rustfs/src/app/object_usecase.rs

@@ -6344,7 +6344,35 @@ impl DefaultObjectUsecase {
             return Err(s3_error!(PreconditionFailed));
         }
 
-        if cp_src_dst_same && src_info.transitioned_object.tier.is_empty() {
+        // A same-name copy is normally serviced as a metadata-only update: the store layer
+        // rewrites xl.meta in place and leaves the data blocks alone. That shortcut is only sound
+        // when the destination's physical bytes are identical to the source's, and encryption
+        // breaks exactly that. The destination metadata is rebuilt from scratch below —
+        // `strip_managed_encryption_metadata` drops the source DEK and `sse_encryption` mints a
+        // fresh one — so reusing the stored ciphertext would leave a new DEK sitting beside bytes
+        // it cannot decrypt, permanently destroying the object (GET fails with an AEAD tag
+        // mismatch). The mirror case is worse because it is silent: an encrypted source copied
+        // without any destination SSE keeps its ciphertext while losing the key metadata, so GET
+        // hands back raw ciphertext as if it were plaintext. So whenever either side is
+        // encrypted, leave metadata_only = false and let the store layer do a full read/write
+        // rewrite through put_object, the same resolution the versioned historical-restore path
+        // uses (issue #4238, crates/ecstore/src/store/object.rs).
+        //
+        // This mirrors MinIO's `isSourceEncrypted || isTargetEncrypted -> metadataOnly = false`
+        // in CopyObjectHandler, with one deliberate difference: MinIO decides "target encrypted"
+        // from request headers alone, while `effective_sse` here also resolves the bucket default
+        // encryption rule. `sse_encryption` mints a DEK from that resolved value, so a
+        // header-only check would miss a self-copy under a bucket default rule. The source half
+        // deliberately reuses `ObjectInfo::is_encrypted` rather than naming individual headers,
+        // so a future encryption flavour is covered here the moment it is recognised there.
+        //
+        // The zero-copy shortcut is only recoverable for encrypted objects by *preserving* the
+        // DEK that sealed the bytes and re-wrapping it under a new master key (MinIO's
+        // `rotateKey` + `keyRotation` flag, which is why it may keep metadataOnly = true). RustFS
+        // has no such rewrap primitive today; adding one is backlog#1637, and it would enter here
+        // as an explicit exception rather than by relaxing this guard.
+        let copy_changes_encryption = src_info.is_encrypted() || effective_sse.is_some() || has_explicit_ssec;
+        if cp_src_dst_same && src_info.transitioned_object.tier.is_empty() && !copy_changes_encryption {
             src_info.metadata_only = true;
         }
 

rustfs/src/storage/sse.rs

@@ -3430,6 +3430,7 @@ mod tests {
         DARE_CIPHER_AES_256_GCM, DARE_CIPHER_CHACHA20_POLY1305, MINIO_INTERNAL_ENCRYPTION_SEAL_ALGORITHM, SEALED_KEY_IV_SIZE,
         SEALED_KEY_SIZE, is_legacy_rustfs_managed_metadata, is_supported_sealed_object_key_cipher,
     };
+    use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER;
 
     #[test]
     fn parse_simple_sse_cmk_rejects_bad_keys_without_crashing() {
@@ -4583,6 +4584,89 @@ mod tests {
         assert!(metadata.contains_key("content-type"));
     }
 
+    /// A same-name CopyObject is normally serviced as a metadata-only update that reuses the
+    /// stored ciphertext, while the handler rebuilds the destination metadata around a *fresh*
+    /// DEK. `ObjectInfo::is_encrypted` — i.e. `is_object_encryption_marker` — is the guard that
+    /// keeps that shortcut away from encrypted objects (see the `metadata_only` decision in
+    /// `rustfs/src/app/object_usecase.rs`). Any encryption flavour whose headers
+    /// `strip_managed_encryption_metadata` drops must therefore also be *detected* there — a
+    /// flavour that is stripped but not detected would resurrect "new DEK + old ciphertext",
+    /// which leaves the object permanently undecryptable.
+    #[test]
+    fn every_strippable_encryption_shape_is_detected_as_encrypted() {
+        let shapes: [(&str, Vec<(&str, &str)>); 4] = [
+            (
+                "sse-s3",
+                vec![
+                    ("x-amz-server-side-encryption", "AES256"),
+                    (MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER, "sealed"),
+                    (MINIO_INTERNAL_ENCRYPTION_IV_HEADER, "iv"),
+                ],
+            ),
+            (
+                "sse-kms",
+                vec![
+                    ("x-amz-server-side-encryption", "aws:kms"),
+                    (MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER, "sealed"),
+                    (MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER, "key-1"),
+                ],
+            ),
+            (
+                "legacy rustfs managed",
+                vec![
+                    ("x-amz-server-side-encryption", "AES256"),
+                    (INTERNAL_ENCRYPTION_KEY_HEADER, "wrapped"),
+                    (INTERNAL_ENCRYPTION_IV_HEADER, "iv"),
+                ],
+            ),
+            (
+                "sse-c",
+                vec![
+                    (SSEC_ALGORITHM_HEADER, "AES256"),
+                    (MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER, "sealed"),
+                ],
+            ),
+        ];
+
+        for (label, entries) in shapes {
+            let metadata: HashMap<String, String> = entries
+                .iter()
+                .map(|(key, value)| ((*key).to_string(), (*value).to_string()))
+                .collect();
+
+            assert!(
+                metadata
+                    .keys()
+                    .any(|key| rustfs_utils::http::is_object_encryption_marker(key)),
+                "{label}: an encrypted object must be detected, otherwise a same-name copy would re-key it \
+                 without rewriting the ciphertext"
+            );
+
+            let mut stripped = metadata.clone();
+            strip_managed_encryption_metadata(&mut stripped);
+            assert_ne!(
+                stripped, metadata,
+                "{label}: the copy path strips this shape's encryption metadata, so the destination cannot \
+                 reuse the source ciphertext"
+            );
+        }
+    }
+
+    #[test]
+    fn plain_object_metadata_is_not_flagged_as_encrypted() {
+        // The metadata-only self-copy shortcut must stay available for unencrypted objects.
+        let metadata: HashMap<String, String> = [("content-type", "text/plain"), ("x-amz-meta-stage", "before")]
+            .iter()
+            .map(|(key, value)| ((*key).to_string(), (*value).to_string()))
+            .collect();
+
+        assert!(
+            !metadata
+                .keys()
+                .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
+        );
+    }
+
     #[cfg(feature = "rio-v2")]
     #[test]
     fn test_legacy_managed_metadata_excludes_sealed_keys() {

scripts/check_fips_wording.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Guard: outward README and CHANGELOG material must not make an unsupported
+# FIPS validation or certification claim. The detailed policy and permitted
+# qualifiers live in docs/operations/kms-cryptographic-compliance.md; this
+# check intentionally scans only the two public project-facing documents.
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT_DIR="${CHECK_FIPS_WORDING_ROOT:-$(cd "${SCRIPT_DIR}/.." && pwd)}"
+
+cd "$ROOT_DIR"
+
+TARGETS=(README.md CHANGELOG.md)
+FORBIDDEN_PATTERNS=(
+  'FIPS[[:space:]]+(140-[23](/[[:space:]]*140-[23])?[[:space:]]+)?(validated|certified|compliant)'
+  'FIPS[[:space:]]+mode'
+  'FIPS-enabled'
+  'NIST[[:space:]]+(certified|approved)'
+  'CMVP[[:space:]]+certificate'
+  '(meets|satisfies)[[:space:]]+FIPS'
+)
+
+status=0
+
+for target in "${TARGETS[@]}"; do
+  if [[ ! -f "$target" ]]; then
+    printf 'FIPS wording guard failed: %s is missing\n' "$target" >&2
+    status=1
+    continue
+  fi
+
+  for pattern in "${FORBIDDEN_PATTERNS[@]}"; do
+    matches="$(grep -E -i -n -- "$pattern" "$target" || true)"
+    if [[ -n "$matches" ]]; then
+      printf 'FIPS wording guard failed: forbidden pattern /%s/ in %s:\n%s\n' \
+        "$pattern" "$target" "$matches" >&2
+      status=1
+    fi
+  done
+done
+
+if [[ "$status" -ne 0 ]]; then
+  printf 'Remove unsupported FIPS validation wording from README.md or CHANGELOG.md.\n' >&2
+  exit "$status"
+fi
+
+printf 'FIPS wording guard passed (README.md and CHANGELOG.md contain no forbidden claims).\n'