From 09c2d150572e5ac8311063c278d4b5649ef92595 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sat, 9 May 2026 22:06:35 +0800 Subject: [PATCH] fix(sse): Temporarily refactored the SSE design for ECStore (#2813) Co-authored-by: houseme Co-authored-by: cxymds --- .vscode/launch.json | 1 + Cargo.lock | 2 + crates/ecstore/Cargo.toml | 2 + crates/ecstore/src/set_disk.rs | 42 +- crates/ecstore/src/store_api/readers.rs | 824 ++++++++++++++++++++++- crates/ecstore/src/store_api/types.rs | 41 +- crates/rio/src/encrypt_reader.rs | 8 + crates/rio/src/lib.rs | 2 +- rustfs/src/app/multipart_usecase.rs | 371 +++++----- rustfs/src/app/object_usecase.rs | 156 +---- rustfs/src/storage/mod.rs | 2 +- rustfs/src/storage/sse.rs | 857 +++++++----------------- 12 files changed, 1338 insertions(+), 970 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index fb1187cbb..5f33c194a 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -188,6 +188,7 @@ "RUSTFS_CONSOLE_ENABLE": "true", "RUSTFS_CONSOLE_ADDRESS": "127.0.0.1:9001", "RUSTFS_OBS_LOG_DIRECTORY": "./target/logs", + "RUSTFS_UNSAFE_BYPASS_DISK_CHECK": "true", // "RUSTFS_OBS_TRACE_ENDPOINT": "http://127.0.0.1:4318/v1/traces", // jeager otlp http endpoint // "RUSTFS_OBS_METRIC_ENDPOINT": "http://127.0.0.1:4318/v1/metrics", // default otlp http endpoint // "RUSTFS_OBS_LOG_ENDPOINT": "http://127.0.0.1:4318/v1/logs", // default otlp http endpoint diff --git a/Cargo.lock b/Cargo.lock index dda1fe56f..9cbb481f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8803,6 +8803,7 @@ dependencies = [ name = "rustfs-ecstore" version = "1.0.0-beta.2" dependencies = [ + "aes-gcm", "async-channel", "async-recursion", "async-trait", @@ -8860,6 +8861,7 @@ dependencies = [ "rustfs-credentials", "rustfs-filemeta", "rustfs-io-metrics", + "rustfs-kms", "rustfs-lock", "rustfs-madmin", "rustfs-policy", diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 0cdc9447a..d438982cf 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -44,6 +44,7 @@ rustfs-credentials = { workspace = true } rustfs-common.workspace = true rustfs-policy.workspace = true rustfs-protos.workspace = true +rustfs-kms.workspace = true rustfs-s3-common = { workspace = true } async-trait.workspace = true bytes.workspace = true @@ -104,6 +105,7 @@ rustix = { workspace = true } rustfs-madmin.workspace = true rustfs-workers.workspace = true reqwest = { workspace = true } +aes-gcm.workspace = true aws-sdk-s3 = { workspace = true } urlencoding = { workspace = true } smallvec = { workspace = true } diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 24c10267f..c139270e2 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -88,8 +88,9 @@ use rustfs_utils::http::headers::{ CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _, }; use rustfs_utils::http::{ - SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, - contains_key_str, get_header_map, get_str, insert_str, remove_header_map, + 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, contains_key_str, get_header_map, get_str, + insert_str, is_encryption_metadata_key, remove_header_map, }; use rustfs_utils::{ HashAlgorithm, @@ -134,6 +135,13 @@ pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap) -> 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) +} + fn capacity_scope_from_disks(disks: &[Option]) -> CapacityScope { let mut unique = HashSet::with_capacity(disks.len()); let mut scoped_disks = Vec::with_capacity(disks.len()); @@ -694,7 +702,7 @@ impl 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 (reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h)?; + let (reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h).await?; // let disks = disks.clone(); let bucket = bucket.to_owned(); @@ -3510,16 +3518,22 @@ impl MultipartOperations for SetDisks { fi.metadata.insert("etag".to_owned(), etag); + let persist_encryption_original_size = should_persist_encryption_original_size(&fi.metadata); + if opts.replication_request { if let Some(actual_size) = get_str(&opts.user_defined, SUFFIX_ACTUAL_OBJECT_SIZE_CAP) { insert_str(&mut fi.metadata, SUFFIX_ACTUAL_SIZE, actual_size.clone()); - fi.metadata - .insert("x-rustfs-encryption-original-size".to_string(), actual_size); + if persist_encryption_original_size { + fi.metadata + .insert("x-rustfs-encryption-original-size".to_string(), actual_size); + } } } else { insert_str(&mut fi.metadata, SUFFIX_ACTUAL_SIZE, object_actual_size.to_string()); - fi.metadata - .insert("x-rustfs-encryption-original-size".to_string(), object_actual_size.to_string()); + if persist_encryption_original_size { + fi.metadata + .insert("x-rustfs-encryption-original-size".to_string(), object_actual_size.to_string()); + } } if fi.is_compressed() { @@ -5721,6 +5735,20 @@ mod tests { .expect("GOVERNANCE shortening with bypass should remain allowed"); } + #[test] + fn test_should_persist_encryption_original_size_rejects_plain_metadata() { + let metadata = HashMap::from([("content-type".to_string(), "application/octet-stream".to_string())]); + + assert!(!should_persist_encryption_original_size(&metadata)); + } + + #[test] + fn test_should_persist_encryption_original_size_accepts_sse_c_metadata() { + let metadata = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]); + + assert!(should_persist_encryption_original_size(&metadata)); + } + #[test] fn test_should_prevent_write() { let oi = ObjectInfo { diff --git a/crates/ecstore/src/store_api/readers.rs b/crates/ecstore/src/store_api/readers.rs index 2ca554cfa..30ec0c0e3 100644 --- a/crates/ecstore/src/store_api/readers.rs +++ b/crates/ecstore/src/store_api/readers.rs @@ -1,4 +1,79 @@ use super::*; +use aes_gcm::{ + Aes256Gcm, Key, Nonce, + aead::{Aead, KeyInit}, +}; +use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD}; +use md5::{Digest, Md5}; +use rustfs_kms::{service_manager::get_global_encryption_service, types::ObjectEncryptionContext}; +use rustfs_rio::DecryptReader; +use rustfs_utils::http::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER}; +use std::collections::HashMap; +use std::env; + +const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id"; +const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key"; +const INTERNAL_ENCRYPTION_IV_HEADER: &str = "x-rustfs-encryption-iv"; +const INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER: &str = "x-rustfs-encryption-original-size"; +const SSEC_ORIGINAL_SIZE_HEADER: &str = "x-amz-server-side-encryption-customer-original-size"; +const DEFAULT_SSE_ALGORITHM: &str = "AES256"; + +fn part_plaintext_size(part: &ObjectPartInfo) -> i64 { + if part.actual_size > 0 { + part.actual_size + } else { + part.size as i64 + } +} + +fn restore_request_active(opts: &ObjectOptions) -> bool { + let restore = &opts.transition.restore_request; + restore.type_.is_some() || restore.days.is_some() || restore.output_location.is_some() || restore.select_parameters.is_some() +} + +fn decode_compression_index(index: Option<&bytes::Bytes>) -> Option { + let bytes = index?; + let mut decoded = rustfs_rio::Index::new(); + if decoded.load(bytes.as_ref()).is_ok() { + Some(decoded) + } else { + None + } +} + +fn get_compressed_offsets(oi: &ObjectInfo, offset: i64) -> (i64, i64, usize, i64, u64) { + let mut skip_length = 0_i64; + let mut cumulative_actual_size = 0_i64; + let mut first_part_idx = 0_usize; + let mut compressed_offset = 0_i64; + + for (i, part) in oi.parts.iter().enumerate() { + cumulative_actual_size += part_plaintext_size(part); + if cumulative_actual_size <= offset { + compressed_offset += part.size as i64; + } else { + first_part_idx = i; + skip_length = cumulative_actual_size - part_plaintext_size(part); + break; + } + } + + let mut part_skip = offset - skip_length; + let decrypt_skip = 0_i64; + let seq_num = 0_u64; + + if part_skip > 0 + && let Some(part) = oi.parts.get(first_part_idx) + && let Some(index) = decode_compression_index(part.index.as_ref()) + && let Ok((comp_off, uncomp_off)) = index.find(part_skip) + && comp_off > 0 + { + compressed_offset += comp_off; + part_skip -= uncomp_off; + } + + (compressed_offset, part_skip, first_part_idx, decrypt_skip, seq_num) +} pub struct PutObjReader { pub stream: HashReader, @@ -46,14 +121,19 @@ pub struct GetObjectReader { pub object_info: ObjectInfo, } +#[derive(Debug, Clone, Copy)] +struct EncryptionMaterial { + key_bytes: [u8; 32], + base_nonce: [u8; 12], +} + impl GetObjectReader { - #[tracing::instrument(level = "debug", skip(reader, rs, opts, _h))] - pub fn new( + pub async fn new( reader: Box, rs: Option, oi: &ObjectInfo, opts: &ObjectOptions, - _h: &HeaderMap, + h: &HeaderMap, ) -> Result<(Self, usize, i64)> { let mut rs = rs; @@ -63,25 +143,27 @@ impl GetObjectReader { rs = HTTPRangeSpec::from_object_info(oi, part_number); } - // TODO:Encrypted + let mut is_encrypted = oi.is_encrypted(); + let (algo, mut is_compressed) = oi.is_compressed_ok()?; - let (algo, is_compressed) = oi.is_compressed_ok()?; + if restore_request_active(opts) { + is_encrypted = false; + is_compressed = false; + } - // TODO: check TRANSITION - - if is_compressed { + if is_compressed && !is_encrypted { let actual_size = oi.get_actual_size()?; let (off, length, dec_off, dec_length) = if let Some(rs) = rs { - // Support range requests for compressed objects - let (dec_off, dec_length) = rs.get_offset_length(actual_size)?; - (0, oi.size, dec_off, dec_length) + let (req_off, req_length) = rs.get_offset_length(actual_size)?; + let (physical_off, decompressed_skip, _, _, _) = get_compressed_offsets(oi, req_off as i64); + (physical_off as usize, oi.size - physical_off, decompressed_skip as usize, req_length) } else { (0, oi.size, 0, actual_size) }; let dec_reader = DecompressReader::new(reader, algo); - let actual_size_usize = if actual_size > 0 { + let actual_size_usize = if actual_size >= 0 { actual_size as usize } else { return Err(Error::other(format!("invalid decompressed size {actual_size}"))); @@ -122,6 +204,65 @@ impl GetObjectReader { )); } + if is_encrypted { + let material = resolve_encryption_material(oi, h).await?; + let is_multipart = is_multipart_encrypted_object(&oi.parts, oi.etag.as_deref()); + let plaintext_size = encrypted_plaintext_size(oi, is_multipart, is_compressed)?; + let plaintext_size_usize = + usize::try_from(plaintext_size).map_err(|_| Error::other(format!("invalid decrypted size {plaintext_size}")))?; + let (plain_offset, plain_length) = if let Some(rs) = rs { + rs.get_offset_length(plaintext_size)? + } else { + (0, plaintext_size) + }; + + let decrypted_reader: Box = if is_multipart { + Box::new(DecryptReader::new_multipart( + reader, + material.key_bytes, + material.base_nonce, + multipart_part_numbers(&oi.parts), + )) + } else { + Box::new(DecryptReader::new(reader, material.key_bytes, material.base_nonce)) + }; + + let final_reader: Box = if is_compressed { + let decompressed_reader = DecompressReader::new(decrypted_reader, algo); + if plain_offset > 0 || plain_length != plaintext_size { + Box::new(RangedDecompressReader::new( + decompressed_reader, + plain_offset, + plain_length, + plaintext_size_usize, + )?) + } else { + Box::new(LimitReader::new(decompressed_reader, plaintext_size_usize)) + } + } else if plain_offset > 0 || plain_length != plaintext_size { + Box::new(RangedDecompressReader::new( + decrypted_reader, + plain_offset, + plain_length, + plaintext_size_usize, + )?) + } else { + Box::new(LimitReader::new(decrypted_reader, plaintext_size_usize)) + }; + + let mut object_info = oi.clone(); + object_info.size = plain_length; + + return Ok(( + GetObjectReader { + stream: final_reader, + object_info, + }, + 0, + oi.size, + )); + } + if let Some(rs) = rs { let (off, length) = rs.get_offset_length(oi.size)?; @@ -188,7 +329,7 @@ impl HTTPRangeSpec { for i in 0..part_number { let part = &oi.parts[i]; start = end + 1; - end = start + (part.size as i64) - 1; + end = start + part_plaintext_size(part) - 1; } Some(HTTPRangeSpec { @@ -481,12 +622,239 @@ impl Drop for StreamConsumer { } } +fn encrypted_plaintext_size(oi: &ObjectInfo, is_multipart: bool, is_compressed: bool) -> Result { + if is_compressed { + return oi.get_actual_size().map_err(Into::into); + } + + if is_multipart { + return Ok(multipart_plaintext_size(&oi.parts, oi.decrypted_size()?)); + } + + oi.decrypted_size().map_err(Into::into) +} + +fn is_multipart_encrypted_object(parts: &[rustfs_filemeta::ObjectPartInfo], etag: Option<&str>) -> bool { + if parts.len() > 1 { + return true; + } + + etag.map(|etag| etag.trim_matches('"').len() != 32).unwrap_or(false) +} + +fn multipart_plaintext_size(parts: &[rustfs_filemeta::ObjectPartInfo], fallback: i64) -> i64 { + let total: i64 = parts.iter().map(part_plaintext_size).sum(); + + if total > 0 { total } else { fallback } +} + +fn multipart_part_numbers(parts: &[rustfs_filemeta::ObjectPartInfo]) -> Vec { + parts.iter().map(|part| part.number).collect() +} + +async fn resolve_encryption_material(oi: &ObjectInfo, headers: &HeaderMap) -> Result { + if oi.user_defined.contains_key(SSEC_ALGORITHM_HEADER) { + return resolve_ssec_material(oi, headers); + } + + if oi.user_defined.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER) { + return resolve_managed_material(&oi.user_defined).await; + } + + Err(Error::other("encrypted object metadata is incomplete")) +} + +fn resolve_ssec_material(oi: &ObjectInfo, headers: &HeaderMap) -> Result { + let algorithm = headers + .get(SSEC_ALGORITHM_HEADER) + .ok_or_else(|| Error::other("missing SSE-C algorithm header"))? + .to_str() + .map_err(|_| Error::other("invalid SSE-C algorithm header"))?; + if algorithm != DEFAULT_SSE_ALGORITHM { + return Err(Error::other(format!("unsupported SSE-C algorithm {algorithm}"))); + } + + let key_b64 = headers + .get(SSEC_KEY_HEADER) + .ok_or_else(|| Error::other("missing SSE-C key header"))? + .to_str() + .map_err(|_| Error::other("invalid SSE-C key header"))?; + let key_md5 = headers + .get(SSEC_KEY_MD5_HEADER) + .ok_or_else(|| Error::other("missing SSE-C key md5 header"))? + .to_str() + .map_err(|_| Error::other("invalid SSE-C key md5 header"))?; + + let key_bytes_vec = BASE64_STANDARD + .decode(key_b64) + .map_err(|_| Error::other("failed to decode SSE-C key"))?; + let key_bytes: [u8; 32] = key_bytes_vec + .try_into() + .map_err(|_| Error::other("SSE-C key must be 32 bytes"))?; + + let expected_md5 = BASE64_STANDARD.encode(md5_bytes(key_bytes)); + if expected_md5 != key_md5 { + return Err(Error::other("SSE-C key MD5 mismatch")); + } + + let stored_md5 = oi + .user_defined + .get(SSEC_KEY_MD5_HEADER) + .ok_or_else(|| Error::other("missing stored SSE-C key md5"))?; + if stored_md5 != &expected_md5 { + return Err(Error::other("SSE-C key does not match object metadata")); + } + + Ok(EncryptionMaterial { + key_bytes, + base_nonce: generate_ssec_nonce(&oi.bucket, &oi.name), + }) +} + +async fn resolve_managed_material(metadata: &HashMap) -> Result { + let encrypted_dek = metadata + .get(INTERNAL_ENCRYPTION_KEY_HEADER) + .ok_or_else(|| Error::other("missing managed encrypted DEK"))?; + let encrypted_dek = BASE64_STANDARD + .decode(encrypted_dek) + .map_err(|e| Error::other(format!("failed to decode managed encrypted DEK: {e}")))?; + + let iv_b64 = metadata + .get(INTERNAL_ENCRYPTION_IV_HEADER) + .ok_or_else(|| Error::other("missing managed encryption IV"))?; + let iv = BASE64_STANDARD + .decode(iv_b64) + .map_err(|e| Error::other(format!("failed to decode managed encryption IV: {e}")))?; + let base_nonce: [u8; 12] = iv + .as_slice() + .try_into() + .map_err(|_| Error::other("managed encryption IV must be 12 bytes"))?; + + let kms_key_id = metadata + .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER) + .map(String::as_str) + .unwrap_or("default"); + + let key_bytes = if let Some(service) = get_global_encryption_service().await { + service + .decrypt_data_key(&encrypted_dek, &ObjectEncryptionContext::new(String::new(), String::new())) + .await + .map_err(|e| Error::other(format!("failed to decrypt managed data key: {e}")))? + .plaintext_key + } else { + decrypt_local_sse_dek(&encrypted_dek, kms_key_id)? + }; + + Ok(EncryptionMaterial { key_bytes, base_nonce }) +} + +fn decrypt_local_sse_dek(encrypted_dek: &[u8], _kms_key_id: &str) -> Result<[u8; 32]> { + let encrypted_dek = std::str::from_utf8(encrypted_dek).map_err(|_| Error::other("managed DEK is not valid UTF-8"))?; + let parts: Vec<&str> = encrypted_dek.split(':').collect(); + if parts.len() != 2 { + return Err(Error::other("invalid managed DEK format")); + } + + let nonce_vec = BASE64_STANDARD + .decode(parts[0]) + .map_err(|_| Error::other("invalid managed DEK nonce"))?; + let ciphertext = BASE64_STANDARD + .decode(parts[1]) + .map_err(|_| Error::other("invalid managed DEK ciphertext"))?; + + let nonce_array: [u8; 12] = nonce_vec + .as_slice() + .try_into() + .map_err(|_| Error::other("invalid managed DEK nonce length"))?; + + let key = Key::::from(local_sse_master_key()?); + let cipher = Aes256Gcm::new(&key); + let plaintext = cipher + .decrypt(&Nonce::from(nonce_array), ciphertext.as_slice()) + .map_err(|e| Error::other(format!("failed to decrypt managed DEK: {e}")))?; + + plaintext + .as_slice() + .try_into() + .map_err(|_| Error::other("managed DEK has invalid plaintext length")) +} + +fn local_sse_master_key() -> Result<[u8; 32]> { + if let Some(key) = decode_master_key_env("__RUSTFS_SSE_SIMPLE_CMK")? { + return Ok(key); + } + + if let Some(key) = decode_master_key_env("RUSTFS_SSE_S3_MASTER_KEY")? { + return Ok(key); + } + + Ok([0u8; 32]) +} + +fn decode_master_key_env(name: &str) -> Result> { + let Ok(value) = env::var(name) else { + return Ok(None); + }; + + let value = value.trim(); + if value.is_empty() { + return Ok(None); + } + + let decoded = BASE64_STANDARD + .decode(value) + .map_err(|e| Error::other(format!("{name} is not valid base64: {e}")))?; + let key = + <[u8; 32]>::try_from(decoded.as_slice()).map_err(|_| Error::other(format!("{name} must decode to exactly 32 bytes")))?; + + Ok(Some(key)) +} + +fn generate_ssec_nonce(bucket: &str, key: &str) -> [u8; 12] { + let digest = md5_bytes(format!("{bucket}-{key}").as_bytes()); + let mut nonce = [0u8; 12]; + nonce.copy_from_slice(&digest[..12]); + nonce +} + +fn md5_bytes(data: impl AsRef<[u8]>) -> [u8; 16] { + let digest = Md5::digest(data.as_ref()); + let mut out = [0u8; 16]; + out.copy_from_slice(&digest); + out +} + #[cfg(test)] mod tests { use super::*; + use base64::Engine; + use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; + use md5::{Digest, Md5}; use std::io::Cursor; + use temp_env::async_with_vars; use tokio::io::AsyncReadExt; + fn md5_bytes(data: impl AsRef<[u8]>) -> [u8; 16] { + let digest = Md5::digest(data.as_ref()); + let mut bytes = [0u8; 16]; + bytes.copy_from_slice(&digest); + bytes + } + + fn ssec_headers_from_key(key_bytes: [u8; 32]) -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert(rustfs_utils::http::SSEC_ALGORITHM_HEADER, HeaderValue::from_static("AES256")); + headers.insert( + rustfs_utils::http::SSEC_KEY_HEADER, + HeaderValue::from_str(&BASE64_STANDARD.encode(key_bytes)).expect("valid base64 header"), + ); + headers.insert( + rustfs_utils::http::SSEC_KEY_MD5_HEADER, + HeaderValue::from_str(&BASE64_STANDARD.encode(md5_bytes(key_bytes))).expect("valid md5 header"), + ); + headers + } + #[tokio::test] async fn test_ranged_decompress_reader() { // Create test data @@ -626,6 +994,76 @@ mod tests { assert!(HTTPRangeSpec::from_object_info(&object_info, 4).is_none()); } + #[test] + fn test_http_range_spec_from_object_info_uses_actual_size() { + let object_info = ObjectInfo { + size: 90, + parts: vec![ + ObjectPartInfo { + etag: String::new(), + number: 1, + size: 20, + actual_size: 30, + ..Default::default() + }, + ObjectPartInfo { + etag: String::new(), + number: 2, + size: 30, + actual_size: 40, + ..Default::default() + }, + ObjectPartInfo { + etag: String::new(), + number: 3, + size: 40, + actual_size: 50, + ..Default::default() + }, + ], + ..Default::default() + }; + + let spec = HTTPRangeSpec::from_object_info(&object_info, 2).unwrap(); + assert_eq!(spec.start, 30); + assert_eq!(spec.end, 69); + } + + #[test] + fn test_http_range_spec_from_object_info_falls_back_to_part_size_when_actual_size_missing() { + let object_info = ObjectInfo { + size: 90, + parts: vec![ + ObjectPartInfo { + etag: String::new(), + number: 1, + size: 20, + actual_size: 0, + ..Default::default() + }, + ObjectPartInfo { + etag: String::new(), + number: 2, + size: 30, + actual_size: 40, + ..Default::default() + }, + ObjectPartInfo { + etag: String::new(), + number: 3, + size: 40, + actual_size: 0, + ..Default::default() + }, + ], + ..Default::default() + }; + + let spec = HTTPRangeSpec::from_object_info(&object_info, 3).unwrap(); + assert_eq!(spec.start, 60); + assert_eq!(spec.end, 99); + } + #[tokio::test] async fn test_ranged_decompress_reader_zero_length() { let original_data = b"Hello, World!"; @@ -676,11 +1114,22 @@ mod tests { assert_eq!(&buf2[..1], b"e"); } - #[test] - fn test_get_object_reader_range_uses_stored_size_for_encrypted_metadata() { + fn encrypt_managed_dek_for_test(dek: [u8; 32], master_key: [u8; 32]) -> String { + let key = Key::::from(master_key); + let cipher = Aes256Gcm::new(&key); + let nonce = Nonce::from([0u8; 12]); + let ciphertext = cipher.encrypt(&nonce, dek.as_slice()).expect("encrypt managed dek"); + format!("{}:{}", BASE64_STANDARD.encode(nonce), BASE64_STANDARD.encode(ciphertext)) + } + + #[tokio::test] + async fn test_get_object_reader_rejects_ssec_read_without_headers() { let object_info = ObjectInfo { size: 10, - user_defined: HashMap::from([("x-amz-server-side-encryption-customer-original-size".to_string(), "20".to_string())]), + user_defined: HashMap::from([ + ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()), + ("x-amz-server-side-encryption-customer-original-size".to_string(), "20".to_string()), + ]), ..Default::default() }; @@ -690,24 +1139,26 @@ mod tests { end: -1, }; - let (_, offset, length) = GetObjectReader::new( + let result = GetObjectReader::new( Box::new(Cursor::new(b"0123456789".to_vec())), Some(range), &object_info, &ObjectOptions::default(), &HeaderMap::new(), ) - .unwrap(); + .await; - assert_eq!(offset, 8); - assert_eq!(length, 2); + assert!(result.is_err()); } - #[test] - fn test_get_object_reader_suffix_range_uses_stored_size_for_encrypted_metadata() { + #[tokio::test] + async fn test_get_object_reader_restore_request_bypasses_encryption_range_rewrite() { let object_info = ObjectInfo { size: 10, - user_defined: HashMap::from([("x-rustfs-encryption-original-size".to_string(), "20".to_string())]), + user_defined: HashMap::from([ + ("x-rustfs-encryption-key".to_string(), "encrypted-key".to_string()), + ("x-rustfs-encryption-original-size".to_string(), "20".to_string()), + ]), ..Default::default() }; @@ -717,16 +1168,341 @@ mod tests { end: -1, }; + let mut opts = ObjectOptions::default(); + opts.transition.restore_request.days = Some(1); + let (_, offset, length) = GetObjectReader::new( Box::new(Cursor::new(b"0123456789".to_vec())), Some(range), &object_info, - &ObjectOptions::default(), + &opts, &HeaderMap::new(), ) + .await .unwrap(); assert_eq!(offset, 6); assert_eq!(length, 4); } + + #[tokio::test] + async fn test_get_object_reader_allows_encrypted_full_object_passthrough() { + async_with_vars([("__RUSTFS_SSE_SIMPLE_CMK", Some(BASE64_STANDARD.encode([0u8; 32])))], async { + let plaintext = b"managed-full-object".to_vec(); + let data_key = [0x21; 32]; + let base_nonce = [0x11; 12]; + let encrypted_dek = encrypt_managed_dek_for_test(data_key, [0u8; 32]); + + let mut encrypted = Vec::new(); + rustfs_rio::EncryptReader::new(Cursor::new(plaintext.clone()), data_key, base_nonce) + .read_to_end(&mut encrypted) + .await + .expect("encrypt managed object"); + + let object_info = ObjectInfo { + size: encrypted.len() as i64, + user_defined: HashMap::from([ + ("x-amz-server-side-encryption".to_string(), "AES256".to_string()), + ("x-rustfs-encryption-key".to_string(), BASE64_STANDARD.encode(encrypted_dek.as_bytes())), + ("x-rustfs-encryption-iv".to_string(), BASE64_STANDARD.encode(base_nonce)), + ("x-rustfs-encryption-original-size".to_string(), plaintext.len().to_string()), + ]), + ..Default::default() + }; + + let (mut reader, offset, length) = GetObjectReader::new( + Box::new(Cursor::new(encrypted.clone())), + None, + &object_info, + &ObjectOptions::default(), + &HeaderMap::new(), + ) + .await + .expect("managed encrypted full-object reads should decrypt inside ecstore"); + + let mut actual = Vec::new(); + reader.read_to_end(&mut actual).await.expect("read managed plaintext"); + + assert_eq!(offset, 0); + assert_eq!(length, object_info.size); + assert_eq!(reader.object_info.size, plaintext.len() as i64); + assert_eq!(actual, plaintext); + }) + .await; + } + + #[tokio::test] + async fn test_get_object_reader_uses_local_managed_fallback_without_env() { + async_with_vars( + [ + ("__RUSTFS_SSE_SIMPLE_CMK", None::), + ("RUSTFS_SSE_S3_MASTER_KEY", None::), + ], + async { + let plaintext = b"managed-local-fallback".to_vec(); + let data_key = [0x22; 32]; + let base_nonce = [0x12; 12]; + let encrypted_dek = encrypt_managed_dek_for_test(data_key, [0u8; 32]); + + let mut encrypted = Vec::new(); + rustfs_rio::EncryptReader::new(Cursor::new(plaintext.clone()), data_key, base_nonce) + .read_to_end(&mut encrypted) + .await + .expect("encrypt managed object with local fallback key"); + + let object_info = ObjectInfo { + size: encrypted.len() as i64, + user_defined: HashMap::from([ + ("x-amz-server-side-encryption".to_string(), "AES256".to_string()), + ("x-rustfs-encryption-key".to_string(), BASE64_STANDARD.encode(encrypted_dek.as_bytes())), + ("x-rustfs-encryption-iv".to_string(), BASE64_STANDARD.encode(base_nonce)), + ("x-rustfs-encryption-original-size".to_string(), plaintext.len().to_string()), + ]), + ..Default::default() + }; + + let (mut reader, _, _) = GetObjectReader::new( + Box::new(Cursor::new(encrypted)), + None, + &object_info, + &ObjectOptions::default(), + &HeaderMap::new(), + ) + .await + .expect("managed encrypted reads should fall back to the local SSE-S3 key"); + + let mut actual = Vec::new(); + reader.read_to_end(&mut actual).await.expect("read managed plaintext"); + + assert_eq!(reader.object_info.size, plaintext.len() as i64); + assert_eq!(actual, plaintext); + }, + ) + .await; + } + + #[tokio::test] + async fn test_get_object_reader_compressed_range_returns_physical_offset_from_index() { + let mut index = rustfs_rio::Index::new(); + index.add(0, 0).unwrap(); + index.add(1_048_576, 2_097_152).unwrap(); + + let object_info = ObjectInfo { + size: 3_000_000, + parts: vec![ObjectPartInfo { + etag: String::new(), + number: 1, + size: 3_000_000, + actual_size: 4_194_304, + index: Some(index.into_vec()), + ..Default::default() + }], + user_defined: HashMap::from([ + ("x-minio-internal-compression".to_string(), "gzip".to_string()), + ("x-minio-internal-actual-size".to_string(), "4194304".to_string()), + ]), + ..Default::default() + }; + + let range = HTTPRangeSpec { + is_suffix_length: false, + start: 2_097_152, + end: 2_097_161, + }; + + let (reader, offset, length) = GetObjectReader::new( + Box::new(Cursor::new(Vec::::new())), + Some(range), + &object_info, + &ObjectOptions::default(), + &HeaderMap::new(), + ) + .await + .unwrap(); + + assert!(offset > 0); + assert!(offset < 2_097_152); + assert_eq!(length, object_info.size - offset as i64); + assert_eq!(reader.object_info.size, 10); + } + + #[tokio::test] + async fn test_get_object_reader_decrypts_ssec_full_object() { + let plaintext = b"ecstore-ssec-full-object".to_vec(); + let key_bytes = [0x31; 32]; + let bucket = "bucket"; + let object = "object"; + let nonce = md5_bytes(format!("{bucket}-{object}").as_bytes()); + let mut base_nonce = [0u8; 12]; + base_nonce.copy_from_slice(&nonce[..12]); + + let mut encrypted = Vec::new(); + rustfs_rio::EncryptReader::new(Cursor::new(plaintext.clone()), key_bytes, base_nonce) + .read_to_end(&mut encrypted) + .await + .expect("encrypt object"); + + let object_info = ObjectInfo { + bucket: bucket.to_string(), + name: object.to_string(), + size: encrypted.len() as i64, + user_defined: HashMap::from([ + ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()), + ( + "x-amz-server-side-encryption-customer-key-md5".to_string(), + BASE64_STANDARD.encode(md5_bytes(key_bytes)), + ), + ( + "x-amz-server-side-encryption-customer-original-size".to_string(), + plaintext.len().to_string(), + ), + ]), + ..Default::default() + }; + + let (mut reader, offset, length) = GetObjectReader::new( + Box::new(Cursor::new(encrypted.clone())), + None, + &object_info, + &ObjectOptions::default(), + &ssec_headers_from_key(key_bytes), + ) + .await + .expect("ssec read should be supported"); + + let mut actual = Vec::new(); + reader.read_to_end(&mut actual).await.expect("read decrypted ssec object"); + + assert_eq!(offset, 0); + assert_eq!(length, encrypted.len() as i64); + assert_eq!(reader.object_info.size, plaintext.len() as i64); + assert_eq!(actual, plaintext); + } + + #[tokio::test] + async fn test_get_object_reader_decrypts_ssec_range_on_plaintext_semantics() { + let plaintext = b"0123456789abcdefghijklmnopqrstuvwxyz".to_vec(); + let key_bytes = [0x41; 32]; + let bucket = "bucket"; + let object = "range-object"; + let nonce = md5_bytes(format!("{bucket}-{object}").as_bytes()); + let mut base_nonce = [0u8; 12]; + base_nonce.copy_from_slice(&nonce[..12]); + + let mut encrypted = Vec::new(); + rustfs_rio::EncryptReader::new(Cursor::new(plaintext.clone()), key_bytes, base_nonce) + .read_to_end(&mut encrypted) + .await + .expect("encrypt ranged object"); + + let object_info = ObjectInfo { + bucket: bucket.to_string(), + name: object.to_string(), + size: encrypted.len() as i64, + user_defined: HashMap::from([ + ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()), + ( + "x-amz-server-side-encryption-customer-key-md5".to_string(), + BASE64_STANDARD.encode(md5_bytes(key_bytes)), + ), + ( + "x-amz-server-side-encryption-customer-original-size".to_string(), + plaintext.len().to_string(), + ), + ]), + ..Default::default() + }; + let range = HTTPRangeSpec { + is_suffix_length: false, + start: 5, + end: 11, + }; + + let (mut reader, offset, length) = GetObjectReader::new( + Box::new(Cursor::new(encrypted.clone())), + Some(range), + &object_info, + &ObjectOptions::default(), + &ssec_headers_from_key(key_bytes), + ) + .await + .expect("ssec range read should be supported"); + + let mut actual = Vec::new(); + reader.read_to_end(&mut actual).await.expect("read ranged decrypted object"); + + assert_eq!(offset, 0); + assert_eq!(length, encrypted.len() as i64); + assert_eq!(reader.object_info.size, 7); + assert_eq!(actual, b"56789ab"); + } + + #[tokio::test] + async fn test_get_object_reader_decrypts_then_decompresses_before_applying_range() { + let plaintext = b"abcdefghijklmnopqrstuvwxyz".to_vec(); + let key_bytes = [0x51; 32]; + let bucket = "bucket"; + let object = "compressed-object"; + let nonce = md5_bytes(format!("{bucket}-{object}").as_bytes()); + let mut base_nonce = [0u8; 12]; + base_nonce.copy_from_slice(&nonce[..12]); + + let mut compressed = Vec::new(); + rustfs_rio::CompressReader::new(Cursor::new(plaintext.clone()), CompressionAlgorithm::default()) + .read_to_end(&mut compressed) + .await + .expect("compress plaintext"); + + let mut encrypted = Vec::new(); + rustfs_rio::EncryptReader::new(Cursor::new(compressed), key_bytes, base_nonce) + .read_to_end(&mut encrypted) + .await + .expect("encrypt compressed plaintext"); + + let object_info = ObjectInfo { + bucket: bucket.to_string(), + name: object.to_string(), + size: encrypted.len() as i64, + user_defined: HashMap::from([ + ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()), + ( + "x-amz-server-side-encryption-customer-key-md5".to_string(), + BASE64_STANDARD.encode(md5_bytes(key_bytes)), + ), + ( + "x-amz-server-side-encryption-customer-original-size".to_string(), + plaintext.len().to_string(), + ), + ("x-minio-internal-compression".to_string(), CompressionAlgorithm::default().to_string()), + ("x-minio-internal-actual-size".to_string(), plaintext.len().to_string()), + ]), + ..Default::default() + }; + let range = HTTPRangeSpec { + is_suffix_length: false, + start: 5, + end: 11, + }; + + let (mut reader, offset, length) = GetObjectReader::new( + Box::new(Cursor::new(encrypted.clone())), + Some(range), + &object_info, + &ObjectOptions::default(), + &ssec_headers_from_key(key_bytes), + ) + .await + .expect("encrypted+compressed range read should be supported"); + + let mut actual = Vec::new(); + reader + .read_to_end(&mut actual) + .await + .expect("read ranged decompressed plaintext"); + + assert_eq!(offset, 0); + assert_eq!(length, encrypted.len() as i64); + assert_eq!(reader.object_info.size, 7); + assert_eq!(actual, b"fghijkl"); + } } diff --git a/crates/ecstore/src/store_api/types.rs b/crates/ecstore/src/store_api/types.rs index 4c8d789b0..ce0aee78e 100644 --- a/crates/ecstore/src/store_api/types.rs +++ b/crates/ecstore/src/store_api/types.rs @@ -383,6 +383,37 @@ impl ObjectInfo { self.etag.as_ref().is_some_and(|v| v.len() != 32) } + pub fn is_encrypted(&self) -> bool { + use rustfs_utils::http::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER}; + + self.user_defined + .keys() + .any(|key| rustfs_utils::http::is_encryption_metadata_key(key)) + || self.user_defined.contains_key(SSEC_ALGORITHM_HEADER) + || self.user_defined.contains_key(SSEC_KEY_HEADER) + || self.user_defined.contains_key(SSEC_KEY_MD5_HEADER) + } + + pub fn encryption_original_size(&self) -> std::io::Result> { + if let Some(size_str) = self + .user_defined + .get("x-rustfs-encryption-original-size") + .or_else(|| self.user_defined.get("x-amz-server-side-encryption-customer-original-size")) + && !size_str.is_empty() + { + let size = size_str + .parse::() + .map_err(|e| std::io::Error::other(format!("Failed to parse encryption original size: {e}")))?; + return Ok(Some(size)); + } + + Ok(None) + } + + pub fn decrypted_size(&self) -> std::io::Result { + Ok(self.encryption_original_size()?.unwrap_or(self.size)) + } + pub fn get_actual_size(&self) -> std::io::Result { if self.actual_size > 0 { return Ok(self.actual_size); @@ -410,15 +441,7 @@ impl ObjectInfo { // 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_str) = self - .user_defined - .get("x-rustfs-encryption-original-size") - .or_else(|| self.user_defined.get("x-amz-server-side-encryption-customer-original-size")) - && !size_str.is_empty() - { - let size = size_str - .parse::() - .map_err(|e| std::io::Error::other(format!("Failed to parse encryption original size: {e}")))?; + if let Some(size) = self.encryption_original_size()? { return Ok(size); } diff --git a/crates/rio/src/encrypt_reader.rs b/crates/rio/src/encrypt_reader.rs index d83d206a1..c39515cd2 100644 --- a/crates/rio/src/encrypt_reader.rs +++ b/crates/rio/src/encrypt_reader.rs @@ -57,6 +57,10 @@ where finished: false, } } + + pub fn new_multipart(inner: R, key: [u8; 32], base_nonce: [u8; 12], part_number: usize) -> Self { + Self::new(inner, key, multipart_part_nonce(base_nonce, part_number)) + } } impl AsyncRead for EncryptReader @@ -474,6 +478,10 @@ fn derive_block_nonce(base: &[u8; 12], block_index: usize) -> [u8; 12] { derive_nonce_offset(base, 8, block_index) } +pub fn multipart_part_nonce(base_nonce: [u8; 12], part_number: usize) -> [u8; 12] { + derive_part_nonce(&base_nonce, part_number) +} + fn derive_part_nonce(base: &[u8; 12], part_number: usize) -> [u8; 12] { derive_nonce_offset(base, 4, part_number) } diff --git a/crates/rio/src/lib.rs b/crates/rio/src/lib.rs index 9663f133d..e01e72dea 100644 --- a/crates/rio/src/lib.rs +++ b/crates/rio/src/lib.rs @@ -88,7 +88,7 @@ mod compress_reader; pub use compress_reader::{CompressReader, DecompressReader}; mod encrypt_reader; -pub use encrypt_reader::{DecryptReader, EncryptReader}; +pub use encrypt_reader::{DecryptReader, EncryptReader, multipart_part_nonce}; mod hardlimit_reader; pub use hardlimit_reader::HardLimitReader; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 4e7a535b2..62d7574a6 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -28,6 +28,7 @@ use crate::storage::s3_api::multipart::{ ListMultipartUploadsParams, build_list_multipart_uploads_output, build_list_parts_output, parse_list_multipart_uploads_params, parse_list_parts_params, }; +use crate::storage::sse::{build_ssec_read_headers, encryption_material_to_metadata, map_get_object_reader_error}; use crate::storage::*; use bytes::Bytes; use futures::StreamExt; @@ -48,7 +49,9 @@ use rustfs_ecstore::set_disk::is_valid_storage_class; use rustfs_ecstore::store_api::{CompletePart, HTTPRangeSpec, MultipartUploadResult, ObjectIO, ObjectOptions, PutObjReader}; use rustfs_ecstore::store_api::{MultipartOperations, ObjectOperations}; use rustfs_filemeta::{ReplicationStatusType, ReplicationType}; -use rustfs_rio::{CompressReader, HashReader}; +use rustfs_rio::{CompressReader, EncryptReader, HashReader}; +#[cfg(test)] +use rustfs_rio::{DecryptReader, HardLimitReader, boxed_reader, wrap_reader}; use rustfs_s3_common::S3Operation; use rustfs_targets::EventName; use rustfs_utils::CompressionAlgorithm; @@ -68,6 +71,7 @@ use tracing::{instrument, warn}; use urlencoding::encode; use uuid::Uuid; +#[cfg(test)] fn merge_part_encryption_metadata( metadata: &HashMap, part_metadata: &HashMap, @@ -77,6 +81,27 @@ fn merge_part_encryption_metadata( merged } +#[cfg(test)] +fn multipart_plaintext_size(parts: &[rustfs_filemeta::ObjectPartInfo], fallback: i64) -> i64 { + let total: i64 = parts + .iter() + .map(|part| { + if part.actual_size > 0 { + part.actual_size + } else { + part.size as i64 + } + }) + .sum(); + + if total > 0 { total } else { fallback } +} + +#[cfg(test)] +fn multipart_part_numbers(parts: &[rustfs_filemeta::ObjectPartInfo]) -> Vec { + parts.iter().map(|part| part.number).collect() +} + /// Returns InvalidRange error if CopySourceRange end exceeds the source object size. /// Used by execute_upload_part_copy to reject out-of-bounds ranges per S3 spec. fn validate_copy_source_range_not_exceeds(range_spec: &HTTPRangeSpec, object_size: i64) -> S3Result<()> { @@ -532,7 +557,7 @@ impl DefaultMultipartUsecase { let server_side_encryption = Some(material.server_side_encryption.clone()); let ssekms_key_id = material.kms_key_id.clone(); - metadata.extend(material.metadata); + metadata.extend(encryption_material_to_metadata(&material)); (server_side_encryption, ssekms_key_id) } @@ -735,37 +760,64 @@ impl DefaultMultipartUsecase { }; (sse, key_id) }; - let part_key = fi.user_defined.get("x-rustfs-encryption-key").cloned(); - let part_nonce = fi.user_defined.get("x-rustfs-encryption-iv").cloned(); - let encryption_request = EncryptionRequest { + EncryptionRequest { bucket: &bucket, key: &key, - server_side_encryption, - ssekms_key_id, + server_side_encryption: server_side_encryption.clone(), + ssekms_key_id: ssekms_key_id.clone(), sse_customer_algorithm: sse_customer_algorithm.clone(), - sse_customer_key, + sse_customer_key: sse_customer_key.clone(), sse_customer_key_md5: sse_customer_key_md5.clone(), content_size: actual_size, - part_number: Some(part_id), - part_key, - part_nonce, - }; + } + .check_upload_part_customer_key_md5(&fi.user_defined, sse_customer_key_md5.clone())?; + let (requested_sse, requested_kms_key_id) = if has_ssec { + let encryption_request = EncryptionRequest { + bucket: &bucket, + key: &key, + server_side_encryption, + ssekms_key_id, + sse_customer_algorithm: sse_customer_algorithm.clone(), + sse_customer_key, + sse_customer_key_md5: sse_customer_key_md5.clone(), + content_size: actual_size, + }; - encryption_request.check_upload_part_customer_key_md5(&fi.user_defined, sse_customer_key_md5.clone())?; - - let (requested_sse, requested_kms_key_id) = match sse_encryption(encryption_request).await? { - Some(material) => { - let requested_sse = Some(material.server_side_encryption.clone()); - let requested_kms_key_id = material.kms_key_id.clone(); - - let encrypted_reader = material.wrap_reader(reader); - reader = - HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) - .map_err(ApiError::from)?; - - (requested_sse, requested_kms_key_id) + match sse_encryption(encryption_request).await? { + Some(material) => { + let requested_sse = Some(material.server_side_encryption.clone()); + let requested_kms_key_id = material.kms_key_id.clone(); + let encrypted_reader = EncryptReader::new_multipart(reader, material.key_bytes, material.base_nonce, part_id); + reader = HashReader::from_reader( + encrypted_reader, + HashReader::SIZE_PRESERVE_LAYER, + actual_size, + None, + None, + false, + ) + .map_err(ApiError::from)?; + (requested_sse, requested_kms_key_id) + } + None => (None, None), } - None => (None, None), + } else if let Some(server_side_encryption) = server_side_encryption { + let managed_material = sse_decryption(DecryptionRequest { + bucket: &bucket, + key: &key, + metadata: &fi.user_defined, + sse_customer_key: None, + sse_customer_key_md5: None, + }) + .await? + .ok_or_else(|| ApiError::from(StorageError::other("Missing managed SSE session material")))?; + let encrypted_reader = + EncryptReader::new_multipart(reader, managed_material.key_bytes, managed_material.base_nonce, part_id); + reader = HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) + .map_err(ApiError::from)?; + (Some(server_side_encryption), ssekms_key_id) + } else { + (None, None) }; let mut reader = PutObjReader::new(reader); @@ -905,6 +957,7 @@ impl DefaultMultipartUsecase { sse_customer_algorithm, sse_customer_key, sse_customer_key_md5, + copy_source_sse_customer_algorithm, copy_source_sse_customer_key, copy_source_sse_customer_key_md5, .. @@ -940,7 +993,11 @@ impl DefaultMultipartUsecase { let mut src_opts = copy_src_opts(&src_bucket, &src_key, &req.headers).map_err(ApiError::from)?; src_opts.version_id = src_version_id.clone(); - let h = HeaderMap::new(); + let h = build_ssec_read_headers( + copy_source_sse_customer_algorithm.as_ref(), + copy_source_sse_customer_key.as_ref(), + copy_source_sse_customer_key_md5.as_ref(), + ); let get_opts = ObjectOptions { version_id: src_opts.version_id.clone(), versioned: src_opts.versioned, @@ -951,9 +1008,9 @@ impl DefaultMultipartUsecase { let src_reader = store .get_object_reader(&src_bucket, &src_key, rs.clone(), h, &get_opts) .await - .map_err(ApiError::from)?; + .map_err(map_get_object_reader_error)?; - let mut src_info = src_reader.object_info; + let src_info = src_reader.object_info; if let Some(if_match) = copy_source_if_match { if let Some(ref etag) = src_info.etag { @@ -992,7 +1049,11 @@ impl DefaultMultipartUsecase { (0, src_info.size) }; - let h = HeaderMap::new(); + let h = build_ssec_read_headers( + copy_source_sse_customer_algorithm.as_ref(), + copy_source_sse_customer_key.as_ref(), + copy_source_sse_customer_key_md5.as_ref(), + ); let get_opts = ObjectOptions { version_id: src_opts.version_id.clone(), versioned: src_opts.versioned, @@ -1003,89 +1064,28 @@ impl DefaultMultipartUsecase { let src_reader = store .get_object_reader(&src_bucket, &src_key, rs.clone(), h, &get_opts) .await - .map_err(ApiError::from)?; + .map_err(map_get_object_reader_error)?; let src_stream = src_reader.stream; let is_compressible = rustfs_utils::http::contains_key_str(&mp_info.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION); - let src_decryption_request = DecryptionRequest { - bucket: &src_bucket, - key: &src_key, - metadata: &src_info.user_defined, - sse_customer_key: copy_source_sse_customer_key.as_ref(), - sse_customer_key_md5: copy_source_sse_customer_key_md5.as_ref(), - part_number: None, - parts: &src_info.parts, - etag: src_info.etag.as_deref(), - }; - let actual_size = length; let mut size = length; - let mut reader = match sse_decryption(src_decryption_request).await? { - Some(material) => { - if let Some(original) = material.original_size { - src_info.actual_size = original; - } - - if material.is_multipart { - let (decrypted_stream, plaintext_size) = - material.wrap_reader(src_stream, size).await.map_err(ApiError::from)?; - size = plaintext_size; - - if is_compressible { - let hrd = HashReader::from_reader(decrypted_stream, size, actual_size, None, None, false) - .map_err(ApiError::from)?; - size = HashReader::SIZE_PRESERVE_LAYER; - HashReader::from_reader( - CompressReader::new(hrd, CompressionAlgorithm::default()), - size, - actual_size, - None, - None, - false, - ) - .map_err(ApiError::from)? - } else { - HashReader::from_reader(decrypted_stream, size, actual_size, None, None, false).map_err(ApiError::from)? - } - } else if is_compressible { - let hrd = - HashReader::from_stream(material.wrap_single_reader(src_stream), size, actual_size, None, None, false) - .map_err(ApiError::from)?; - size = HashReader::SIZE_PRESERVE_LAYER; - HashReader::from_reader( - CompressReader::new(hrd, CompressionAlgorithm::default()), - size, - actual_size, - None, - None, - false, - ) - .map_err(ApiError::from)? - } else { - HashReader::from_stream(material.wrap_single_reader(src_stream), size, actual_size, None, None, false) - .map_err(ApiError::from)? - } - } - None => { - if is_compressible { - let hrd = - HashReader::from_stream(src_stream, size, actual_size, None, None, false).map_err(ApiError::from)?; - size = HashReader::SIZE_PRESERVE_LAYER; - HashReader::from_reader( - CompressReader::new(hrd, CompressionAlgorithm::default()), - size, - actual_size, - None, - None, - false, - ) - .map_err(ApiError::from)? - } else { - HashReader::from_stream(src_stream, size, actual_size, None, None, false).map_err(ApiError::from)? - } - } + let mut reader = if is_compressible { + let hrd = HashReader::from_stream(src_stream, size, actual_size, None, None, false).map_err(ApiError::from)?; + size = HashReader::SIZE_PRESERVE_LAYER; + HashReader::from_reader( + CompressReader::new(hrd, CompressionAlgorithm::default()), + size, + actual_size, + None, + None, + false, + ) + .map_err(ApiError::from)? + } else { + HashReader::from_stream(src_stream, size, actual_size, None, None, false).map_err(ApiError::from)? }; let server_side_encryption = mp_info @@ -1096,6 +1096,7 @@ impl DefaultMultipartUsecase { .map_err(|e| ApiError::from(StorageError::other(format!("Invalid server-side encryption: {e}")))) }) .transpose()?; + let has_ssec = sse_customer_algorithm.is_some(); let ssekms_key_id = match server_side_encryption.as_ref() { Some(sse) if sse.as_str() == ServerSideEncryption::AWS_KMS => mp_info .user_defined @@ -1103,38 +1104,65 @@ impl DefaultMultipartUsecase { .map(|s| s.to_string()), _ => None, }; - let part_key = mp_info.user_defined.get("x-rustfs-encryption-key").cloned(); - let part_nonce = mp_info.user_defined.get("x-rustfs-encryption-iv").cloned(); - let encryption_request = EncryptionRequest { + EncryptionRequest { bucket: &bucket, key: &key, - server_side_encryption, - ssekms_key_id, + server_side_encryption: server_side_encryption.clone(), + ssekms_key_id: ssekms_key_id.clone(), sse_customer_algorithm: sse_customer_algorithm.clone(), - sse_customer_key, + sse_customer_key: sse_customer_key.clone(), sse_customer_key_md5: sse_customer_key_md5.clone(), content_size: actual_size, - part_number: Some(part_id), - part_key, - part_nonce, - }; + } + .check_upload_part_customer_key_md5(&mp_info.user_defined, sse_customer_key_md5.clone())?; - encryption_request.check_upload_part_customer_key_md5(&mp_info.user_defined, sse_customer_key_md5.clone())?; + let (requested_sse, requested_kms_key_id, dst_user_defined) = if has_ssec { + let encryption_request = EncryptionRequest { + bucket: &bucket, + key: &key, + server_side_encryption, + ssekms_key_id, + sse_customer_algorithm: sse_customer_algorithm.clone(), + sse_customer_key, + sse_customer_key_md5: sse_customer_key_md5.clone(), + content_size: actual_size, + }; - let (requested_sse, requested_kms_key_id, dst_user_defined) = match sse_encryption(encryption_request).await? { - Some(material) => { - let requested_sse = Some(material.server_side_encryption.clone()); - let requested_kms_key_id = material.kms_key_id.clone(); - - let encrypted_reader = material.wrap_reader(reader); - reader = - HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) - .map_err(ApiError::from)?; - let dst_user_defined = merge_part_encryption_metadata(&mp_info.user_defined, &material.metadata); - - (requested_sse, requested_kms_key_id, dst_user_defined) + match sse_encryption(encryption_request).await? { + Some(material) => { + let requested_sse = Some(material.server_side_encryption.clone()); + let requested_kms_key_id = material.kms_key_id.clone(); + let encrypted_reader = EncryptReader::new_multipart(reader, material.key_bytes, material.base_nonce, part_id); + reader = HashReader::from_reader( + encrypted_reader, + HashReader::SIZE_PRESERVE_LAYER, + actual_size, + None, + None, + false, + ) + .map_err(ApiError::from)?; + (requested_sse, requested_kms_key_id, mp_info.user_defined.clone()) + } + None => (None, None, mp_info.user_defined.clone()), } - None => (None, None, mp_info.user_defined.clone()), + } else if let Some(server_side_encryption) = server_side_encryption { + let managed_material = sse_decryption(DecryptionRequest { + bucket: &bucket, + key: &key, + metadata: &mp_info.user_defined, + sse_customer_key: None, + sse_customer_key_md5: None, + }) + .await? + .ok_or_else(|| ApiError::from(StorageError::other("Missing managed SSE session material")))?; + let encrypted_reader = + EncryptReader::new_multipart(reader, managed_material.key_bytes, managed_material.base_nonce, part_id); + reader = HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) + .map_err(ApiError::from)?; + (Some(server_side_encryption), ssekms_key_id, mp_info.user_defined.clone()) + } else { + (None, None, mp_info.user_defined.clone()) }; if let Some(checksum_algorithm) = mp_info @@ -1294,77 +1322,61 @@ mod tests { .await .expect("prepare multipart encryption") .expect("managed multipart session material"); - let session_metadata = session_material.metadata.clone(); + let session_metadata = encryption_material_to_metadata(&session_material); let session_nonce = session_metadata .get("x-rustfs-encryption-iv") .cloned() .expect("session nonce metadata"); - let part_key = session_metadata.get("x-rustfs-encryption-key").cloned(); - let part_nonce = Some(session_nonce.clone()); let part_one_plaintext = vec![0x31; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 23]; let part_two_plaintext = vec![0x32; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE * 2 + 7]; - let part_one_material = sse_encryption(EncryptionRequest { + let part_one_material = sse_decryption(DecryptionRequest { bucket: "bucket", key: "object", - server_side_encryption: Some(session_material.server_side_encryption.clone()), - ssekms_key_id: session_material.kms_key_id.clone(), - sse_customer_algorithm: None, + metadata: &session_metadata, sse_customer_key: None, sse_customer_key_md5: None, - content_size: part_one_plaintext.len() as i64, - part_number: Some(1), - part_key: part_key.clone(), - part_nonce: part_nonce.clone(), }) .await - .expect("encrypt part one") + .expect("decrypt session one") .expect("part one material"); - let part_one_metadata = merge_part_encryption_metadata(&session_metadata, &part_one_material.metadata); let mut encrypted_one = Vec::new(); - part_one_material - .wrap_reader(Cursor::new(part_one_plaintext.clone())) - .read_to_end(&mut encrypted_one) - .await - .expect("read encrypted part one"); + EncryptReader::new_multipart( + Cursor::new(part_one_plaintext.clone()), + part_one_material.key_bytes, + part_one_material.base_nonce, + 1, + ) + .read_to_end(&mut encrypted_one) + .await + .expect("read encrypted part one"); - let part_two_material = sse_encryption(EncryptionRequest { + let part_two_material = sse_decryption(DecryptionRequest { bucket: "bucket", key: "object", - server_side_encryption: Some(session_material.server_side_encryption.clone()), - ssekms_key_id: session_material.kms_key_id.clone(), - sse_customer_algorithm: None, + metadata: &session_metadata, sse_customer_key: None, sse_customer_key_md5: None, - content_size: part_two_plaintext.len() as i64, - part_number: Some(2), - part_key, - part_nonce, }) .await - .expect("encrypt part two") + .expect("decrypt session two") .expect("part two material"); - let part_two_metadata = merge_part_encryption_metadata(&session_metadata, &part_two_material.metadata); let mut encrypted_two = Vec::new(); - part_two_material - .wrap_reader(Cursor::new(part_two_plaintext.clone())) - .read_to_end(&mut encrypted_two) - .await - .expect("read encrypted part two"); + EncryptReader::new_multipart( + Cursor::new(part_two_plaintext.clone()), + part_two_material.key_bytes, + part_two_material.base_nonce, + 2, + ) + .read_to_end(&mut encrypted_two) + .await + .expect("read encrypted part two"); assert_eq!( session_metadata.get("x-rustfs-encryption-iv").map(String::as_str), Some(session_nonce.as_str()) ); - assert_ne!( - part_one_metadata.get("x-rustfs-encryption-iv"), - session_metadata.get("x-rustfs-encryption-iv") - ); - assert_ne!( - part_two_metadata.get("x-rustfs-encryption-iv"), - session_metadata.get("x-rustfs-encryption-iv") - ); let parts = vec![ ObjectPartInfo { @@ -1391,18 +1403,21 @@ mod tests { metadata: &session_metadata, sse_customer_key: None, sse_customer_key_md5: None, - part_number: None, - parts: &parts, - etag: Some("multipart-etag"), }) .await .expect("decrypt multipart") .expect("managed decryption material"); - let (mut decrypted_reader, plaintext_size) = decryption_material - .wrap_reader(Cursor::new(encrypted_stream), -1) - .await - .expect("wrap multipart reader"); + let plaintext_size = multipart_plaintext_size(&parts, -1); + let mut decrypted_reader = HardLimitReader::new( + boxed_reader(DecryptReader::new_multipart( + wrap_reader(Cursor::new(encrypted_stream)), + decryption_material.key_bytes, + decryption_material.base_nonce, + multipart_part_numbers(&parts), + )), + plaintext_size, + ); let mut decrypted = Vec::new(); decrypted_reader diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 4422f9a29..d26a5cef7 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -31,6 +31,7 @@ use crate::storage::options::{ }; use crate::storage::request_context::spawn_traced; use crate::storage::s3_api::multipart::parse_list_parts_params; +use crate::storage::sse::{SSEType, build_ssec_read_headers, encryption_material_to_metadata, map_get_object_reader_error}; use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig}; use crate::storage::*; use bytes::Bytes; @@ -84,7 +85,7 @@ use rustfs_filemeta::{ use rustfs_io_metrics; use rustfs_notify::EventArgsBuilder; use rustfs_policy::policy::action::{Action, S3Action}; -use rustfs_rio::{CompressReader, DynReader, HashReader, wrap_reader}; +use rustfs_rio::{CompressReader, DynReader, EncryptReader, HashReader, wrap_reader}; use rustfs_s3_common::S3Operation; use rustfs_s3select_api::{ object_store::bytes_stream, @@ -118,6 +119,7 @@ use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use std::collections::HashMap; use std::ops::Add; use std::path::Path; + use std::str::FromStr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; @@ -1279,7 +1281,7 @@ impl DefaultObjectUsecase { opts: &ObjectOptions, part_number: Option, ) -> S3Result> { - let h = HeaderMap::new(); + let h = req.headers.clone(); let io_planning = Self::acquire_get_object_io_planning(manager, wrapper, timeout_config, bucket, key).await?; let store = get_validated_store(bucket).await?; @@ -1306,7 +1308,7 @@ impl DefaultObjectUsecase { let reader = store .get_object_reader(bucket, key, rs.clone(), h, opts) .await - .map_err(ApiError::from)?; + .map_err(map_get_object_reader_error)?; let info = reader.object_info; @@ -1372,13 +1374,9 @@ impl DefaultObjectUsecase { metadata: &info.user_defined, sse_customer_key: req.input.sse_customer_key.as_ref(), sse_customer_key_md5: req.input.sse_customer_key_md5.as_ref(), - part_number: None, - parts: &info.parts, - etag: info.etag.as_deref(), }; - let mut response_content_length = content_length; - let encrypted_stream = reader.stream; + let response_content_length = content_length; let ( server_side_encryption, @@ -1390,27 +1388,18 @@ impl DefaultObjectUsecase { ) = match sse_decryption(decryption_request).await? { Some(material) => { let server_side_encryption = Some(material.server_side_encryption.clone()); - let sse_customer_algorithm = Some(material.algorithm.clone()); + let sse_customer_algorithm = matches!(material.sse_type, SSEType::SseC).then_some(material.algorithm.clone()); let sse_customer_key_md5 = material.customer_key_md5.clone(); - let ssekms_key_id = material.kms_key_id.clone(); - - let (decrypted_stream, plaintext_size) = material - .wrap_reader(encrypted_stream, content_length) - .await - .map_err(ApiError::from)?; - - response_content_length = plaintext_size; - ( server_side_encryption, sse_customer_algorithm, sse_customer_key_md5, - ssekms_key_id, + material.kms_key_id, true, - decrypted_stream, + wrap_reader(reader.stream), ) } - None => (None, None, None, None, false, wrap_reader(encrypted_stream)), + None => (None, None, None, None, false, wrap_reader(reader.stream)), }; Ok(GetObjectReadSetup { @@ -1896,9 +1885,6 @@ impl DefaultObjectUsecase { sse_customer_key, sse_customer_key_md5: sse_customer_key_md5.clone(), content_size: actual_size, - part_number: None, - part_key: None, - part_nonce: None, }; let encryption_material = match sse_encryption(encryption_request).await { @@ -1914,11 +1900,11 @@ impl DefaultObjectUsecase { effective_sse = Some(material.server_side_encryption.clone()); effective_kms_key_id = material.kms_key_id.clone(); - let encrypted_reader = material.wrap_reader(reader); + let encrypted_reader = EncryptReader::new(reader, material.key_bytes, material.base_nonce); reader = HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) .map_err(ApiError::from)?; - let encryption_metadata = material.metadata; + let encryption_metadata = encryption_material_to_metadata(&material); metadata.extend(encryption_metadata.clone()); opts.user_defined.extend(encryption_metadata); } @@ -2577,6 +2563,7 @@ impl DefaultObjectUsecase { sse_customer_algorithm, sse_customer_key, sse_customer_key_md5, + copy_source_sse_customer_algorithm, copy_source_sse_customer_key, copy_source_sse_customer_key_md5, metadata_directive, @@ -2679,12 +2666,16 @@ impl DefaultObjectUsecase { }) }); - let h = HeaderMap::new(); + let h = build_ssec_read_headers( + copy_source_sse_customer_algorithm.as_ref(), + copy_source_sse_customer_key.as_ref(), + copy_source_sse_customer_key_md5.as_ref(), + ); let gr = store .get_object_reader(&src_bucket, &src_key, None, h, &src_get_opts) .await - .map_err(ApiError::from)?; + .map_err(map_get_object_reader_error)?; let mut src_info = gr.object_info.clone(); @@ -2716,25 +2707,6 @@ impl DefaultObjectUsecase { src_info.metadata_only = true; } - let decryption_request = DecryptionRequest { - bucket: &src_bucket, - key: &src_key, - metadata: &src_info.user_defined, - sse_customer_key: copy_source_sse_customer_key.as_ref(), - sse_customer_key_md5: copy_source_sse_customer_key_md5.as_ref(), - part_number: None, - parts: &src_info.parts, - etag: src_info.etag.as_deref(), - }; - - let decryption_material = sse_decryption(decryption_request).await?; - - if let Some(material) = decryption_material.as_ref() - && let Some(original) = material.original_size - { - src_info.actual_size = original; - } - strip_managed_encryption_metadata(&mut src_info.user_defined); let actual_size = src_info.get_actual_size().map_err(ApiError::from)?; @@ -2782,67 +2754,20 @@ impl DefaultObjectUsecase { } apply_bucket_default_lock_retention(&bucket, &mut src_info.user_defined, has_explicit_object_lock_retention).await?; - let mut reader = match decryption_material { - Some(material) => { - if material.is_multipart { - let (decrypted_stream, plaintext_size) = - material.wrap_reader(gr.stream, length).await.map_err(ApiError::from)?; - length = plaintext_size; - - if should_compress { - let hrd = HashReader::from_reader(decrypted_stream, length, actual_size, None, None, false) - .map_err(ApiError::from)?; - length = HashReader::SIZE_PRESERVE_LAYER; - HashReader::from_reader( - CompressReader::new(hrd, CompressionAlgorithm::default()), - length, - actual_size, - None, - None, - false, - ) - .map_err(ApiError::from)? - } else { - HashReader::from_reader(decrypted_stream, length, actual_size, None, None, false) - .map_err(ApiError::from)? - } - } else if should_compress { - let hrd = - HashReader::from_stream(material.wrap_single_reader(gr.stream), length, actual_size, None, None, false) - .map_err(ApiError::from)?; - length = HashReader::SIZE_PRESERVE_LAYER; - HashReader::from_reader( - CompressReader::new(hrd, CompressionAlgorithm::default()), - length, - actual_size, - None, - None, - false, - ) - .map_err(ApiError::from)? - } else { - HashReader::from_stream(material.wrap_single_reader(gr.stream), length, actual_size, None, None, false) - .map_err(ApiError::from)? - } - } - None => { - if should_compress { - let hrd = - HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)?; - length = HashReader::SIZE_PRESERVE_LAYER; - HashReader::from_reader( - CompressReader::new(hrd, CompressionAlgorithm::default()), - length, - actual_size, - None, - None, - false, - ) - .map_err(ApiError::from)? - } else { - HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)? - } - } + let mut reader = if should_compress { + let hrd = HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)?; + length = HashReader::SIZE_PRESERVE_LAYER; + HashReader::from_reader( + CompressReader::new(hrd, CompressionAlgorithm::default()), + length, + actual_size, + None, + None, + false, + ) + .map_err(ApiError::from)? + } else { + HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)? }; let encryption_request = EncryptionRequest { @@ -2854,20 +2779,17 @@ impl DefaultObjectUsecase { sse_customer_key, sse_customer_key_md5: sse_customer_key_md5.clone(), content_size: actual_size, - part_number: None, - part_key: None, - part_nonce: None, }; if let Some(material) = sse_encryption(encryption_request).await? { effective_sse = Some(material.server_side_encryption.clone()); effective_kms_key_id = material.kms_key_id.clone(); - let encrypted_reader = material.wrap_reader(reader); + let encrypted_reader = EncryptReader::new(reader, material.key_bytes, material.base_nonce); reader = HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) .map_err(ApiError::from)?; - src_info.user_defined.extend(material.metadata); + src_info.user_defined.extend(encryption_material_to_metadata(&material)); } src_info.put_object_reader = Some(PutObjReader::new(reader)); @@ -4380,20 +4302,17 @@ impl DefaultObjectUsecase { sse_customer_key: sse_customer_key.clone(), sse_customer_key_md5: sse_customer_key_md5.clone(), content_size: actual_size, - part_number: None, - part_key: None, - part_nonce: None, }) .await? { effective_sse = Some(material.server_side_encryption.clone()); effective_kms_key_id = material.kms_key_id.clone(); - let encrypted_reader = material.wrap_reader(hrd); + let encrypted_reader = EncryptReader::new(hrd, material.key_bytes, material.base_nonce); hrd = HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) .map_err(ApiError::from)?; - let encryption_metadata = material.metadata; + let encryption_metadata = encryption_material_to_metadata(&material); metadata.extend(encryption_metadata.clone()); opts.user_defined.extend(encryption_metadata); } @@ -4509,7 +4428,6 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; use std::task::{Context, Poll}; use tokio::io::{AsyncRead, ReadBuf}; - fn build_request(input: T, method: Method) -> S3Request { S3Request { input, diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index 6a76367ff..2263ed81a 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -23,7 +23,7 @@ pub mod options; pub mod request_context; pub mod rpc; pub(crate) mod s3_api; -mod sse; +pub(crate) mod sse; pub mod timeout_wrapper; pub mod tonic_service; diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index 533e78c25..6584a59bf 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -49,12 +49,10 @@ //! sse_customer_key: sse_customer_key.as_deref(), //! sse_customer_key_md5: sse_customer_key_md5.as_deref(), //! content_size: actual_size, -//! part_number: None, //! }; //! //! if let Some(material) = sse_encryption(request).await? { -//! reader = material.wrap_reader(reader); -//! metadata.extend(material.metadata); +//! metadata.extend(encryption_material_to_metadata(&material)); //! } //! //! // Unified decryption API @@ -64,13 +62,10 @@ //! metadata: &metadata, //! sse_customer_key: sse_customer_key.as_deref(), //! sse_customer_key_md5: sse_customer_key_md5.as_deref(), -//! part_number: None, //! }; //! //! if let Some(material) = sse_decryption(request).await? { -//! let (decrypted_reader, plaintext_size) = material.wrap_reader(reader, actual_size).await?; -//! reader = decrypted_reader; -//! content_size = plaintext_size; +//! content_size = material.original_size.unwrap_or(actual_size); //! } //! ``` @@ -80,16 +75,10 @@ use aes_gcm::{ }; use async_trait::async_trait; use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD}; -use http::HeaderMap; +use http::{HeaderMap, HeaderValue}; use rand::Rng; use rustfs_ecstore::error::StorageError; -use rustfs_filemeta::ObjectPartInfo; -use rustfs_kms::{ - DataKey, - service_manager::get_global_encryption_service, - types::{EncryptionMetadata, ObjectEncryptionContext}, -}; -use rustfs_rio::{DecryptReader, DynReader, EncryptReader, HardLimitReader, ReadStream, boxed_reader, wrap_reader}; +use rustfs_kms::{DataKey, service_manager::get_global_encryption_service, types::ObjectEncryptionContext}; use rustfs_utils::get_env_opt_str; use s3s::S3ErrorCode; use s3s::dto::ServerSideEncryption; @@ -98,10 +87,15 @@ use std::sync::{Arc, OnceLock}; use tracing::{debug, error}; const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id"; +const SSEC_ORIGINAL_SIZE_HEADER: &str = "x-amz-server-side-encryption-customer-original-size"; use crate::error::ApiError; use rustfs_ecstore::bucket::metadata_sys; use rustfs_ecstore::error::Error; +use rustfs_utils::http::headers::{ + AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, + AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, +}; use s3s::dto::{SSECustomerAlgorithm, SSECustomerKey, SSECustomerKeyMD5, SSEKMSKeyId}; // ============================================================================ @@ -347,11 +341,6 @@ pub struct EncryptionRequest<'a> { pub sse_customer_key_md5: Option, /// Content size (for metadata) pub content_size: i64, - - /// Part number (for multipart upload, None for single-part) - pub part_number: Option, - pub part_key: Option, - pub part_nonce: Option, } impl EncryptionRequest<'_> { @@ -533,6 +522,52 @@ pub(crate) fn validate_sse_headers_for_read(metadata: &HashMap, Ok(()) } +pub(crate) fn map_get_object_reader_error(err: StorageError) -> ApiError { + if let Some(message) = map_ssec_get_object_reader_error_message(&err) { + return ApiError { + code: S3ErrorCode::InvalidRequest, + message, + source: Some(Box::new(err)), + }; + } + + ApiError::from(err) +} + +fn map_ssec_get_object_reader_error_message(err: &StorageError) -> Option { + let StorageError::Io(io_err) = err else { + return None; + }; + + let detail = io_err.to_string(); + match detail.as_str() { + "missing SSE-C algorithm header" + | "invalid SSE-C algorithm header" + | "missing SSE-C key header" + | "invalid SSE-C key header" + | "missing SSE-C key md5 header" + | "invalid SSE-C key md5 header" => Some( + "The object was stored using a form of Server Side Encryption. The correct parameters must be provided to retrieve the object." + .to_string(), + ), + "failed to decode SSE-C key" => Some("Invalid SSE-C key: not valid Base64.".to_string()), + "SSE-C key must be 32 bytes" => Some("SSE-C key must be exactly 32 bytes.".to_string()), + "SSE-C key MD5 mismatch" => { + Some("The calculated MD5 hash of the key did not match the hash that was provided.".to_string()) + } + "missing stored SSE-C key md5" => Some("Object has no stored SSE-C key metadata.".to_string()), + "SSE-C key does not match object metadata" => Some( + "The provided encryption parameters did not match the ones used originally to encrypt the object.".to_string(), + ), + _ => detail.strip_prefix("unsupported SSE-C algorithm ").map(|algorithm| { + format!( + "Unsupported SSE-C algorithm: {}. Only {} is supported.", + algorithm, DEFAULT_SSE_ALGORITHM + ) + }), + } +} + /// Request parameters for unified decryption #[derive(Debug)] pub struct DecryptionRequest<'a> { @@ -546,15 +581,9 @@ pub struct DecryptionRequest<'a> { pub sse_customer_key: Option<&'a SSECustomerKey>, /// SSE-C key MD5 (Base64-encoded) - required if object was encrypted with SSE-C pub sse_customer_key_md5: Option<&'a SSECustomerKeyMD5>, - /// Part number (for multipart upload, None for single-part) - pub part_number: Option, // Unused Fields - /// Parts information for multipart objects - pub parts: &'a [ObjectPartInfo], - /// Object-level ETag, used to distinguish multipart objects from single-part objects. - pub etag: Option<&'a str>, } -/// Unified encryption material returned by `apply_encryption()` +/// Encryption material returned by `sse_encryption()` / `sse_prepare_encryption()`. #[derive(Debug)] pub struct EncryptionMaterial { #[allow(unused)] @@ -567,13 +596,17 @@ pub struct EncryptionMaterial { /// Encryption key bytes pub key_bytes: [u8; 32], - /// Nonce/IV for encryption - pub nonce: [u8; 12], - /// Metadata to store with the object - pub metadata: HashMap, + /// Base nonce/IV used by rio to derive block/part nonces. + pub base_nonce: [u8; 12], + /// Encrypted DEK for managed SSE. Absent for SSE-C. + pub encrypted_data_key: Option>, + /// SSE-C key MD5 if customer-managed encryption is in use. + pub customer_key_md5: Option, + /// Original plaintext size when it should be persisted alongside metadata. + pub original_size: Option, } -/// Unified decryption material returned by `apply_decryption()` +/// Decryption material returned by `sse_decryption()`. #[derive(Debug)] pub struct DecryptionMaterial { #[allow(unused)] @@ -585,23 +618,8 @@ pub struct DecryptionMaterial { /// Decryption key bytes pub key_bytes: [u8; 32], - /// Nonce/IV for decryption - pub nonce: [u8; 12], - /// Original unencrypted size (if available) - pub original_size: Option, - - /// Whether this is a multipart object - pub is_multipart: bool, - /// Part information for multipart objects - pub parts: Vec, -} - -fn is_multipart_object(etag: Option<&str>, parts: &[ObjectPartInfo]) -> bool { - if parts.len() > 1 { - return true; - } - - etag.map(|etag| etag.trim_matches('"').len() != 32).unwrap_or(false) + /// Base nonce/IV used by rio to derive block/part nonces. + pub base_nonce: [u8; 12], } /// Type of encryption used @@ -615,65 +633,84 @@ pub enum SSEType { SseC, } -impl EncryptionMaterial { - /// Wrap a reader with encryption - pub fn wrap_reader(&self, reader: R) -> Box> - where - R: rustfs_rio::ReadStream + 'static, +pub(crate) fn build_ssec_read_headers( + algorithm: Option<&SSECustomerAlgorithm>, + key: Option<&SSECustomerKey>, + key_md5: Option<&SSECustomerKeyMD5>, +) -> HeaderMap { + let mut headers = HeaderMap::new(); + + if let Some(algorithm) = algorithm + && let Ok(value) = HeaderValue::from_str(algorithm.as_str()) { - Box::new(EncryptReader::new(reader, self.key_bytes, self.nonce)) + headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, value); } + + if let Some(key) = key + && let Ok(value) = HeaderValue::from_str(key.as_str()) + { + headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, value); + } + + if let Some(key_md5) = key_md5 + && let Ok(value) = HeaderValue::from_str(key_md5.as_str()) + { + headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, value); + } + + headers } -impl DecryptionMaterial { - /// Wrap a reader with decryption - /// For multipart objects, use `wrap_multipart_stream` instead - pub fn wrap_single_reader(&self, reader: R) -> Box> - where - R: rustfs_rio::ReadStream + 'static, - { - Box::new(DecryptReader::new(reader, self.key_bytes, self.nonce)) +pub fn encryption_material_to_metadata(material: &EncryptionMaterial) -> HashMap { + let mut metadata = HashMap::new(); + + match material.sse_type { + SSEType::SseC => { + metadata.insert( + "x-amz-server-side-encryption".to_string(), + material.server_side_encryption.as_str().to_string(), + ); + metadata.insert( + "x-amz-server-side-encryption-customer-algorithm".to_string(), + material.algorithm.as_str().to_string(), + ); + if let Some(customer_key_md5) = &material.customer_key_md5 { + metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), customer_key_md5.to_string()); + } + if let Some(original_size) = material.original_size { + metadata.insert(SSEC_ORIGINAL_SIZE_HEADER.to_string(), original_size.to_string()); + } + } + SSEType::SseS3 | SSEType::SseKms => { + let encrypted_data_key = material + .encrypted_data_key + .as_deref() + .expect("managed SSE materials must carry an encrypted data key"); + metadata.insert("x-rustfs-encryption-key".to_string(), BASE64_STANDARD.encode(encrypted_data_key)); + metadata.insert("x-rustfs-encryption-iv".to_string(), BASE64_STANDARD.encode(material.base_nonce)); + metadata.insert("x-rustfs-encryption-algorithm".to_string(), material.algorithm.as_str().to_string()); + metadata.insert( + "x-amz-server-side-encryption".to_string(), + material.server_side_encryption.as_str().to_string(), + ); + + let internal_key_id = material + .kms_key_id + .clone() + .unwrap_or_else(|| SSEKMSKeyId::from("default".to_string())); + metadata.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), internal_key_id.clone()); + + if matches!(material.sse_type, SSEType::SseKms) { + metadata.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), internal_key_id); + } + + if let Some(original_size) = material.original_size { + metadata.insert("x-rustfs-encryption-original-size".to_string(), original_size.to_string()); + } + } } - /// Wrap a stream with multipart decryption - /// Returns the decrypted reader and the total plaintext size - pub async fn wrap_multipart_stream(&self, encrypted_stream: R) -> Result<(DynReader, i64), StorageError> - where - R: ReadStream + 'static, - { - decrypt_multipart_managed_stream(encrypted_stream, &self.parts, self.key_bytes, self.nonce).await - } - - /// Unified method to wrap stream with decryption and hard limit - /// Handles both single-part and multipart objects, applies decryption and size limiting - /// Accepts a readable stream (from object storage) and returns (decrypted_reader, plaintext_size) - pub async fn wrap_reader(self, stream: R, actual_size: i64) -> Result<(DynReader, i64), StorageError> - where - R: ReadStream + 'static, - { - let (mut final_stream, response_content_length): (DynReader, i64) = if self.is_multipart { - // Multipart decryption - let (decrypted_reader, plain_size) = self.wrap_multipart_stream(stream).await?; - (decrypted_reader, plain_size) - } else { - // Single-part decryption keeps Reader capabilities via the generic wrapper helper. - let decrypt_reader = self.wrap_single_reader(wrap_reader(stream)); - let plain_size = self.original_size.unwrap_or(actual_size); - (decrypt_reader, plain_size) - }; - - // Add hard limit reader to prevent over-reading - // final_stream is already a DynReader, no need to wrap with WarpReader - let limit_reader = HardLimitReader::new(final_stream, response_content_length); - final_stream = Box::new(limit_reader); - - debug!( - "{:?} decryption applied: plaintext_size={}, encrypted_size={}", - self.sse_type, response_content_length, actual_size - ); - - Ok((final_stream, response_content_length)) - } + metadata } // ============================================================================ @@ -728,17 +765,9 @@ pub async fn sse_encryption(request: EncryptionRequest<'_>) -> Result) -> Result) -> Re // apply encryption material let material = match sse_type { - Some(SseTypeV2::SseS3(sse)) => { - apply_managed_encryption_material(request.bucket, request.key, sse, None, 0, None, None, None).await? - } + Some(SseTypeV2::SseS3(sse)) => apply_managed_encryption_material(request.bucket, request.key, sse, None, 0).await?, Some(SseTypeV2::SseKms(sse, kms_key_id)) => { - apply_managed_encryption_material(request.bucket, request.key, sse, kms_key_id, 0, None, None, None).await? + apply_managed_encryption_material(request.bucket, request.key, sse, kms_key_id, 0).await? } Some(SseTypeV2::SseC(algorithm, _, key_md5)) => apply_ssec_prepare_encryption_material(algorithm, key_md5).await?, None => return Ok(None), @@ -841,18 +865,13 @@ pub async fn sse_prepare_encryption(request: PrepareEncryptionRequest<'_>) -> Re /// metadata: &metadata, /// sse_customer_key: sse_customer_key.as_deref(), /// sse_customer_key_md5: sse_customer_key_md5.as_deref(), -/// part_number: None, /// }; /// /// if let Some(material) = sse_decryption(request).await? { -/// let (decrypted_reader, plaintext_size) = material.wrap_reader(reader, actual_size).await?; -/// reader = decrypted_reader; -/// content_size = plaintext_size; +/// content_size = material.original_size.unwrap_or(actual_size); /// } /// ``` pub async fn sse_decryption(request: DecryptionRequest<'_>) -> Result, ApiError> { - let is_multipart = is_multipart_object(request.etag, request.parts); - // Check for SSE-C encryption if request .metadata @@ -872,25 +891,14 @@ pub async fn sse_decryption(request: DecryptionRequest<'_>) -> Result Result { - // Build metadata - let mut metadata = HashMap::new(); - - metadata.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string()); - metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), algorithm.clone()); - metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), sse_key_md5); - Ok(EncryptionMaterial { sse_type: SSEType::SseC, server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AES256), kms_key_id: None, algorithm, key_bytes: [0; 32], - nonce: [0; 12], - metadata, + base_nonce: [0; 12], + encrypted_data_key: None, + customer_key_md5: Some(sse_key_md5), + original_size: None, }) } @@ -930,7 +933,6 @@ async fn apply_ssec_encryption_material( sse_key: SSECustomerKey, sse_key_md5: SSECustomerKeyMD5, content_size: i64, - part_number: Option, ) -> Result { let params = SsecParams { algorithm, @@ -942,31 +944,18 @@ async fn apply_ssec_encryption_material( // Generate nonce (deterministic for SSE-C) let base_nonce = generate_ssec_nonce(bucket, key); - let nonce = if let Some(part_num) = part_number { - derive_part_nonce(base_nonce, part_num) - } else { - base_nonce - }; // Build metadata - let mut metadata = HashMap::new(); - - metadata.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string()); - metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), validated.algorithm.clone()); - metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), validated.key_md5.clone()); - metadata.insert( - "x-amz-server-side-encryption-customer-original-size".to_string(), - content_size.to_string(), - ); - Ok(EncryptionMaterial { sse_type: SSEType::SseC, server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AES256), kms_key_id: None, algorithm: validated.algorithm, key_bytes: validated.key_bytes, - nonce, - metadata, + base_nonce, + encrypted_data_key: None, + customer_key_md5: Some(validated.key_md5), + original_size: Some(content_size), }) } @@ -976,7 +965,6 @@ async fn apply_ssec_decryption_material( metadata: &HashMap, sse_key: &str, sse_key_md5: &str, - part_number: Option, ) -> Result { // Validate provided key let algorithm = metadata @@ -994,15 +982,6 @@ async fn apply_ssec_decryption_material( // Generate nonce (same as encryption) let base_nonce = generate_ssec_nonce(bucket, key); - let nonce = if let Some(part_num) = part_number { - derive_part_nonce(base_nonce, part_num) - } else { - base_nonce - }; - - let original_size = metadata - .get("x-amz-server-side-encryption-customer-original-size") - .and_then(|s| s.parse::().ok()); Ok(DecryptionMaterial { sse_type: SSEType::SseC, @@ -1012,11 +991,7 @@ async fn apply_ssec_decryption_material( customer_key_md5: None, key_bytes: validated.key_bytes, - nonce, - original_size, - - is_multipart: false, - parts: Vec::new(), + base_nonce, }) } @@ -1024,21 +999,13 @@ async fn apply_ssec_decryption_material( // Internal Implementation - Managed SSE (SSE-S3 / SSE-KMS) // ============================================================================ -#[allow(clippy::too_many_arguments)] async fn apply_managed_encryption_material( bucket: &str, key: &str, server_side_encryption: ServerSideEncryption, kms_key_id: Option, content_size: i64, - part_number: Option, - part_key: Option, - part_nonce: Option, ) -> Result { - // For multipart, we only generate keys at CompleteMultipartUpload - // During UploadPart, we use the same base nonce with incremented counter - // This is handled externally, so here we just generate the base material - if !is_managed_sse(&server_side_encryption) { return Err(ApiError::from(StorageError::other(format!( "Unsupported server-side encryption: {}", @@ -1052,11 +1019,6 @@ async fn apply_managed_encryption_material( _ => SSEType::SseS3, }; - let mut context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string()); - if content_size >= 0 { - context = context.with_size(content_size as u64); - } - // Determine KMS key ID to use for internal key wrapping. let mut kms_key_candidate = kms_key_id.clone(); if kms_key_candidate.is_none() { @@ -1079,95 +1041,23 @@ async fn apply_managed_encryption_material( }; let provider = get_sse_dek_provider().await?; - - let (data_key, encrypted_data_key) = if let Some(part_number) = part_number - && let Some(part_nonce) = part_nonce - && let Some(part_key) = part_key - && part_number >= 1 - // upload_part mode, dek generate by create_multipart_upload - { - let _base_nonce = BASE64_STANDARD - .decode(part_nonce.as_bytes()) - .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decode nonce: {e}"))))?; - if _base_nonce.len() != 12 { - return Err(ApiError::from(StorageError::other("Invalid encryption nonce length; expected 12 bytes"))); - } - let mut base_nonce_array = [0u8; 12]; - base_nonce_array.copy_from_slice(&_base_nonce[..12]); - let encrypted_data_key = BASE64_STANDARD - .decode(part_key.as_bytes()) - .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decode data key: {e}"))))?; - let _data_key = provider - .decrypt_sse_dek(encrypted_data_key.as_slice(), &kms_key_to_use) - .await?; - let data_key = DataKey { - plaintext_key: _data_key, - nonce: derive_part_nonce(base_nonce_array, part_number), - }; - - // load original data key from metadata - (data_key, encrypted_data_key) - } else { - // Use factory pattern to get provider (test or production mode) - let (data_key, encrypted_data_key) = provider - .generate_sse_dek(bucket, key, &kms_key_to_use) - .await - .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {e}"))))?; - (data_key, encrypted_data_key) - }; + let (data_key, encrypted_data_key) = provider + .generate_sse_dek(bucket, key, &kms_key_to_use) + .await + .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {e}"))))?; let algorithm = server_side_encryption.as_str().to_string(); - let encryption_metadata = EncryptionMetadata { - algorithm: algorithm.clone(), - key_id: kms_key_to_use.clone(), - key_version: 1, - iv: data_key.nonce.to_vec(), - tag: None, - encryption_context: context.encryption_context.clone(), - encrypted_at: jiff::Zoned::now(), - original_size: if content_size >= 0 { content_size as u64 } else { 0 }, - encrypted_data_key, - }; - - // Build metadata headers - let mut metadata = HashMap::new(); - - // Try to use service for metadata formatting if available, otherwise build manually - if let Some(service) = get_global_encryption_service().await { - metadata = service.metadata_to_headers(&encryption_metadata); - } else { - // Manual metadata building for test mode - metadata.insert( - "x-rustfs-encryption-key".to_string(), - BASE64_STANDARD.encode(&encryption_metadata.encrypted_data_key), - ); - metadata.insert("x-rustfs-encryption-iv".to_string(), BASE64_STANDARD.encode(&encryption_metadata.iv)); - metadata.insert("x-rustfs-encryption-algorithm".to_string(), encryption_metadata.algorithm.clone()); - metadata.insert("x-amz-server-side-encryption".to_string(), server_side_encryption.as_str().to_string()); - } - - if matches!(encryption_type, SSEType::SseKms) { - metadata.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), kms_key_to_use.clone()); - } else { - metadata.remove("x-amz-server-side-encryption-aws-kms-key-id"); - } - metadata.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), kms_key_to_use.clone()); - - metadata.insert( - "x-rustfs-encryption-original-size".to_string(), - encryption_metadata.original_size.to_string(), - ); - Ok(EncryptionMaterial { sse_type: encryption_type, server_side_encryption, kms_key_id: matches!(encryption_type, SSEType::SseKms).then_some(kms_key_to_use), algorithm, - key_bytes: data_key.plaintext_key, - nonce: data_key.nonce, - metadata, + base_nonce: data_key.nonce, + encrypted_data_key: Some(encrypted_data_key), + customer_key_md5: None, + original_size: Some(content_size), }) } @@ -1175,7 +1065,6 @@ async fn apply_managed_decryption_material( _bucket: &str, _key: &str, metadata: &HashMap, - part_number: Option, ) -> Result, ApiError> { if !metadata.contains_key("x-rustfs-encryption-key") || !metadata.contains_key("x-amz-server-side-encryption") { return Ok(None); @@ -1240,15 +1129,6 @@ async fn apply_managed_decryption_material( let mut base_nonce = [0u8; 12]; base_nonce.copy_from_slice(&iv[..12]); - let nonce = if let Some(part_num) = part_number { - derive_part_nonce(base_nonce, part_num) - } else { - base_nonce - }; - - let original_size = metadata - .get("x-rustfs-encryption-original-size") - .and_then(|s| s.parse::().ok()); let encryption_type = match server_side_encryption.as_str() { ServerSideEncryption::AES256 => SSEType::SseS3, @@ -1264,11 +1144,7 @@ async fn apply_managed_decryption_material( customer_key_md5: None, key_bytes, - nonce, - original_size, - - is_multipart: false, - parts: Vec::new(), + base_nonce, })) } @@ -1644,57 +1520,6 @@ pub fn strip_managed_encryption_metadata(metadata: &mut HashMap) } } -// ============================================================================ -// Multipart Encryption Support -// ============================================================================ - -pub fn derive_part_nonce(base: [u8; 12], part_number: usize) -> [u8; 12] { - derive_nonce_offset(base, 4, part_number) -} - -#[cfg(test)] -fn derive_legacy_part_nonce(base: [u8; 12], part_number: usize) -> [u8; 12] { - derive_nonce_offset(base, 8, part_number) -} - -fn derive_nonce_offset(mut base: [u8; 12], start: usize, offset: usize) -> [u8; 12] { - let current = u32::from_be_bytes([base[start], base[start + 1], base[start + 2], base[start + 3]]); - let incremented = current.wrapping_add(offset as u32); - base[start..start + 4].copy_from_slice(&incremented.to_be_bytes()); - base -} - -pub(crate) async fn decrypt_multipart_managed_stream( - encrypted_stream: R, - parts: &[ObjectPartInfo], - key_bytes: [u8; 32], - base_nonce: [u8; 12], -) -> Result<(DynReader, i64), StorageError> -where - R: ReadStream + 'static, -{ - let total_plain_size = parts - .iter() - .map(|part| { - if part.actual_size > 0 { - part.actual_size - } else { - part.size as i64 - } - }) - .sum(); - - let multipart_parts = parts.iter().map(|part| part.number).collect(); - let reader = boxed_reader(DecryptReader::new_multipart( - wrap_reader(encrypted_stream), - key_bytes, - base_nonce, - multipart_parts, - )); - - Ok((reader, total_plain_size)) -} - // ============================================================================ // SSE-C Functions // ============================================================================ @@ -1832,6 +1657,7 @@ fn ssec_invalid_request(message: &str) -> ApiError { mod tests { use super::*; use http::HeaderValue; + use rustfs_rio::{DecryptReader, EncryptReader}; #[test] fn test_extract_ssec_params_from_headers() { @@ -1951,212 +1777,6 @@ mod tests { assert!(is_managed_sse(&ServerSideEncryption::from_static("aws:kms"))); } - #[test] - fn test_derive_part_nonce() { - let base = [1, 2, 3, 4, 5, 6, 7, 8, 0, 0, 0, 10]; - let part1 = derive_part_nonce(base, 1); - let part2 = derive_part_nonce(base, 2); - - assert_eq!(&base[..4], &part1[..4]); - assert_eq!(&base[8..], &part1[8..]); - assert_ne!(&base[4..8], &part1[4..8]); - assert_ne!(&part1[4..8], &part2[4..8]); - } - - #[tokio::test] - async fn test_decrypt_multipart_managed_stream_accepts_legacy_part_nonce_layout() { - use std::io::Cursor; - use tokio::io::AsyncReadExt; - - let key_bytes = [7u8; 32]; - let base_nonce = [3u8; 12]; - - let part_one_plaintext = vec![0x11; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 19]; - let part_two_plaintext = vec![0x22; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 37]; - - let part_one_nonce = derive_legacy_part_nonce(base_nonce, 1); - let part_two_nonce = derive_legacy_part_nonce(base_nonce, 2); - - let first_part = { - let mut buf = Vec::new(); - EncryptReader::new(Cursor::new(part_one_plaintext.clone()), key_bytes, part_one_nonce) - .read_to_end(&mut buf) - .await - .unwrap(); - buf - }; - let second_part = { - let mut buf = Vec::new(); - EncryptReader::new(Cursor::new(part_two_plaintext.clone()), key_bytes, part_two_nonce) - .read_to_end(&mut buf) - .await - .unwrap(); - buf - }; - - let mut encrypted_stream = Vec::with_capacity(first_part.len() + second_part.len()); - encrypted_stream.extend_from_slice(&first_part); - encrypted_stream.extend_from_slice(&second_part); - - let parts = vec![ - ObjectPartInfo { - number: 1, - size: first_part.len(), - actual_size: part_one_plaintext.len() as i64, - ..Default::default() - }, - ObjectPartInfo { - number: 2, - size: second_part.len(), - actual_size: part_two_plaintext.len() as i64, - ..Default::default() - }, - ]; - - let (mut decrypted_reader, plaintext_size) = - decrypt_multipart_managed_stream(Cursor::new(encrypted_stream), &parts, key_bytes, base_nonce) - .await - .unwrap(); - - let mut decrypted = Vec::new(); - decrypted_reader.read_to_end(&mut decrypted).await.unwrap(); - - let mut expected = part_one_plaintext; - expected.extend_from_slice(&part_two_plaintext); - - assert_eq!(plaintext_size, expected.len() as i64); - assert_eq!(decrypted, expected); - } - - #[tokio::test] - async fn test_decrypt_multipart_managed_stream_supports_current_nonce_layout() { - use std::io::Cursor; - use tokio::io::AsyncReadExt; - - let key_bytes = [9u8; 32]; - let base_nonce = [5u8; 12]; - - let part_one_plaintext = vec![0x33; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 11]; - let part_two_plaintext = vec![0x44; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE * 2 + 7]; - let part_one_nonce = derive_part_nonce(base_nonce, 1); - let part_two_nonce = derive_part_nonce(base_nonce, 2); - - let first_part = { - let mut buf = Vec::new(); - EncryptReader::new(Cursor::new(part_one_plaintext.clone()), key_bytes, part_one_nonce) - .read_to_end(&mut buf) - .await - .unwrap(); - buf - }; - let second_part = { - let mut buf = Vec::new(); - EncryptReader::new(Cursor::new(part_two_plaintext.clone()), key_bytes, part_two_nonce) - .read_to_end(&mut buf) - .await - .unwrap(); - buf - }; - - let mut encrypted_stream = Vec::with_capacity(first_part.len() + second_part.len()); - encrypted_stream.extend_from_slice(&first_part); - encrypted_stream.extend_from_slice(&second_part); - - let parts = vec![ - ObjectPartInfo { - number: 1, - size: first_part.len(), - actual_size: part_one_plaintext.len() as i64, - ..Default::default() - }, - ObjectPartInfo { - number: 2, - size: second_part.len(), - actual_size: part_two_plaintext.len() as i64, - ..Default::default() - }, - ]; - - let (mut decrypted_reader, plaintext_size) = - decrypt_multipart_managed_stream(Cursor::new(encrypted_stream), &parts, key_bytes, base_nonce) - .await - .unwrap(); - - let mut decrypted = Vec::new(); - decrypted_reader.read_to_end(&mut decrypted).await.unwrap(); - - let mut expected = part_one_plaintext; - expected.extend_from_slice(&part_two_plaintext); - - assert_eq!(plaintext_size, expected.len() as i64); - assert_eq!(decrypted, expected); - } - - #[tokio::test] - async fn test_decrypt_multipart_managed_stream_uses_actual_part_numbers_for_nonce_derivation() { - use std::io::Cursor; - use tokio::io::AsyncReadExt; - - let key_bytes = [0xAu8; 32]; - let base_nonce = [0xBu8; 12]; - - let part_three_plaintext = vec![0x55; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 13]; - let part_five_plaintext = vec![0x66; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 29]; - - let part_three_nonce = derive_part_nonce(base_nonce, 3); - let part_five_nonce = derive_part_nonce(base_nonce, 5); - - let encrypted_three = { - let mut buf = Vec::new(); - EncryptReader::new(Cursor::new(part_three_plaintext.clone()), key_bytes, part_three_nonce) - .read_to_end(&mut buf) - .await - .unwrap(); - buf - }; - let encrypted_five = { - let mut buf = Vec::new(); - EncryptReader::new(Cursor::new(part_five_plaintext.clone()), key_bytes, part_five_nonce) - .read_to_end(&mut buf) - .await - .unwrap(); - buf - }; - - let mut encrypted_stream = Vec::with_capacity(encrypted_three.len() + encrypted_five.len()); - encrypted_stream.extend_from_slice(&encrypted_three); - encrypted_stream.extend_from_slice(&encrypted_five); - - let parts = vec![ - ObjectPartInfo { - number: 3, - size: encrypted_three.len(), - actual_size: part_three_plaintext.len() as i64, - ..Default::default() - }, - ObjectPartInfo { - number: 5, - size: encrypted_five.len(), - actual_size: part_five_plaintext.len() as i64, - ..Default::default() - }, - ]; - - let (mut decrypted_reader, plaintext_size) = - decrypt_multipart_managed_stream(Cursor::new(encrypted_stream), &parts, key_bytes, base_nonce) - .await - .unwrap(); - - let mut decrypted = Vec::new(); - decrypted_reader.read_to_end(&mut decrypted).await.unwrap(); - - let mut expected = part_three_plaintext; - expected.extend_from_slice(&part_five_plaintext); - - assert_eq!(plaintext_size, expected.len() as i64); - assert_eq!(decrypted, expected); - } - #[test] fn test_generate_ssec_nonce() { let nonce1 = generate_ssec_nonce("bucket1", "key1"); @@ -2252,9 +1872,6 @@ mod tests { sse_customer_key: Some(sse_key.clone()), sse_customer_key_md5: None, content_size, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request_missing_md5).await.unwrap_err(); @@ -2269,9 +1886,6 @@ mod tests { sse_customer_key: None, sse_customer_key_md5: Some(sse_key_md5.clone()), content_size, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request_missing_key).await.unwrap_err(); @@ -2286,9 +1900,6 @@ mod tests { sse_customer_key: Some(sse_key), sse_customer_key_md5: Some(sse_key_md5), content_size, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request_missing_algorithm).await.unwrap_err(); @@ -2342,11 +1953,28 @@ mod tests { .await .expect("prepare should accept ssec headers"); assert!(material.is_some()); - let metadata = &material.expect("ssec metadata should be generated").metadata; + let metadata = encryption_material_to_metadata(&material.expect("ssec metadata should be generated")); assert_eq!(metadata.get("x-amz-server-side-encryption").unwrap(), "AES256"); assert_eq!(metadata.get("x-amz-server-side-encryption-customer-algorithm").unwrap(), "AES256"); } + #[test] + fn test_encryption_material_to_metadata_persists_ssec_original_size() { + let metadata = encryption_material_to_metadata(&EncryptionMaterial { + sse_type: SSEType::SseC, + server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AES256), + kms_key_id: None, + algorithm: SSECustomerAlgorithm::from("AES256".to_string()), + key_bytes: [0u8; 32], + base_nonce: [0u8; 12], + encrypted_data_key: None, + customer_key_md5: Some("d41d8cd98f00b204e9800998ecf8427e".to_string()), + original_size: Some(1024), + }); + + assert_eq!(metadata.get(SSEC_ORIGINAL_SIZE_HEADER).map(String::as_str), Some("1024")); + } + #[tokio::test] async fn test_sse_encryption_rejects_kms_key_with_invalid_algorithm() { let bucket = "test-bucket"; @@ -2362,9 +1990,6 @@ mod tests { sse_customer_key: None, sse_customer_key_md5: None, content_size, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request).await.unwrap_err(); @@ -2386,9 +2011,6 @@ mod tests { sse_customer_key: None, sse_customer_key_md5: None, content_size, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request).await.unwrap_err(); @@ -2412,9 +2034,6 @@ mod tests { sse_customer_key: Some(sse_key), sse_customer_key_md5: Some(sse_key_md5), content_size, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request).await.unwrap_err(); @@ -2441,22 +2060,17 @@ mod tests { #[tokio::test] async fn test_sse_encryption_persists_aws_kms_header_for_kms_objects() { - let request = EncryptionRequest { - bucket: "test-bucket", - key: "test-key", - server_side_encryption: Some("aws:kms".to_string().into()), - ssekms_key_id: Some("test-key".to_string()), - sse_customer_algorithm: None, - sse_customer_key: None, - sse_customer_key_md5: None, - content_size: 1024, - part_number: None, - part_key: None, - part_nonce: None, - }; - - let material = sse_encryption(request).await.expect("kms encryption should succeed"); - let metadata = material.expect("managed kms encryption should return material").metadata; + let metadata = encryption_material_to_metadata(&EncryptionMaterial { + sse_type: SSEType::SseKms, + server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS), + kms_key_id: Some("test-key".to_string()), + algorithm: SSECustomerAlgorithm::from(ServerSideEncryption::AWS_KMS.to_string()), + key_bytes: [7u8; 32], + base_nonce: [9u8; 12], + encrypted_data_key: Some(vec![1, 2, 3, 4]), + customer_key_md5: None, + original_size: Some(1024), + }); assert_eq!(metadata.get("x-amz-server-side-encryption").map(String::as_str), Some("aws:kms")); assert_eq!( @@ -2478,21 +2092,16 @@ mod tests { sse_customer_key: None, sse_customer_key_md5: None, content_size: 1024, - part_number: None, - part_key: None, - part_nonce: None, }; let material = sse_encryption(request).await.expect("sse-s3 encryption should succeed"); let material = material.expect("managed sse-s3 encryption should return material"); + let metadata = encryption_material_to_metadata(&material); assert_eq!(material.kms_key_id, None); - assert_eq!(material.metadata.get("x-amz-server-side-encryption").map(String::as_str), Some("AES256")); - assert!(!material.metadata.contains_key("x-amz-server-side-encryption-aws-kms-key-id")); - assert_eq!( - material.metadata.get(INTERNAL_ENCRYPTION_KEY_ID_HEADER).map(String::as_str), - Some("default") - ); + assert_eq!(metadata.get("x-amz-server-side-encryption").map(String::as_str), Some("AES256")); + assert!(!metadata.contains_key("x-amz-server-side-encryption-aws-kms-key-id")); + assert_eq!(metadata.get(INTERNAL_ENCRYPTION_KEY_ID_HEADER).map(String::as_str), Some("default")); } #[test] @@ -2509,32 +2118,6 @@ mod tests { assert!(metadata.contains_key("content-type")); } - #[test] - fn test_is_multipart_object_treats_single_part_multipart_etag_as_multipart() { - let metadata = HashMap::from([("etag".to_string(), "0123456789abcdef0123456789abcdef-1".to_string())]); - let parts = vec![ObjectPartInfo { - number: 1, - size: 128, - actual_size: 64, - ..Default::default() - }]; - - assert!(is_multipart_object(metadata.get("etag").map(String::as_str), &parts)); - } - - #[test] - fn test_is_multipart_object_keeps_regular_single_part_object_as_non_multipart() { - let metadata = HashMap::from([("etag".to_string(), "0123456789abcdef0123456789abcdef".to_string())]); - let parts = vec![ObjectPartInfo { - number: 1, - size: 128, - actual_size: 64, - ..Default::default() - }]; - - assert!(!is_multipart_object(metadata.get("etag").map(String::as_str), &parts)); - } - #[test] fn test_verify_ssec_key_match_success() { let md5 = "test_md5".to_string(); @@ -2572,9 +2155,6 @@ mod tests { sse_customer_key: None, sse_customer_key_md5: None, content_size: 1, - part_number: Some(1), - part_key: None, - part_nonce: None, }; let mismatch = "aBcDeFgHiJkLmNoPqRsTuVwXyZ0123456789+/==".to_string(); @@ -2597,9 +2177,6 @@ mod tests { sse_customer_key: None, sse_customer_key_md5: None, content_size: 1, - part_number: Some(1), - part_key: None, - part_nonce: None, }; let result = request.check_upload_part_customer_key_md5(&metadata, Some(md5)); @@ -3073,6 +2650,33 @@ mod tests { assert_eq!(err.code, S3ErrorCode::InvalidArgument); } + #[test] + fn test_map_get_object_reader_error_converts_missing_ssec_headers_to_invalid_request() { + let err = map_get_object_reader_error(StorageError::other("missing SSE-C algorithm header")); + assert_eq!(err.code, S3ErrorCode::InvalidRequest); + assert_eq!( + err.message, + "The object was stored using a form of Server Side Encryption. The correct parameters must be provided to retrieve the object." + ); + } + + #[test] + fn test_map_get_object_reader_error_converts_ssec_md5_mismatch_to_invalid_request() { + let err = map_get_object_reader_error(StorageError::other("SSE-C key MD5 mismatch")); + assert_eq!(err.code, S3ErrorCode::InvalidRequest); + assert_eq!( + err.message, + "The calculated MD5 hash of the key did not match the hash that was provided." + ); + } + + #[test] + fn test_map_get_object_reader_error_leaves_non_ssec_errors_unchanged() { + let err = map_get_object_reader_error(StorageError::other("plain io failure")); + assert_eq!(err.code, S3ErrorCode::InternalError); + assert_eq!(err.message, "Io error: plain io failure"); + } + #[test] fn test_validate_ssec_params_returns_invalid_request_on_bad_algorithm() { let key = BASE64_STANDARD.encode([42u8; 32]); @@ -3116,9 +2720,6 @@ mod tests { sse_customer_key: None, sse_customer_key_md5: None, content_size: 1024, - part_number: None, - part_key: None, - part_nonce: None, }; let result = sse_encryption(request).await; match &result { @@ -3151,9 +2752,6 @@ mod tests { sse_customer_key: Some(sse_key.clone()), sse_customer_key_md5: Some(wrong_md5), content_size: 1024, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request_wrong_md5).await.unwrap_err(); assert_eq!(err.code, S3ErrorCode::InvalidRequest); @@ -3167,9 +2765,6 @@ mod tests { sse_customer_key: Some(sse_key), sse_customer_key_md5: Some(BASE64_STANDARD.encode(md5::compute([42u8; 32]).0)), content_size: 1024, - part_number: None, - part_key: None, - part_nonce: None, }; let err = sse_encryption(request_unsupported_algorithm).await.unwrap_err(); assert!(err.code == S3ErrorCode::InvalidRequest || err.code == S3ErrorCode::InvalidArgument);