Merge commit from fork

Cargo.lock

@@ -10594,6 +10594,7 @@ dependencies = [
  "faster-hex",
  "futures",
  "hex-simd",
+ "hmac 0.13.0",
  "hotpath",
  "http 1.5.0",
  "http-body-util",

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/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(),
         }
     }
@@ -9205,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.clone();
+        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/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();

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/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/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>) {