refactor(sse): decouple encryption from ecstore

crates/ecstore/Cargo.toml

@@ -57,7 +57,6 @@ rustfs-policy.workspace = true
 rustfs-protos.workspace = true
 rustfs-replication.workspace = true
 rustfs-lifecycle.workspace = true
-rustfs-kms.workspace = true
 rustfs-s3-types = { workspace = true }
 rustfs-data-usage.workspace = true
 rustfs-object-capacity.workspace = true
@@ -124,8 +123,6 @@ libc.workspace = true
 rustix = { workspace = true, features = ["process", "fs"] }
 rustfs-madmin.workspace = true
 reqwest = { workspace = true }
-aes-gcm = { workspace = true, features = ["rand_core"] }
-chacha20poly1305.workspace = true
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
 urlencoding = { workspace = true }
 smallvec = { workspace = true, features = ["serde"] }

crates/ecstore/src/api/mod.rs

@@ -381,10 +381,12 @@ pub mod notification {
 
 pub mod object {
     pub use crate::object_api::{
-        BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, GetObjectBodySource,
-        GetObjectReader, ObjectInfo, ObjectMutationHook, ObjectOptions, PutObjReader, RangedDecompressReader, StreamConsumer,
-        get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
-        register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
+        BLOCK_SIZE_V2, ERASURE_ALGORITHM, EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook,
+        GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReader, ObjectEncryptionResolver, ObjectInfo,
+        ObjectMutationHook, ObjectOptions, PutObjReader, RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode,
+        ReadEncryptionRequest, StreamConsumer, get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook,
+        register_get_object_body_cache_hook, register_object_mutation_hook, unregister_get_object_body_cache_hook,
+        unregister_object_mutation_hook,
     };
     pub use crate::store::PreparedGetObjectReader;
 }

crates/ecstore/src/client/utils.rs

@@ -46,16 +46,6 @@ lazy_static! {
         m.insert("x-amz-replication-status".to_string(), true);
         m
     };
-    static ref SSE_HEADERS: HashMap<String, bool> = {
-        let mut m = HashMap::new();
-        m.insert("x-amz-server-side-encryption".to_string(), true);
-        m.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), true);
-        m.insert("x-amz-server-side-encryption-context".to_string(), true);
-        m.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), true);
-        m.insert("x-amz-server-side-encryption-customer-key".to_string(), true);
-        m.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), true);
-        m
-    };
 }
 
 pub fn is_standard_query_value(qs_key: &str) -> bool {
@@ -70,16 +60,12 @@ pub fn is_standard_header(header_key: &str) -> bool {
     *SUPPORTED_HEADERS.get(&header_key.to_lowercase()).unwrap_or(&false)
 }
 
-pub fn is_sse_header(header_key: &str) -> bool {
-    *SSE_HEADERS.get(&header_key.to_lowercase()).unwrap_or(&false)
-}
-
 pub fn is_amz_header(header_key: &str) -> bool {
     let key = header_key.to_lowercase();
     key.starts_with("x-amz-meta-")
         || key.starts_with("x-amz-grant-")
         || key == "x-amz-acl"
-        || is_sse_header(header_key)
+        || rustfs_utils::http::is_sse_header(header_key)
         || key.starts_with("x-amz-checksum-")
 }
 

crates/ecstore/src/object_api/encryption.rs

@@ -0,0 +1,80 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use async_trait::async_trait;
+use http::{HeaderMap, HeaderValue};
+use std::collections::HashMap;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ReadEncryptionMode {
+    Direct { base_nonce: [u8; 12] },
+    Object,
+}
+
+pub struct ReadEncryptionMaterial {
+    pub key_bytes: [u8; 32],
+    pub mode: ReadEncryptionMode,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EncryptionResolutionErrorKind {
+    InvalidRequest,
+    InvalidMetadata,
+    ServiceUnavailable,
+    DecryptionFailed,
+}
+
+#[derive(Debug)]
+pub struct EncryptionResolutionError {
+    kind: EncryptionResolutionErrorKind,
+    message: String,
+}
+
+impl EncryptionResolutionError {
+    pub fn new(kind: EncryptionResolutionErrorKind, message: impl Into<String>) -> Self {
+        Self {
+            kind,
+            message: message.into(),
+        }
+    }
+
+    pub fn kind(&self) -> EncryptionResolutionErrorKind {
+        self.kind
+    }
+}
+
+impl Display for EncryptionResolutionError {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(&self.message)
+    }
+}
+
+impl Error for EncryptionResolutionError {}
+
+pub struct ReadEncryptionRequest<'a> {
+    pub bucket: &'a str,
+    pub object: &'a str,
+    pub metadata: &'a HashMap<String, String>,
+    pub headers: &'a HeaderMap<HeaderValue>,
+}
+
+#[async_trait]
+pub trait ObjectEncryptionResolver: Send + Sync {
+    async fn resolve_read_material(
+        &self,
+        request: ReadEncryptionRequest<'_>,
+    ) -> Result<Option<ReadEncryptionMaterial>, EncryptionResolutionError>;
+}

crates/ecstore/src/object_api/mod.rs

@@ -84,6 +84,7 @@ pub(crate) fn legacy_encrypted_range_seek_enabled() -> bool {
 }
 
 mod body_cache_hook;
+mod encryption;
 mod hook_slot;
 mod object_mutation_hook;
 mod readers;
@@ -98,6 +99,10 @@ pub use body_cache_hook::{
 pub(crate) use body_cache_hook::{
     get_object_body_cache_hook, get_object_body_cache_hook_suppressed, without_get_object_body_cache_hook,
 };
+pub use encryption::{
+    EncryptionResolutionError, EncryptionResolutionErrorKind, ObjectEncryptionResolver, ReadEncryptionMaterial,
+    ReadEncryptionMode, ReadEncryptionRequest,
+};
 pub(crate) use object_mutation_hook::notify_object_mutation;
 pub use object_mutation_hook::{ObjectMutationHook, register_object_mutation_hook, unregister_object_mutation_hook};
 pub use readers::*;

crates/ecstore/src/object_api/types.rs

@@ -271,29 +271,9 @@ impl ObjectInfo {
     }
 
     pub fn is_encrypted(&self) -> bool {
-        // Corresponding to the logic in rustfs/src/sse.rs/encryption_material_to_metadata function
-        use rustfs_utils::http::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
-
-        self.user_defined.keys().any(|key| {
-            let lower = key.to_ascii_lowercase();
-            lower.starts_with("x-minio-encryption-")
-                || lower.starts_with("x-minio-internal-server-side-encryption-")
-                || matches!(
-                    lower.as_str(),
-                    "x-minio-internal-encrypted-multipart"
-                        | "x-rustfs-encryption-key"
-                        | "x-rustfs-encryption-algorithm"
-                        | "x-rustfs-encryption-iv"
-                        | "x-rustfs-encryption-key-id"
-                        | "x-rustfs-encryption-context"
-                        | "x-rustfs-encryption-tag"
-                        | "x-amz-server-side-encryption-aws-kms-key-id"
-                        | SSEC_ALGORITHM_HEADER
-                        | SSEC_KEY_HEADER
-                        | SSEC_KEY_MD5_HEADER
-                        | "x-amz-server-side-encryption"
-                )
-        })
+        self.user_defined
+            .keys()
+            .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
     }
 
     /// Maximum inline size for non-versioned objects (128 KiB).
@@ -337,26 +317,7 @@ impl ObjectInfo {
     }
 
     pub fn encryption_original_size(&self) -> std::io::Result<Option<i64>> {
-        let actual_size = rustfs_utils::http::get_str(&self.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE);
-        if let Some(size_str) = self
-            .user_defined
-            .get("x-rustfs-encryption-original-size")
-            .map(String::as_str)
-            .or_else(|| {
-                self.user_defined
-                    .get("x-amz-server-side-encryption-customer-original-size")
-                    .map(String::as_str)
-            })
-            .or(actual_size.as_deref())
-            && !size_str.is_empty()
-        {
-            let size = size_str
-                .parse::<i64>()
-                .map_err(|e| std::io::Error::other(format!("Failed to parse encryption original size: {e}")))?;
-            return Ok(Some(size));
-        }
-
-        Ok(None)
+        rustfs_utils::http::get_object_encryption_original_size(&self.user_defined)
     }
 
     pub fn decrypted_size(&self) -> std::io::Result<i64> {
@@ -386,9 +347,6 @@ impl ObjectInfo {
             return Ok(actual_size);
         }
 
-        // Check if object is encrypted
-        // Managed SSE stores original size in x-rustfs-encryption-original-size metadata
-        // SSE-C stores original size in x-amz-server-side-encryption-customer-original-size
         if let Some(size) = self.encryption_original_size()? {
             return Ok(size);
         }
@@ -878,6 +836,19 @@ mod tests {
         assert!(!object.is_inline_fast_path_eligible(), "transitioned objects must fall back");
     }
 
+    #[test]
+    fn minio_internal_encryption_metadata_is_not_treated_as_plaintext() {
+        let object = ObjectInfo {
+            user_defined: Arc::new(HashMap::from([(
+                "X-Minio-Internal-Server-Side-Encryption-Sealed-Key".to_string(),
+                "sealed".to_string(),
+            )])),
+            ..Default::default()
+        };
+
+        assert!(object.is_encrypted());
+    }
+
     #[test]
     fn versions_after_marker_handles_null_version_marker() {
         let first_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").unwrap();

crates/ecstore/src/runtime/instance.rs

@@ -46,6 +46,7 @@ use crate::bucket::metadata_sys::BucketMetadataSys;
 use crate::bucket::replication::{DynReplicationPool, ReplicationStats};
 use crate::disk::DiskStore;
 use crate::layout::endpoints::{EndpointServerPools, SetupType};
+use crate::object_api::ObjectEncryptionResolver;
 use crate::services::event_notification::EventNotifier;
 use crate::services::tier::tier::TierConfigMgr;
 use rustfs_lock::{GlobalLockManager, get_global_lock_manager};
@@ -159,6 +160,8 @@ pub struct InstanceContext {
     /// workers (scanner/heal/tier/lifecycle) without touching another instance.
     /// Replaces the process-global cancel-token static.
     background_cancel_token: OnceLock<CancellationToken>,
+    /// Resolves object-encryption material at the application boundary.
+    object_encryption_resolver: OnceLock<Arc<dyn ObjectEncryptionResolver>>,
     tier_delete_journal_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
     transition_transaction_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
     #[cfg(test)]
@@ -197,6 +200,7 @@ impl InstanceContext {
             local_disk_set_drives: Arc::new(RwLock::new(Vec::new())),
             bucket_metadata_sys: std::sync::Mutex::new(None),
             background_cancel_token: OnceLock::new(),
+            object_encryption_resolver: OnceLock::new(),
             tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()),
             transition_transaction_recovery_stores: std::sync::Mutex::new(HashSet::new()),
             #[cfg(test)]
@@ -209,6 +213,19 @@ impl InstanceContext {
         self.lock_manager.clone()
     }
 
+    /// Install the application-owned object-encryption resolver once.
+    pub fn set_object_encryption_resolver(
+        &self,
+        resolver: Arc<dyn ObjectEncryptionResolver>,
+    ) -> Result<(), Arc<dyn ObjectEncryptionResolver>> {
+        self.object_encryption_resolver.set(resolver)
+    }
+
+    /// Return the configured object-encryption resolver, if startup installed one.
+    pub fn object_encryption_resolver(&self) -> Option<&dyn ObjectEncryptionResolver> {
+        self.object_encryption_resolver.get().map(Arc::as_ref)
+    }
+
     /// Set this instance's S3 region.
     ///
     /// Write-once: panics on a second write, preserving the startup fail-fast

crates/ecstore/src/runtime/sources.rs

@@ -46,7 +46,6 @@ use crate::{
 use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
 use rustfs_config::server_config::{Config, get_global_server_config, set_global_server_config};
 use rustfs_io_metrics::internode_metrics::global_internode_metrics;
-use rustfs_kms::{ObjectEncryptionService, get_global_encryption_service};
 use rustfs_lock::client::LockClient;
 use s3s::dto::BucketLifecycleConfiguration;
 use s3s::region::Region;
@@ -105,10 +104,6 @@ pub(crate) fn record_erasure_write_quorum_failure(stage: &'static str, dominant_
     global_internode_metrics().record_erasure_write_quorum_failure(stage, dominant_error);
 }
 
-pub(crate) async fn object_encryption_service() -> Option<Arc<ObjectEncryptionService>> {
-    get_global_encryption_service().await
-}
-
 pub fn object_store_handle() -> Option<Arc<ECStore>> {
     resolve_object_store_handle()
 }

crates/ecstore/src/set_disk/metadata.rs

@@ -505,9 +505,7 @@ impl SetDisks {
     }
 
     fn file_info_has_encryption_metadata(meta: &FileInfo) -> bool {
-        meta.metadata
-            .keys()
-            .any(|name| http::is_encryption_metadata_key(name) || http::is_sse_header(name))
+        meta.metadata.keys().any(|name| http::is_object_encryption_marker(name))
     }
 
     fn starts_with_ignore_ascii_case(value: &str, prefix: &str) -> bool {

crates/ecstore/src/set_disk/mod.rs

@@ -143,15 +143,17 @@ use rustfs_object_capacity::capacity_scope::{
     CapacityScope, CapacityScopeDisk, current_dirty_generation, record_capacity_scope, record_global_dirty_scope,
 };
 use rustfs_s3_types::EventName;
+#[cfg(test)]
+use rustfs_utils::http::SSEC_ALGORITHM_HEADER;
 use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
 use rustfs_utils::http::headers::AMZ_STORAGE_CLASS;
 use rustfs_utils::http::headers::{
     CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _,
 };
 use rustfs_utils::http::{
-    SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE,
-    SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, contains_key_str,
-    get_header_map, get_str, insert_str, is_encryption_metadata_key, remove_header_map,
+    SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
+    SUFFIX_RESTORE_OPERATION_ID, contains_key_str, get_header_map, get_str, insert_str, is_object_encryption_marker,
+    remove_header_map,
 };
 use rustfs_utils::{
     HashAlgorithm,
@@ -407,10 +409,7 @@ pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap<String, S
 }
 
 fn should_persist_encryption_original_size(metadata: &HashMap<String, String>) -> bool {
-    metadata.keys().any(|key| is_encryption_metadata_key(key))
-        || metadata.contains_key(SSEC_ALGORITHM_HEADER)
-        || metadata.contains_key(SSEC_KEY_HEADER)
-        || metadata.contains_key(SSEC_KEY_MD5_HEADER)
+    metadata.keys().any(|key| is_object_encryption_marker(key))
 }
 
 /// Per-set memoized capacity dirty scope.

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

@@ -39,6 +39,7 @@ use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppress
 use crate::services::tier::tier::{TierConfigMgr, TierOperationLease};
 use crate::store::ECStore;
 use futures::FutureExt as _;
+use http::HeaderValue;
 use std::future::Future;
 
 fn erasure_from_file_info(fi: &FileInfo, uses_legacy: bool) -> Result<coding::Erasure> {
@@ -46,6 +47,17 @@ fn erasure_from_file_info(fi: &FileInfo, uses_legacy: bool) -> Result<coding::Er
         .map_err(Error::from)
 }
 
+async fn get_object_reader_with_context(
+    ctx: &InstanceContext,
+    reader: Box<dyn AsyncRead + Unpin + Send + Sync>,
+    range: Option<HTTPRangeSpec>,
+    object_info: &ObjectInfo,
+    opts: &ObjectOptions,
+    headers: &HeaderMap<HeaderValue>,
+) -> Result<(GetObjectReader, usize, i64)> {
+    GetObjectReader::new_with_resolver(reader, range, object_info, opts, headers, ctx.object_encryption_resolver()).await
+}
+
 /// Length of the full plaintext body when — and only when — this read's output
 /// is exactly the object's complete plaintext, so the app-layer body cache may
 /// serve it in place of the erasure read.
@@ -704,7 +716,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                             size_bucket,
                         );
                         record_get_object_reader_path_observation(GET_OBJECT_PATH_CODEC_STREAMING, object_class, size_bucket);
-                        let (mut reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?;
+                        let (mut reader, _offset, _length) =
+                            get_object_reader_with_context(&self.ctx, stream, range, &object_info, opts, &h).await?;
                         // Carry the hook probe result so the app layer skips its
                         // now-redundant lookup on the streaming miss path (ODC-16).
                         reader.body_source = body_source;
@@ -742,7 +755,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let (rd, wd) = tokio::io::duplex(duplex_buffer_size);
         debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer");
 
-        let (mut reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h).await?;
+        let (mut reader, offset, length) =
+            get_object_reader_with_context(&self.ctx, Box::new(rd), range, &object_info, opts, &h).await?;
         // Carry the hook probe result so the app layer skips its now-redundant
         // lookup on the streaming miss path (ODC-16).
         reader.body_source = body_source;
@@ -4275,6 +4289,61 @@ mod erasure_construction_tests {
     }
 }
 
+#[cfg(test)]
+mod object_encryption_resolver_wiring_tests {
+    use super::*;
+    use crate::object_api::{EncryptionResolutionError, ObjectEncryptionResolver, ReadEncryptionMaterial, ReadEncryptionRequest};
+    use std::io::Cursor;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    struct CountingResolver {
+        calls: AtomicUsize,
+    }
+
+    #[async_trait::async_trait]
+    impl ObjectEncryptionResolver for CountingResolver {
+        async fn resolve_read_material(
+            &self,
+            _request: ReadEncryptionRequest<'_>,
+        ) -> std::result::Result<Option<ReadEncryptionMaterial>, EncryptionResolutionError> {
+            self.calls.fetch_add(1, Ordering::Relaxed);
+            Ok(None)
+        }
+    }
+
+    #[tokio::test]
+    async fn get_object_reader_forwards_instance_resolver() {
+        let resolver = Arc::new(CountingResolver {
+            calls: AtomicUsize::new(0),
+        });
+        let ctx = InstanceContext::new();
+        assert!(
+            ctx.set_object_encryption_resolver(resolver.clone()).is_ok(),
+            "fresh context should accept resolver"
+        );
+        let object_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            size: 1,
+            user_defined: Arc::new(HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())])),
+            ..Default::default()
+        };
+
+        let result = get_object_reader_with_context(
+            &ctx,
+            Box::new(Cursor::new(Vec::<u8>::new())),
+            None,
+            &object_info,
+            &ObjectOptions::default(),
+            &HeaderMap::new(),
+        )
+        .await;
+
+        assert!(result.is_err(), "resolver returning no material must fail closed");
+        assert_eq!(resolver.calls.load(Ordering::Relaxed), 1);
+    }
+}
+
 #[cfg(test)]
 pub(in crate::set_disk::ops) mod hermetic_set_disks_support {
     //! Shared hermetic `SetDisks` construction for the ops tests below: the

crates/ecstore/tests/README.md

@@ -2,7 +2,7 @@
 
 ## MinIO-generated encrypted fixtures
 
-`minio_generated_read_test.rs` validates the `bitrot -> GetObjectReader` path against raw MinIO backend data captured by
+`rustfs/src/storage/minio_generated_read_test.rs` validates the `bitrot -> GetObjectReader` path against raw MinIO backend data captured by
 `.\rustfs\scripts\minio_fixture_lab\lab.py`.
 
 It currently covers multipart fixtures for:
@@ -20,5 +20,5 @@ Example:
 ```powershell
 $env:RUSTFS_MINIO_FIXTURE_ROOT = '.\rustfs\tmp\minio-fixture-lab-local-key'
 $env:RUSTFS_MINIO_STATIC_KMS_KEY_B64 = '<base64-32-byte-local-minio-kms-key>'
-cargo +1.97.1 test -p rustfs-ecstore --features rio-v2 --test minio_generated_read_test -- --ignored
+cargo +1.97.1 test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored
 ```

crates/ecstore/tests/minio_generated_read_test.rs

@@ -1,299 +0,0 @@
-#![cfg(feature = "rio-v2")]
-
-use std::fs;
-use std::io::Cursor;
-use std::path::{Path, PathBuf};
-
-mod storage_api;
-
-use rustfs_filemeta::{FileInfo, FileInfoOpts, get_file_info};
-use serde::Deserialize;
-use sha2::{Digest, Sha256};
-use storage_api::minio_generated_read::{
-    DiskAPI as _, DiskOption, Endpoint, Erasure, GetObjectReader, ObjectInfo, ObjectOptions, create_bitrot_reader, new_disk,
-};
-use temp_env::async_with_vars;
-use tokio::io::{AsyncReadExt, AsyncWrite};
-
-#[derive(Debug, Deserialize)]
-struct ManifestRecord {
-    bucket: String,
-    object: String,
-    backend_files: Vec<String>,
-}
-
-#[derive(Default)]
-struct VecAsyncWriter {
-    bytes: Vec<u8>,
-}
-
-impl AsyncWrite for VecAsyncWriter {
-    fn poll_write(
-        mut self: std::pin::Pin<&mut Self>,
-        _cx: &mut std::task::Context<'_>,
-        buf: &[u8],
-    ) -> std::task::Poll<std::io::Result<usize>> {
-        self.bytes.extend_from_slice(buf);
-        std::task::Poll::Ready(Ok(buf.len()))
-    }
-
-    fn poll_flush(self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<()>> {
-        std::task::Poll::Ready(Ok(()))
-    }
-
-    fn poll_shutdown(self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<()>> {
-        std::task::Poll::Ready(Ok(()))
-    }
-}
-
-fn fixture_root() -> PathBuf {
-    std::env::var_os("RUSTFS_MINIO_FIXTURE_ROOT")
-        .map(PathBuf::from)
-        .unwrap_or_else(|| PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../rio-v2/tests/fixtures/minio-generated"))
-}
-
-fn case_dir(case_id: &str) -> PathBuf {
-    fixture_root().join("cases").join(case_id)
-}
-
-fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> T {
-    let text = fs::read_to_string(path).unwrap_or_else(|err| panic!("read {}: {err}", path.display()));
-    serde_json::from_str(&text).unwrap_or_else(|err| panic!("parse {}: {err}", path.display()))
-}
-
-fn require_fixture_case(case_id: &str) -> PathBuf {
-    let path = case_dir(case_id);
-    assert!(
-        path.is_dir(),
-        "fixture case missing: {}. Run scripts/minio_fixture_lab/lab.py capture-matrix first.",
-        path.display()
-    );
-    path
-}
-
-fn read_plaintext_sha256(case_dir: &Path) -> String {
-    fs::read_to_string(case_dir.join("plaintext.sha256"))
-        .unwrap_or_else(|err| panic!("read plaintext.sha256 under {}: {err}", case_dir.display()))
-        .trim()
-        .to_string()
-}
-
-fn minio_static_kms_key_b64() -> String {
-    std::env::var("RUSTFS_MINIO_STATIC_KMS_KEY_B64")
-        .unwrap_or_else(|_| panic!("RUSTFS_MINIO_STATIC_KMS_KEY_B64 must point to the 32-byte static MinIO KMS key"))
-}
-
-fn object_xl_meta_path(case_dir: &Path, manifest: &ManifestRecord) -> PathBuf {
-    let expected = format!("disk1/{}/{}/xl.meta", manifest.bucket, manifest.object);
-    let relative = manifest
-        .backend_files
-        .iter()
-        .find(|entry| entry.as_str() == expected)
-        .unwrap_or_else(|| panic!("object xl.meta missing from manifest backend_files: {expected}"));
-    case_dir.join("backend").join(relative)
-}
-
-fn load_file_info(case_dir: &Path, manifest: &ManifestRecord) -> FileInfo {
-    let xl_meta_path = object_xl_meta_path(case_dir, manifest);
-    let xl_meta = fs::read(&xl_meta_path).unwrap_or_else(|err| panic!("read {}: {err}", xl_meta_path.display()));
-    get_file_info(
-        &xl_meta,
-        &manifest.bucket,
-        &manifest.object,
-        "",
-        FileInfoOpts {
-            data: true,
-            include_free_versions: true,
-        },
-    )
-    .unwrap_or_else(|err| panic!("decode {}: {err}", xl_meta_path.display()))
-}
-
-fn load_object_info(file_info: &FileInfo, manifest: &ManifestRecord) -> ObjectInfo {
-    ObjectInfo::from_file_info(file_info, &manifest.bucket, &manifest.object, false)
-}
-
-fn sha256_hex(bytes: &[u8]) -> String {
-    hex_simd::encode_to_string(Sha256::digest(bytes), hex_simd::AsciiCase::Lower)
-}
-
-async fn load_fixture_reader_input(case_id: &str) -> (ObjectInfo, Vec<u8>, String) {
-    let case_dir = require_fixture_case(case_id);
-    let manifest: ManifestRecord = read_json(&case_dir.join("manifest.json"));
-    let expected_sha256 = read_plaintext_sha256(&case_dir);
-    let file_info = load_file_info(&case_dir, &manifest);
-    let encrypted = encrypted_fixture_bytes(&case_dir, &manifest, &file_info).await;
-    let object_info = load_object_info(&file_info, &manifest);
-
-    (object_info, encrypted, expected_sha256)
-}
-
-async fn read_fixture_plaintext(encrypted: Vec<u8>, object_info: ObjectInfo, kms_key_b64: String) -> Result<Vec<u8>, String> {
-    let object_size = object_info.size;
-
-    async_with_vars(
-        [
-            ("__RUSTFS_SSE_SIMPLE_CMK", Some(kms_key_b64)),
-            ("RUSTFS_SSE_S3_MASTER_KEY", None::<String>),
-        ],
-        async move {
-            let (mut reader, offset, length) = GetObjectReader::new(
-                Box::new(Cursor::new(encrypted)),
-                None,
-                &object_info,
-                &ObjectOptions::default(),
-                &http::HeaderMap::new(),
-            )
-            .await
-            .map_err(|err| format!("construct GetObjectReader from MinIO raw fixture: {err:?}"))?;
-
-            if offset != 0 || length != object_size {
-                return Err(format!("unexpected fixture range offset={offset} length={length} size={object_size}"));
-            }
-
-            let mut plaintext = Vec::new();
-            reader
-                .read_to_end(&mut plaintext)
-                .await
-                .map_err(|err| format!("read plaintext from MinIO raw fixture: {err}"))?;
-
-            Ok(plaintext)
-        },
-    )
-    .await
-}
-
-async fn encrypted_fixture_bytes(case_dir: &Path, manifest: &ManifestRecord, file_info: &FileInfo) -> Vec<u8> {
-    let mut disks = Vec::with_capacity(file_info.erasure.distribution.len());
-    for disk_number in 1..=file_info.erasure.distribution.len() {
-        let disk_root = case_dir.join("backend").join(format!("disk{disk_number}"));
-        let disk_root_str = disk_root
-            .to_str()
-            .unwrap_or_else(|| panic!("non-utf8 disk root {}", disk_root.display()));
-        let mut endpoint = Endpoint::try_from(disk_root_str).expect("fixture disk endpoint");
-        endpoint.set_pool_index(0);
-        endpoint.set_set_index(0);
-        endpoint.set_disk_index(disk_number - 1);
-        let disk = new_disk(
-            &endpoint,
-            &DiskOption {
-                cleanup: false,
-                health_check: false,
-            },
-        )
-        .await
-        .unwrap_or_else(|err| panic!("open fixture disk {disk_number}: {err}"));
-        disks.push(disk);
-    }
-    let mut disk_order = vec![None; disks.len()];
-    for (idx, disk) in disks.iter().enumerate() {
-        let block_index = file_info.erasure.distribution[idx];
-        disk_order[block_index - 1] = Some(disk);
-    }
-    let data_dir = file_info
-        .data_dir
-        .as_ref()
-        .unwrap_or_else(|| panic!("fixture {} is missing data_dir", manifest.object));
-
-    let mut encrypted = Vec::new();
-    for part in &file_info.parts {
-        let checksum_info = file_info.erasure.get_checksum_info(part.number);
-        let path = format!("{}/{}/part.{}", manifest.object, data_dir, part.number);
-        let shard_read_len = file_info.erasure.shard_file_size(part.size as i64);
-        let mut readers = Vec::with_capacity(disks.len());
-        for (idx, disk) in disk_order.iter().enumerate() {
-            let reader = create_bitrot_reader(
-                None,
-                *disk,
-                &manifest.bucket,
-                &path,
-                0,
-                shard_read_len as usize,
-                file_info.erasure.shard_size(),
-                checksum_info.algorithm.clone(),
-                false,
-                false,
-            )
-            .await
-            .unwrap_or_else(|err| panic!("create bitrot reader for disk{} {path}: {err:?}", idx + 1));
-            readers.push(reader);
-        }
-
-        let erasure = Erasure::new(
-            file_info.erasure.data_blocks,
-            file_info.erasure.parity_blocks,
-            file_info.erasure.block_size,
-        );
-        let mut writer = VecAsyncWriter::default();
-        let (written, err) = erasure.decode(&mut writer, readers, 0, part.size, part.size).await;
-        if let Some(err) = err {
-            panic!("decode erasure shards for {path}: {err}");
-        }
-        assert_eq!(written, part.size, "decoded part size should match xl.meta part size");
-        encrypted.extend_from_slice(&writer.bytes);
-    }
-    for disk in disks {
-        disk.close().await.expect("close fixture disk");
-    }
-    encrypted
-}
-
-#[tokio::test]
-#[ignore = "requires generated MinIO fixture data and a local static KMS key"]
-async fn reads_minio_generated_sse_s3_multipart_fixture() {
-    assert_fixture_round_trip("sse-s3-multipart-8m", 8 * 1024 * 1024).await;
-}
-
-#[tokio::test]
-#[ignore = "requires generated MinIO fixture data and a local static KMS key"]
-async fn reads_minio_generated_sse_kms_multipart_fixture() {
-    assert_fixture_round_trip("sse-kms-multipart-8m", 8 * 1024 * 1024).await;
-}
-
-#[tokio::test]
-#[ignore = "requires generated MinIO fixture data and a local static KMS key"]
-async fn rejects_minio_generated_sse_s3_fixture_with_wrong_kms_key() {
-    let (object_info, encrypted, _) = load_fixture_reader_input("sse-s3-multipart-8m").await;
-    let wrong_key_b64 = "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=".to_string();
-
-    let result = read_fixture_plaintext(encrypted, object_info, wrong_key_b64).await;
-
-    assert!(result.is_err(), "wrong KMS key must fail closed");
-}
-
-#[tokio::test]
-#[ignore = "requires generated MinIO fixture data and a local static KMS key"]
-async fn rejects_minio_generated_sse_s3_fixture_with_truncated_ciphertext() {
-    let (object_info, mut encrypted, expected_sha256) = load_fixture_reader_input("sse-s3-multipart-8m").await;
-    encrypted.truncate(encrypted.len() / 2);
-
-    let result = read_fixture_plaintext(encrypted, object_info, minio_static_kms_key_b64()).await;
-
-    if let Ok(plaintext) = result {
-        assert_ne!(
-            sha256_hex(&plaintext),
-            expected_sha256,
-            "truncated ciphertext must not restore the original plaintext"
-        );
-    }
-}
-
-async fn assert_fixture_round_trip(case_id: &str, expected_size: i64) {
-    let (object_info, encrypted, expected_sha256) = load_fixture_reader_input(case_id).await;
-    // `ObjectInfo.size` is the on-disk size. For SSE objects that is the
-    // DARE-encrypted size (plaintext + 32 bytes per 64 KiB block), which is
-    // deliberately larger than the logical object size. The size a client sees
-    // (and what MinIO records via `x-*-internal-actual-size`) comes from
-    // `decrypted_size()`/`get_actual_size()`, so assert against that — the raw
-    // `size` field would never equal the plaintext length for encrypted objects.
-    let decrypted_size = object_info.decrypted_size().expect("decrypted size from MinIO metadata");
-    let kms_key_b64 = minio_static_kms_key_b64();
-
-    let plaintext = read_fixture_plaintext(encrypted, object_info, kms_key_b64)
-        .await
-        .expect("fixture must restore with the configured KMS key");
-
-    assert_eq!(decrypted_size, expected_size);
-    assert_eq!(plaintext.len(), expected_size as usize);
-    assert_eq!(sha256_hex(&plaintext), expected_sha256);
-}

crates/rio-v2/tests/minio_fixture_lab/README.md

@@ -137,7 +137,7 @@ tests read):
 ./capture_via_docker.sh
 
 RUSTFS_MINIO_STATIC_KMS_KEY_B64=IyqsU3kMFloCNup4BsZtf/rmfHVcTgznO2F25CkEH1g= \
-  cargo test -p rustfs-ecstore --features rio-v2 --test minio_generated_read_test -- --ignored
+  cargo test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored
 ```
 
 This is exactly what the nightly `minio-interop` GitHub Actions workflow runs

crates/utils/src/http/header_compat.rs

@@ -25,6 +25,11 @@ const RUSTFS_PREFIX: &str = "x-rustfs-";
 const MINIO_PREFIX: &str = "x-minio-";
 const MINIO_ENCRYPTION_PREFIX: &str = "x-minio-encryption-";
 const RUSTFS_ENCRYPTION_PREFIX: &str = "x-rustfs-encryption-";
+const MINIO_INTERNAL_ENCRYPTION_PREFIX: &str = "x-minio-internal-server-side-encryption-";
+const MINIO_INTERNAL_ENCRYPTED_MULTIPART: &str = "x-minio-internal-encrypted-multipart";
+const RUSTFS_ENCRYPTION_ORIGINAL_SIZE: &str = "x-rustfs-encryption-original-size";
+const MINIO_ENCRYPTION_ORIGINAL_SIZE: &str = "x-minio-encryption-original-size";
+const SSEC_ORIGINAL_SIZE: &str = "x-amz-server-side-encryption-customer-original-size";
 
 // Suffix constants (part after x-rustfs- or x-minio-). Use with get_header/insert_header.
 pub const SUFFIX_FORCE_DELETE: &str = "force-delete";
@@ -40,11 +45,49 @@ pub const SUFFIX_SOURCE_REPLICATION_REQUEST: &str = "source-replication-request"
 pub const SUFFIX_SOURCE_REPLICATION_CHECK: &str = "source-replication-check";
 pub const SUFFIX_REPLICATION_SSEC_CRC: &str = "replication-ssec-crc";
 
-/// Returns true if the key is an internal encryption metadata key (x-rustfs-encryption-* or
-/// x-minio-encryption-*). Case-insensitive for metadata filtering.
+/// Returns true if the key is object-encryption metadata understood by RustFS or MinIO.
+/// Case-insensitive for metadata filtering.
 pub fn is_encryption_metadata_key(key: &str) -> bool {
     let lower = key.to_lowercase();
-    lower.starts_with(RUSTFS_ENCRYPTION_PREFIX) || lower.starts_with(MINIO_ENCRYPTION_PREFIX)
+    lower.starts_with(RUSTFS_ENCRYPTION_PREFIX)
+        || lower.starts_with(MINIO_ENCRYPTION_PREFIX)
+        || lower.starts_with(MINIO_INTERNAL_ENCRYPTION_PREFIX)
+        || lower == MINIO_INTERNAL_ENCRYPTED_MULTIPART
+}
+
+/// Returns true when a metadata key proves that object data is encrypted.
+///
+/// Original-size metadata alone is not proof: older plaintext objects can
+/// retain that compatibility field after metadata migration.
+pub fn is_object_encryption_marker(key: &str) -> bool {
+    (is_encryption_metadata_key(key)
+        && !key.eq_ignore_ascii_case(RUSTFS_ENCRYPTION_ORIGINAL_SIZE)
+        && !key.eq_ignore_ascii_case(MINIO_ENCRYPTION_ORIGINAL_SIZE))
+        || super::is_sse_header(key)
+}
+
+/// Reads the logical object size recorded by encryption metadata.
+pub fn get_object_encryption_original_size(metadata: &std::collections::HashMap<String, String>) -> std::io::Result<Option<i64>> {
+    let actual_size = super::get_str(metadata, super::SUFFIX_ACTUAL_SIZE);
+    let size = get_case_insensitive(metadata, RUSTFS_ENCRYPTION_ORIGINAL_SIZE)
+        .or_else(|| get_case_insensitive(metadata, SSEC_ORIGINAL_SIZE))
+        .or(actual_size.as_deref());
+
+    let Some(size) = size.filter(|size| !size.is_empty()) else {
+        return Ok(None);
+    };
+    size.parse::<i64>()
+        .map(Some)
+        .map_err(|error| std::io::Error::other(format!("Failed to parse encryption original size: {error}")))
+}
+
+fn get_case_insensitive<'a>(metadata: &'a std::collections::HashMap<String, String>, key: &str) -> Option<&'a str> {
+    metadata.get(key).map(String::as_str).or_else(|| {
+        metadata
+            .iter()
+            .find(|(candidate, _)| candidate.eq_ignore_ascii_case(key))
+            .map(|(_, value)| value.as_str())
+    })
 }
 
 fn rustfs_key(suffix: &str) -> String {
@@ -106,10 +149,37 @@ mod tests {
         assert!(is_encryption_metadata_key("x-rustfs-encryption-iv"));
         assert!(is_encryption_metadata_key("X-Rustfs-Encryption-Key"));
         assert!(is_encryption_metadata_key("x-minio-encryption-iv"));
+        assert!(is_encryption_metadata_key("X-Minio-Internal-Server-Side-Encryption-Sealed-Key"));
+        assert!(is_encryption_metadata_key("X-Minio-Internal-Encrypted-Multipart"));
         assert!(!is_encryption_metadata_key("x-amz-meta-custom"));
         assert!(!is_encryption_metadata_key("x-rustfs-internal-healing"));
     }
 
+    #[test]
+    fn object_encryption_marker_excludes_size_only_metadata() {
+        assert!(!is_object_encryption_marker(RUSTFS_ENCRYPTION_ORIGINAL_SIZE));
+        assert!(is_object_encryption_marker("X-Minio-Internal-Server-Side-Encryption-Sealed-Key"));
+        assert!(is_object_encryption_marker("x-amz-server-side-encryption"));
+    }
+
+    #[test]
+    fn object_encryption_original_size_is_case_insensitive() {
+        let metadata = std::collections::HashMap::from([(
+            "X-Amz-Server-Side-Encryption-Customer-Original-Size".to_string(),
+            "42".to_string(),
+        )]);
+        assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42));
+    }
+
+    #[test]
+    fn object_encryption_original_size_prefers_rustfs_metadata() {
+        let metadata = std::collections::HashMap::from([
+            (SSEC_ORIGINAL_SIZE.to_string(), "21".to_string()),
+            (RUSTFS_ENCRYPTION_ORIGINAL_SIZE.to_string(), "42".to_string()),
+        ]);
+        assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42));
+    }
+
     #[test]
     fn test_get_header() {
         let mut headers = HeaderMap::new();