gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-28 09:08:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sse): Temporarily refactored the SSE design for ECStore (#2813)

Browse Source 
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: cxymds <Cxymds@qq.com>
This commit is contained in:
唐小鸭
2026-05-09 22:06:35 +08:00
committed by GitHub
parent 6275918d92
commit 09c2d15057
12 changed files with 1338 additions and 970 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/launch.json
Vendored
+1
View File
@@ -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
Cargo.lock
Generated
+2
View File
@@ -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",
crates/ecstore/Cargo.toml
+2
View File
@@ -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 }
crates/ecstore/src/set_disk.rs
+35 -7
View File
@@ -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<String, S
metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY);
}
fn should_persist_encryption_original_size(metadata: &HashMap<String, String>) -> bool {
metadata.keys().any(|key| is_encryption_metadata_key(key))
|| metadata.contains_key(SSEC_ALGORITHM_HEADER)
|| metadata.contains_key(SSEC_KEY_HEADER)
|| metadata.contains_key(SSEC_KEY_MD5_HEADER)
}
fn capacity_scope_from_disks(disks: &[Option<DiskStore>]) -> 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 {
crates/ecstore/src/store_api/readers.rs
+800 -24
View File
@@ -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<rustfs_rio::Index> {
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<dyn AsyncRead + Unpin + Send + Sync>,
rs: Option<HTTPRangeSpec>,
oi: &ObjectInfo,
opts: &ObjectOptions,
_h: &HeaderMap<HeaderValue>,
h: &HeaderMap<HeaderValue>,
) -> 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<dyn AsyncRead + Unpin + Send + Sync> = 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<dyn AsyncRead + Unpin + Send + Sync> = 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<R: AsyncRead + Unpin + Send + 'static> Drop for StreamConsumer<R> {
}
}
fn encrypted_plaintext_size(oi: &ObjectInfo, is_multipart: bool, is_compressed: bool) -> Result<i64> {
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<usize> {
parts.iter().map(|part| part.number).collect()
}
async fn resolve_encryption_material(oi: &ObjectInfo, headers: &HeaderMap<HeaderValue>) -> Result<EncryptionMaterial> {
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<HeaderValue>) -> Result<EncryptionMaterial> {
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<String, String>) -> Result<EncryptionMaterial> {
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::<Aes256Gcm>::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<Option<[u8; 32]>> {
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<HeaderValue> {
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::<Aes256Gcm>::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::<String>),
("RUSTFS_SSE_S3_MASTER_KEY", None::<String>),
],
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::<u8>::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");
}
}
crates/ecstore/src/store_api/types.rs
+32 -9
View File
@@ -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<Option<i64>> {
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::<i64>()
.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<i64> {
Ok(self.encryption_original_size()?.unwrap_or(self.size))
}
pub fn get_actual_size(&self) -> std::io::Result<i64> {
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::<i64>()
.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);
}
crates/rio/src/encrypt_reader.rs
+8
View File
@@ -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<R> AsyncRead for EncryptReader<R>
@@ -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)
}
crates/rio/src/lib.rs
+1 -1
View File
@@ -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;
rustfs/src/app/multipart_usecase.rs
+193 -178
View File
@@ -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<String, String>,
part_metadata: &HashMap<String, String>,
@@ -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<usize> {
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
rustfs/src/app/object_usecase.rs
+37 -119
View File
@@ -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<usize>,
) -> S3Result<GetObjectPreparedRead<'a>> {
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<T>(input: T, method: Method) -> S3Request<T> {
S3Request {
input,
rustfs/src/storage/mod.rs
+1 -1
View File
@@ -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;
rustfs/src/storage/sse.rs
+226 -631
View File
File diff suppressed because it is too large Load Diff