diff --git a/crates/e2e_test/src/kms/kms_ilm_sse_kms_test.rs b/crates/e2e_test/src/kms/kms_ilm_sse_kms_test.rs index e1fad39cd..173e87767 100644 --- a/crates/e2e_test/src/kms/kms_ilm_sse_kms_test.rs +++ b/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(); diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 72d9cca86..e0849a637 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/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 { 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 { @@ -4422,6 +4421,10 @@ fn validate_transition_remote_version(oi: &ObjectInfo) -> Result>, + resolver: Option<&dyn ObjectEncryptionResolver>, ) -> Result { 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 { @@ -5776,6 +5788,7 @@ mod tests { &object_info, &ObjectOptions::default(), &manager, + None, ) .await .expect("transitioned reader should open"); @@ -5840,6 +5853,7 @@ mod tests { &object_info, &ObjectOptions::default(), &manager, + None, ) .await { @@ -5880,6 +5894,7 @@ mod tests { &object_info, &ObjectOptions::default(), &manager, + None, ) .await { @@ -6117,6 +6132,7 @@ mod tests { &oi, &ObjectOptions::default(), &manager, + None, ) .await { @@ -6140,6 +6156,7 @@ mod tests { &oi, &ObjectOptions::default(), &manager, + None, ) .await { diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index cc3a1986a..6e5985c10 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/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, + oi: &ObjectInfo, + opts: &ObjectOptions, + h: &HeaderMap, + resolver: Option<&dyn ObjectEncryptionResolver>, + ) -> Result { + 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, + oi: &ObjectInfo, + ) -> Result { + self.into_reader(reader, oi).map(|(reader, _, _)| reader) + } + #[cfg(test)] async fn build(rs: Option, oi: &ObjectInfo, opts: &ObjectOptions, h: &HeaderMap) -> Result { Self::build_with_resolver(rs, oi, opts, h, Some(&tests::TEST_RESOLVER)).await @@ -500,8 +545,17 @@ impl ReadPlan { resolver: Option<&dyn ObjectEncryptionResolver>, ) -> Result { 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); } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 95fe3438a..db77a1f8b 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -899,6 +899,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)); @@ -6065,6 +6066,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 { @@ -6134,6 +6136,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)?;