diff --git a/crates/filemeta/README.md b/crates/filemeta/README.md index 515b79c12..412de3d87 100644 --- a/crates/filemeta/README.md +++ b/crates/filemeta/README.md @@ -19,6 +19,12 @@ **RustFS FileMeta** provides advanced file metadata management and indexing capabilities for the [RustFS](https://rustfs.com) distributed object storage system. For the complete RustFS experience, please visit the [main RustFS repository](https://github.com/rustfs/rustfs). +## Quick Use + +``` +cargo run -p rustfs-filemeta --example dump_fileinfo -- "/path/to/file/xl.meta" +``` + ## ✨ Features - High-performance metadata storage and retrieval diff --git a/crates/filemeta/examples/dump_fileinfo.rs b/crates/filemeta/examples/dump_fileinfo.rs new file mode 100644 index 000000000..80579576a --- /dev/null +++ b/crates/filemeta/examples/dump_fileinfo.rs @@ -0,0 +1,50 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use rustfs_filemeta::{FileInfoOpts, get_file_info}; +use std::{env, fs, path::PathBuf}; +fn main() { + let path = env::args() + .nth(1) + .map(PathBuf::from) + .expect("usage: dump_fileinfo "); + let data = fs::read(&path).expect("read xl.meta"); + let fi = get_file_info( + &data, + "debug-bucket", + "debug-object", + "", + FileInfoOpts { + data: false, + include_free_versions: true, + }, + ) + .expect("decode file info"); + println!("path: {}", path.display()); + println!("size: {}", fi.size); + println!("etag: {:?}", fi.get_etag()); + println!("parts: {}", fi.parts.len()); + for (idx, part) in fi.parts.iter().enumerate() { + println!( + "part#{idx}: number={} size={} actual_size={} etag={}", + part.number, part.size, part.actual_size, part.etag + ); + } + println!("metadata entries: {}", fi.metadata.len()); + let mut keys = fi.metadata.keys().cloned().collect::>(); + keys.sort(); + for key in keys { + println!("meta[{key}]={}", fi.metadata.get(&key).unwrap()); + } +} diff --git a/crates/rio/src/encrypt_reader.rs b/crates/rio/src/encrypt_reader.rs index 80e37d9f3..d83d206a1 100644 --- a/crates/rio/src/encrypt_reader.rs +++ b/crates/rio/src/encrypt_reader.rs @@ -174,6 +174,8 @@ pin_project! { base_nonce: [u8; 12], // Base nonce recorded in object metadata current_nonce_base: [u8; 12], // Active base nonce for the current encrypted segment multipart_mode: bool, + multipart_parts: Vec, + current_part_index: usize, current_part: usize, block_index: usize, buffer: Vec, @@ -200,6 +202,8 @@ where base_nonce: nonce, current_nonce_base: nonce, multipart_mode: false, + multipart_parts: Vec::new(), + current_part_index: 0, current_part: 0, block_index: 0, buffer: Vec::new(), @@ -214,8 +218,8 @@ where } } - pub fn new_multipart(inner: R, key: [u8; 32], base_nonce: [u8; 12]) -> Self { - let first_part = 1; + pub fn new_multipart(inner: R, key: [u8; 32], base_nonce: [u8; 12], multipart_parts: Vec) -> Self { + let first_part = multipart_parts.first().copied().unwrap_or(1); let initial_nonce = derive_part_nonce(&base_nonce, first_part); debug!("decrypt_reader: initialized multipart mode"); @@ -226,6 +230,8 @@ where base_nonce, current_nonce_base: initial_nonce, multipart_mode: true, + multipart_parts, + current_part_index: 0, current_part: first_part, block_index: 0, buffer: Vec::new(), @@ -265,91 +271,95 @@ where return Poll::Ready(Ok(())); } - // Read header (8 bytes) - while !*this.header_done && *this.header_read < 8 { - let mut temp = [0u8; 8]; - let mut temp_buf = ReadBuf::new(&mut temp[0..8 - *this.header_read]); - match this.inner.as_mut().poll_read(cx, &mut temp_buf) { - Poll::Pending => return Poll::Pending, - Poll::Ready(Ok(())) => { - let n = temp_buf.filled().len(); - if n == 0 { - if *this.header_read == 0 { - *this.finished = true; - return Poll::Ready(Ok(())); + if *this.ciphertext_len == 0 { + // Read header (8 bytes) only when there is no in-flight payload. + while !*this.header_done && *this.header_read < 8 { + let mut temp = [0u8; 8]; + let mut temp_buf = ReadBuf::new(&mut temp[0..8 - *this.header_read]); + match this.inner.as_mut().poll_read(cx, &mut temp_buf) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Ok(())) => { + let n = temp_buf.filled().len(); + if n == 0 { + if *this.header_read == 0 { + *this.finished = true; + return Poll::Ready(Ok(())); + } + return Poll::Ready(Err(Error::new( + std::io::ErrorKind::UnexpectedEof, + "unexpected EOF while reading encrypted block header", + ))); } - return Poll::Ready(Err(Error::new( - std::io::ErrorKind::UnexpectedEof, - "unexpected EOF while reading encrypted block header", - ))); + this.header_buf[*this.header_read..*this.header_read + n].copy_from_slice(&temp_buf.filled()[..n]); + *this.header_read += n; } - this.header_buf[*this.header_read..*this.header_read + n].copy_from_slice(&temp_buf.filled()[..n]); - *this.header_read += n; + Poll::Ready(Err(e)) => return Poll::Ready(Err(e)), } - Poll::Ready(Err(e)) => return Poll::Ready(Err(e)), } - } - if !*this.header_done && *this.header_read == 8 { - *this.header_done = true; - } + if !*this.header_done && *this.header_read == 8 { + *this.header_done = true; + } - if !*this.header_done { - return Poll::Pending; - } + if !*this.header_done { + return Poll::Pending; + } - let typ = this.header_buf[0]; - let len = - (this.header_buf[1] as usize) | ((this.header_buf[2] as usize) << 8) | ((this.header_buf[3] as usize) << 16); - let crc = (this.header_buf[4] as u32) - | ((this.header_buf[5] as u32) << 8) - | ((this.header_buf[6] as u32) << 16) - | ((this.header_buf[7] as u32) << 24); + let typ = this.header_buf[0]; + let len = + (this.header_buf[1] as usize) | ((this.header_buf[2] as usize) << 8) | ((this.header_buf[3] as usize) << 16); + *this.header_read = 0; + *this.header_done = false; - *this.header_read = 0; - *this.header_done = false; + if typ == 0xFF { + if *this.multipart_mode { + let next_part = if *this.current_part_index + 1 < this.multipart_parts.len() { + *this.current_part_index += 1; + this.multipart_parts[*this.current_part_index] + } else { + *this.current_part + 1 + }; + debug!( + next_part = next_part, + "decrypt_reader: reached segment terminator, advancing to next part" + ); + *this.current_part = next_part; + *this.current_nonce_base = derive_part_nonce(this.base_nonce, *this.current_part); + *this.block_index = 0; + *this.ciphertext_read = 0; + *this.ciphertext_len = 0; + continue; + } - if typ == 0xFF { - if *this.multipart_mode { - debug!( - next_part = *this.current_part + 1, - "decrypt_reader: reached segment terminator, advancing to next part" - ); - *this.current_part += 1; - *this.current_nonce_base = derive_part_nonce(this.base_nonce, *this.current_part); + *this.finished = true; *this.block_index = 0; *this.ciphertext_read = 0; *this.ciphertext_len = 0; continue; } - *this.finished = true; + tracing::debug!(typ = typ, len = len, "decrypt block header"); + + if len == 0 { + tracing::warn!("encountered zero-length encrypted block, treating as end of stream"); + *this.finished = true; + *this.ciphertext_read = 0; + *this.ciphertext_len = 0; + continue; + } + + let Some(payload_len) = len.checked_sub(4) else { + tracing::error!("invalid encrypted block length: typ={} len={} header={:?}", typ, len, this.header_buf); + return Poll::Ready(Err(Error::other("Invalid encrypted block length"))); + }; + + if this.ciphertext_buf.len() < payload_len { + this.ciphertext_buf.resize(payload_len, 0); + } + *this.ciphertext_len = payload_len; *this.ciphertext_read = 0; - *this.ciphertext_len = 0; - continue; } - tracing::debug!(typ = typ, len = len, "decrypt block header"); - - if len == 0 { - tracing::warn!("encountered zero-length encrypted block, treating as end of stream"); - *this.finished = true; - *this.ciphertext_read = 0; - *this.ciphertext_len = 0; - continue; - } - - let Some(payload_len) = len.checked_sub(4) else { - tracing::error!("invalid encrypted block length: typ={} len={} header={:?}", typ, len, this.header_buf); - return Poll::Ready(Err(Error::other("Invalid encrypted block length"))); - }; - - if this.ciphertext_buf.len() < payload_len { - this.ciphertext_buf.resize(payload_len, 0); - } - *this.ciphertext_len = payload_len; - *this.ciphertext_read = 0; - while *this.ciphertext_read < *this.ciphertext_len { let mut temp_buf = ReadBuf::new(&mut this.ciphertext_buf[*this.ciphertext_read..*this.ciphertext_len]); match this.inner.as_mut().poll_read(cx, &mut temp_buf) { @@ -420,12 +430,16 @@ where return Poll::Ready(Err(Error::other("Plaintext length mismatch"))); } + let expected_crc = (this.header_buf[4] as u32) + | ((this.header_buf[5] as u32) << 8) + | ((this.header_buf[6] as u32) << 16) + | ((this.header_buf[7] as u32) << 24); let actual_crc = { let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); hasher.update(&plaintext); hasher.finalize() as u32 }; - if actual_crc != crc { + if actual_crc != expected_crc { *this.ciphertext_read = 0; *this.ciphertext_len = 0; return Poll::Ready(Err(Error::other("CRC32 mismatch"))); @@ -528,6 +542,49 @@ mod tests { } } + struct PendingChunkedCursor { + inner: Cursor>, + max_chunk: usize, + should_pending: bool, + } + + impl PendingChunkedCursor { + fn new(data: Vec, max_chunk: usize) -> Self { + Self { + inner: Cursor::new(data), + max_chunk, + should_pending: true, + } + } + } + + impl AsyncRead for PendingChunkedCursor { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if self.should_pending { + self.should_pending = false; + cx.waker().wake_by_ref(); + return Poll::Pending; + } + + if self.max_chunk == 0 || buf.remaining() == 0 { + return Poll::Ready(Ok(())); + } + + let remaining = self.inner.get_ref().len() as u64 - self.inner.position(); + if remaining == 0 { + return Poll::Ready(Ok(())); + } + + let to_read = remaining.min(self.max_chunk as u64).min(buf.remaining() as u64) as usize; + let start = self.inner.position() as usize; + let end = start + to_read; + buf.put_slice(&self.inner.get_ref()[start..end]); + self.inner.set_position(end as u64); + self.should_pending = true; + Poll::Ready(Ok(())) + } + } + fn encrypt_with_legacy_nonce_reuse(data: &[u8], key: [u8; 32], nonce: [u8; 12]) -> Vec { let cipher = Aes256Gcm::new_from_slice(&key).expect("valid key"); let nonce = Nonce::try_from(nonce.as_slice()).expect("valid nonce"); @@ -697,6 +754,29 @@ mod tests { assert_eq!(decrypted, data); } + #[tokio::test] + async fn test_decrypt_reader_large_with_pending_chunks() { + let size = 1024 * 1024; + let mut data = vec![0u8; size]; + rand::rng().fill(&mut data[..]); + let mut key = [0u8; 32]; + let mut nonce = [0u8; 12]; + rand::rng().fill_bytes(&mut key); + rand::rng().fill_bytes(&mut nonce); + + let reader = Cursor::new(data.clone()); + let mut encrypt_reader = EncryptReader::new(reader, key, nonce); + let mut encrypted = Vec::new(); + encrypt_reader.read_to_end(&mut encrypted).await.unwrap(); + + let reader = PendingChunkedCursor::new(encrypted, 3); + let mut decrypt_reader = DecryptReader::new(reader, key, nonce); + let mut decrypted = Vec::new(); + decrypt_reader.read_to_end(&mut decrypted).await.unwrap(); + + assert_eq!(decrypted, data); + } + #[tokio::test] async fn test_decrypt_reader_large_through_reader_stream() { let size = 1024 * 1024; @@ -781,7 +861,7 @@ mod tests { combined.extend_from_slice(&encrypted_two); let reader = BufReader::new(Cursor::new(combined)); - let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce); + let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce, vec![1, 2]); let mut decrypted = Vec::new(); decrypt_reader.read_to_end(&mut decrypted).await.unwrap(); @@ -855,7 +935,7 @@ mod tests { combined.extend_from_slice(&encrypted_two); let reader = BufReader::new(Cursor::new(combined)); - let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce); + let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce, vec![1, 2]); let mut decrypted = Vec::new(); decrypt_reader.read_to_end(&mut decrypted).await.unwrap(); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 9f9f9f041..a03853c43 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -68,6 +68,15 @@ use tracing::{instrument, warn}; use urlencoding::encode; use uuid::Uuid; +fn merge_part_encryption_metadata( + metadata: &HashMap, + part_metadata: &HashMap, +) -> HashMap { + let mut merged = metadata.clone(); + merged.extend(part_metadata.clone()); + merged +} + /// 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<()> { @@ -644,7 +653,7 @@ impl DefaultMultipartUsecase { }; let opts = ObjectOptions::default(); - let mut fi = store + let fi = store .get_multipart_info(&bucket, &key, &upload_id, &opts) .await .map_err(ApiError::from)?; @@ -753,8 +762,6 @@ impl DefaultMultipartUsecase { HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) .map_err(ApiError::from)?; - fi.user_defined.extend(material.metadata); - (requested_sse, requested_kms_key_id) } None => (None, None), @@ -924,7 +931,7 @@ impl DefaultMultipartUsecase { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - let mut mp_info = store + let mp_info = store .get_multipart_info(&bucket, &key, &upload_id, &ObjectOptions::default()) .await .map_err(ApiError::from)?; @@ -1113,7 +1120,7 @@ impl DefaultMultipartUsecase { encryption_request.check_upload_part_customer_key_md5(&mp_info.user_defined, sse_customer_key_md5.clone())?; - let (requested_sse, requested_kms_key_id) = match sse_encryption(encryption_request).await? { + 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(); @@ -1122,18 +1129,17 @@ impl DefaultMultipartUsecase { 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); - mp_info.user_defined.extend(material.metadata); - - (requested_sse, requested_kms_key_id) + (requested_sse, requested_kms_key_id, dst_user_defined) } - None => (None, None), + None => (None, None, mp_info.user_defined.clone()), }; let mut reader = PutObjReader::new(reader); let dst_opts = ObjectOptions { - user_defined: mp_info.user_defined.clone(), + user_defined: dst_user_defined, ..Default::default() }; @@ -1166,6 +1172,9 @@ impl DefaultMultipartUsecase { mod tests { use super::*; use http::{Extensions, HeaderMap, Method, Uri, header::HeaderValue}; + use rustfs_filemeta::ObjectPartInfo; + use std::{collections::HashMap, io::Cursor}; + use tokio::io::AsyncReadExt; fn build_request(input: T, method: Method) -> S3Request { S3Request { @@ -1220,6 +1229,161 @@ mod tests { assert_eq!(location, "/bucket/nested/object"); } + #[test] + fn merge_part_encryption_metadata_keeps_source_metadata_unchanged() { + let multipart_metadata = HashMap::from([ + ("x-rustfs-encryption-iv".to_string(), "base-nonce".to_string()), + ("x-rustfs-encryption-key".to_string(), "base-key".to_string()), + ]); + let part_metadata = HashMap::from([ + ("x-rustfs-encryption-iv".to_string(), "part-nonce".to_string()), + ("x-rustfs-encryption-original-size".to_string(), "1024".to_string()), + ]); + + let merged = merge_part_encryption_metadata(&multipart_metadata, &part_metadata); + + assert_eq!(multipart_metadata.get("x-rustfs-encryption-iv").map(String::as_str), Some("base-nonce")); + assert_eq!(merged.get("x-rustfs-encryption-iv").map(String::as_str), Some("part-nonce")); + assert_eq!(merged.get("x-rustfs-encryption-key").map(String::as_str), Some("base-key")); + } + + #[tokio::test] + async fn managed_multipart_roundtrip_preserves_session_nonce_between_parts() { + let prepare_request = PrepareEncryptionRequest { + bucket: "bucket", + key: "object", + server_side_encryption: Some(ServerSideEncryption::from_static(ServerSideEncryption::AES256)), + ssekms_key_id: None, + sse_customer_algorithm: None, + sse_customer_key_md5: None, + }; + let session_material = sse_prepare_encryption(prepare_request) + .await + .expect("prepare multipart encryption") + .expect("managed multipart session material"); + let session_metadata = session_material.metadata.clone(); + 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 { + 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, + 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("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"); + + let part_two_material = sse_encryption(EncryptionRequest { + 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, + 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("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"); + + 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 { + number: 1, + size: encrypted_one.len(), + actual_size: part_one_plaintext.len() as i64, + ..Default::default() + }, + ObjectPartInfo { + number: 2, + size: encrypted_two.len(), + actual_size: part_two_plaintext.len() as i64, + ..Default::default() + }, + ]; + + let mut encrypted_stream = Vec::with_capacity(encrypted_one.len() + encrypted_two.len()); + encrypted_stream.extend_from_slice(&encrypted_one); + encrypted_stream.extend_from_slice(&encrypted_two); + + let decryption_material = sse_decryption(DecryptionRequest { + bucket: "bucket", + key: "object", + 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 mut decrypted = Vec::new(); + decrypted_reader + .read_to_end(&mut decrypted) + .await + .expect("read decrypted multipart data"); + + 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 execute_abort_multipart_upload_returns_internal_error_when_store_uninitialized() { let input = AbortMultipartUploadInput::builder() diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index c0802262d..533e78c25 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -547,7 +547,7 @@ pub struct DecryptionRequest<'a> { /// 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, + 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. @@ -1318,18 +1318,20 @@ pub trait SseDekProvider: Send + Sync { // ============================================================================ /// Production KMS-backed DEK provider -/// Wraps the global ObjectEncryptionService to provide SSE DEK operations -struct KmsSseDekProvider { - service: Arc, -} +/// Resolves the latest global ObjectEncryptionService on each call. +struct KmsSseDekProvider; impl KmsSseDekProvider { /// Create a new KMS-backed provider pub async fn new() -> Result { - let service = get_global_encryption_service() + Self::current_service() .await .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?; - Ok(Self { service }) + Ok(Self) + } + + async fn current_service() -> Option> { + get_global_encryption_service().await } } @@ -1339,8 +1341,10 @@ impl SseDekProvider for KmsSseDekProvider { let context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string()); let kms_key_option = Some(kms_key_id.to_string()); - let (data_key, encrypted_data_key) = self - .service + let service = Self::current_service() + .await + .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?; + let (data_key, encrypted_data_key) = service .create_data_key(&kms_key_option, &context) .await .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {}", e))))?; @@ -1351,8 +1355,10 @@ impl SseDekProvider for KmsSseDekProvider { async fn decrypt_sse_dek(&self, encrypted_dek: &[u8], _kms_key_id: &str) -> Result<[u8; 32], ApiError> { // Create a minimal context for decryption let context = ObjectEncryptionContext::new("".to_string(), "".to_string()); - let data_key = self - .service + let service = Self::current_service() + .await + .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?; + let data_key = service .decrypt_data_key(encrypted_dek, &context) .await .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {}", e))))?; @@ -1678,7 +1684,13 @@ where }) .sum(); - let reader = boxed_reader(DecryptReader::new_multipart(wrap_reader(encrypted_stream), key_bytes, base_nonce)); + 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)) } @@ -2080,6 +2092,71 @@ mod tests { 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"); @@ -2762,6 +2839,73 @@ mod tests { println!("✅ Full cycle (generate -> encrypt DEK -> decrypt DEK -> decrypt data) test passed!"); } + #[tokio::test] + async fn test_kms_sse_dek_provider_uses_latest_reconfigured_service() { + use rustfs_kms::config::KmsConfig; + use rustfs_kms::types::{CreateKeyRequest, KeyUsage}; + use std::sync::OnceLock; + use tempfile::TempDir; + use tokio::sync::Mutex; + + static KMS_TEST_LOCK: OnceLock> = OnceLock::new(); + let _guard = KMS_TEST_LOCK.get_or_init(|| Mutex::new(())).lock().await; + + let manager = rustfs_kms::init_global_kms_service_manager(); + + let first_dir = TempDir::new().expect("first temp dir"); + manager + .reconfigure(KmsConfig::local(first_dir.path().to_path_buf())) + .await + .expect("first KMS reconfigure should succeed"); + manager + .get_encryption_service() + .await + .expect("first encryption service should exist") + .create_key(CreateKeyRequest { + key_name: Some("first-key".to_string()), + key_usage: KeyUsage::EncryptDecrypt, + description: None, + policy: None, + tags: HashMap::new(), + origin: None, + }) + .await + .expect("first key should be created"); + + let provider = KmsSseDekProvider::new().await.expect("provider should initialize"); + provider + .generate_sse_dek("bucket", "object", "first-key") + .await + .expect("provider should use the initial service"); + + let second_dir = TempDir::new().expect("second temp dir"); + manager + .reconfigure(KmsConfig::local(second_dir.path().to_path_buf())) + .await + .expect("second KMS reconfigure should succeed"); + manager + .get_encryption_service() + .await + .expect("second encryption service should exist") + .create_key(CreateKeyRequest { + key_name: Some("second-key".to_string()), + key_usage: KeyUsage::EncryptDecrypt, + description: None, + policy: None, + tags: HashMap::new(), + origin: None, + }) + .await + .expect("second key should be created"); + + provider + .generate_sse_dek("bucket", "object", "second-key") + .await + .expect("provider should resolve the latest reconfigured service"); + + manager.stop().await.expect("kms service should stop cleanly"); + } + #[test] fn test_encryption_type_enum() { // Test EncryptionType enum