fix(tier): decrypt transitioned objects instead of serving their ciphertext

A GET on a managed-SSE object that lifecycle had transitioned to a remote tier returned the ciphertext with the plaintext's Content-Length and no error: silent corruption on read-through, and worse than a failed request because nothing signals it. Restore of the same object failed server-side with IncompleteBody while POST ?restore still answered 200, so the object simply never came back and HEAD never showed an x-amz-restore marker.

Both symptoms are one cause. The transitioned read path built its fetch through new_getobjectreader, which decides nothing about encryption: it derived the range from the parts table — whose sizes are PLAINTEXT sizes — then used that range to fetch the object's STORED bytes from the tier, and handed the stream to the caller without any decrypt transform. The GET therefore served the first plaintext-length bytes of ciphertext; the restore copy-back, which validates against the stored size, came up short by exactly the encryption overhead.

The path now builds the same ReadPlan the local read path uses, so a single place decides how stored bytes map to requested bytes. ReadPlan gains a two-phase API — build_for_request to learn the storage coordinates before issuing the tier fetch, into_object_reader to wrap the returned stream — because the tier fetch has to be positioned before a stream exists. The encryption resolver reaches the path from InstanceContext, the same source the local read uses.

A restore read additionally stops synthesizing a range from the part number. A restore serves the stored representation (restore_request_active already forces the Plain branch), so a plaintext-coordinate range would be reinterpreted as a storage range and truncate the payload by its encoding overhead. An explicit caller range is already in storage coordinates on that path and is still honored, which two existing tests pin.

crates/e2e_test/src/kms/kms_ilm_sse_kms_test.rs drops its #[ignore]: the transition test now runs and asserts the plaintext round-trips byte-identically through transition, read-through and restore. The same file had its enforcement switch stuck at false from a control experiment; it is back to true, so the test again exercises what its name and module docs claim.

Fixes #6025. Refs rustfs/backlog#1582, rustfs/backlog#1637.

crates/e2e_test/src/kms/kms_ilm_sse_kms_test.rs

@@ -97,7 +97,7 @@ async fn start_enforcing_ilm_server(env: &mut LocalKMSTestEnvironment) -> TestRe
 
     let envs = [
         ("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true"),
-        ("RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY", "false"),
+        ("RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY", "true"),
         ("RUSTFS_SCANNER_CYCLE", "1"),
         ("RUSTFS_ILM_PROCESS_TIME", "1"),
         ("RUSTFS_ILM_DEBUG_DAY_SECS", "2"),
@@ -486,7 +486,6 @@ async fn ilm_expiration_on_sse_kms_bucket_under_enforcement() -> TestResult {
 /// depend on scanner scheduling; the 1s scanner cycle stays on as a backstop.
 #[tokio::test]
 #[serial]
-#[ignore = "pins rustfs/rustfs#6025: GET on a transitioned managed-SSE object silently returns corrupt bytes (fails with enforcement on AND off, so it is not an authorization regression); un-ignore with the fix"]
 async fn ilm_transition_on_sse_kms_bucket_under_enforcement_reads_back() -> TestResult {
     init_logging();
 

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

@@ -46,15 +46,13 @@ use crate::bucket::lifecycle::transition_transaction::run_transition_transaction
 use crate::bucket::object_lock::ObjectLockApi;
 use crate::bucket::versioning::VersioningApi as _;
 use crate::bucket::versioning_sys::BucketVersioningSys;
-use crate::client::object_api_utils::new_getobjectreader;
 use crate::disk::error::DiskError;
 use crate::disk::{DeleteOptions, Disk, DiskAPI, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, STORAGE_FORMAT_FILE};
 use crate::error::Error;
 use crate::error::StorageError;
-use crate::error::{
-    error_resp_to_object_err, is_err_object_not_found, is_err_read_quorum, is_err_version_not_found, is_network_or_host_down,
-};
+use crate::error::{is_err_object_not_found, is_err_read_quorum, is_err_version_not_found, is_network_or_host_down};
 use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions};
+use crate::object_api::{ObjectEncryptionResolver, ReadPlan};
 use crate::services::tier::{
     tier::{TierConfigMgr, TierOperationLease, tier_destination_id_from_metadata},
     warm_backend::WarmBackendGetOpts,
@@ -4400,9 +4398,10 @@ pub async fn get_transitioned_object_reader(
     h: &HeaderMap,
     oi: &ObjectInfo,
     opts: &ObjectOptions,
+    resolver: Option<&dyn ObjectEncryptionResolver>,
 ) -> Result<GetObjectReader, std::io::Error> {
     let tier_config_mgr = runtime_sources::tier_config_mgr_handle();
-    get_transitioned_object_reader_with_tier_manager(bucket, object, rs, h, oi, opts, &tier_config_mgr).await
+    get_transitioned_object_reader_with_tier_manager(bucket, object, rs, h, oi, opts, &tier_config_mgr, resolver).await
 }
 
 fn validate_transition_remote_version(oi: &ObjectInfo) -> Result<bool, std::io::Error> {
@@ -4422,6 +4421,10 @@ fn validate_transition_remote_version(oi: &ObjectInfo) -> Result<bool, std::io::
     }
 }
 
+// The resolver joins the tier manager as the second injected port this read
+// needs; grouping the request half into a struct would churn every call site of
+// a bug fix.
+#[allow(clippy::too_many_arguments)]
 pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
     bucket: &str,
     object: &str,
@@ -4430,6 +4433,7 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
     oi: &ObjectInfo,
     opts: &ObjectOptions,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
+    resolver: Option<&dyn ObjectEncryptionResolver>,
 ) -> Result<GetObjectReader, std::io::Error> {
     validate_transition_remote_version(oi)?;
     let expected_identity = tier_destination_id_from_metadata(&oi.user_defined)?;
@@ -4447,11 +4451,16 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
 
     tgt_client.validate_remote_version_id(&oi.transitioned_object.version_id)?;
 
-    let ret = new_getobjectreader(rs, oi, opts, h);
-    if let Err(err) = ret {
-        return Err(error_resp_to_object_err(err, vec![bucket, object]));
-    }
-    let (get_fn, off, length) = ret.expect("get_transitioned_object_reader should succeed after error check");
+    // The same read plan the local path uses, so the tier fetch is positioned in
+    // the object's *stored* coordinate system and the stream is handed the same
+    // decrypt/decompress transforms. Reading an encrypted object's ciphertext
+    // through a plaintext-coordinate range and skipping the transform is how a
+    // transitioned SSE object used to come back as silently corrupt bytes of the
+    // right length (rustfs/rustfs#6025).
+    let plan = ReadPlan::build_for_request(rs.clone(), oi, opts, h, resolver)
+        .await
+        .map_err(|err| std::io::Error::other(format!("building the read plan for {bucket}/{object} failed: {err}")))?;
+    let (off, length) = (plan.storage_offset() as i64, plan.storage_length());
     let mut gopts = WarmBackendGetOpts::default();
 
     if off >= 0 && length >= 0 {
@@ -4488,7 +4497,10 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
             );
             e
         })?;
-    Ok(attach_tier_operation_lease(get_fn(reader, h.clone()), tgt_client))
+    let object_reader = plan
+        .into_object_reader(Box::new(reader), oi)
+        .map_err(|err| std::io::Error::other(format!("wrapping the tier stream for {bucket}/{object} failed: {err}")))?;
+    Ok(attach_tier_operation_lease(object_reader, tgt_client))
 }
 
 struct TierOperationLeaseReader {

crates/ecstore/src/object_api/readers.rs

@@ -479,7 +479,15 @@ enum ReadTransform {
     },
 }
 
-struct ReadPlan {
+/// How an object's stored bytes must be fetched and transformed to serve a
+/// request.
+///
+/// Public so callers that fetch the stored bytes from somewhere other than the
+/// local erasure set — the remote-tier read path — can position their own fetch
+/// with [`ReadPlan::storage_offset`] / [`ReadPlan::storage_length`] and then
+/// hand the resulting stream to [`ReadPlan::into_object_reader`], instead of
+/// reimplementing the transform decisions (rustfs/rustfs#6025).
+pub struct ReadPlan {
     storage_offset: usize,
     storage_length: i64,
     object_size: i64,
@@ -487,6 +495,43 @@ struct ReadPlan {
 }
 
 impl ReadPlan {
+    /// Byte offset into the object's **stored** bytes where the fetch must
+    /// start. Encrypted and compressed objects address their storage in a
+    /// different coordinate system than the plaintext range the caller asked
+    /// for, which is exactly the distinction this plan resolves.
+    pub fn storage_offset(&self) -> usize {
+        self.storage_offset
+    }
+
+    /// Number of **stored** bytes the fetch must deliver, in the same
+    /// coordinate system as [`Self::storage_offset`].
+    pub fn storage_length(&self) -> i64 {
+        self.storage_length
+    }
+
+    /// Build the plan for a request without consuming a stream, so a caller
+    /// that has to issue its own positioned fetch can read the offsets first.
+    pub async fn build_for_request(
+        rs: Option<HTTPRangeSpec>,
+        oi: &ObjectInfo,
+        opts: &ObjectOptions,
+        h: &HeaderMap<HeaderValue>,
+        resolver: Option<&dyn ObjectEncryptionResolver>,
+    ) -> Result<Self> {
+        Self::build_with_resolver(rs, oi, opts, h, resolver).await
+    }
+
+    /// Wrap `reader` — the stored bytes this plan asked for, already positioned
+    /// at [`Self::storage_offset`] — in the transforms that turn them into the
+    /// bytes the caller requested.
+    pub fn into_object_reader(
+        self,
+        reader: Box<dyn AsyncRead + Unpin + Send + Sync>,
+        oi: &ObjectInfo,
+    ) -> Result<GetObjectReader> {
+        self.into_reader(reader, oi).map(|(reader, _, _)| reader)
+    }
+
     #[cfg(test)]
     async fn build(rs: Option<HTTPRangeSpec>, oi: &ObjectInfo, opts: &ObjectOptions, h: &HeaderMap<HeaderValue>) -> Result<Self> {
         Self::build_with_resolver(rs, oi, opts, h, Some(&tests::TEST_RESOLVER)).await
@@ -500,8 +545,17 @@ impl ReadPlan {
         resolver: Option<&dyn ObjectEncryptionResolver>,
     ) -> Result<Self> {
         let mut rs = rs;
+        // A part number addresses the object's PLAINTEXT bytes. A restore read
+        // serves the stored representation instead (see
+        // [`restore_request_active`]), where that synthesized range would be
+        // reinterpreted as a storage range and truncate an encrypted or
+        // compressed payload by exactly its encoding overhead — the copy-back
+        // then fails its length check partway through
+        // (rustfs/rustfs#6025). An explicit caller range is already in storage
+        // coordinates on that path and is still honored.
         if let Some(part_number) = opts.part_number
             && rs.is_none()
+            && !restore_request_active(opts)
         {
             rs = http_range_spec_from_object_info(oi, part_number);
         }

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

@@ -903,6 +903,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 &object_info,
                 &opts,
                 &self.ctx.tier_config_mgr(),
+                self.ctx.object_encryption_resolver(),
             )
             .await?;
             return Ok(finish_set_disk_read_lock(gr, read_lock_guard.take(), bucket, object));
@@ -5824,6 +5825,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 &oi,
                 &opts,
                 &self_.ctx.tier_config_mgr(),
+                self_.ctx.object_encryption_resolver(),
             )
             .await;
             if let Err(err) = gr {
@@ -5893,6 +5895,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                     &oi,
                     &part_opts,
                     &self_.ctx.tier_config_mgr(),
+                    self_.ctx.object_encryption_resolver(),
                 )
                 .await
                 .map_err(StorageError::Io)?;