From 457f4e0170375b4d926315c614ce53ea3b672c03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sun, 19 Apr 2026 10:10:42 +0800 Subject: [PATCH] fix(s3): improve GitLab registry compatibility (#2596) Co-authored-by: houseme --- crates/ecstore/src/set_disk.rs | 92 ++++++++++----- crates/rio/src/hash_reader.rs | 45 +++++++- .../tests/lifecycle_integration_test.rs | 3 +- rustfs/src/app/multipart_usecase.rs | 31 ++++- rustfs/src/app/object_usecase.rs | 44 ++++++-- rustfs/src/server/http.rs | 10 +- rustfs/src/server/layer.rs | 106 ++++++++++++++++++ 7 files changed, 292 insertions(+), 39 deletions(-) diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 9ab1f0828..2ace75f3e 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -2808,18 +2808,10 @@ impl MultipartOperations for SetDisks { if part_numbers.is_empty() { return Ok(ret); } - let start_op = part_numbers.iter().find(|&&v| v != 0 && v == part_number_marker); - if part_number_marker > 0 && start_op.is_none() { + let Some(remaining_part_numbers) = parts_after_marker(&part_numbers, part_number_marker) else { return Ok(ret); - } - - if let Some(start) = start_op { - if start + 1 > part_numbers.len() { - return Ok(ret); - } - - part_numbers = part_numbers[start + 1..].to_vec(); - } + }; + part_numbers = remaining_part_numbers.to_vec(); let mut parts = Vec::with_capacity(part_numbers.len()); @@ -3346,22 +3338,17 @@ impl MultipartOperations for SetDisks { return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default())); }; - let part_crc = match checksum_type { - rustfs_rio::ChecksumType::SHA256 => p.checksum_sha256.clone(), - rustfs_rio::ChecksumType::SHA1 => p.checksum_sha1.clone(), - rustfs_rio::ChecksumType::CRC32 => p.checksum_crc32.clone(), - rustfs_rio::ChecksumType::CRC32C => p.checksum_crc32c.clone(), - rustfs_rio::ChecksumType::CRC64_NVME => p.checksum_crc64nvme.clone(), - _ => { - error!( - "complete_multipart_upload checksum type={checksum_type}, part_id={}, bucket={}, object={}", - p.part_num, bucket, object - ); - return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default())); - } + let Some(part_crc) = complete_part_checksum(p, checksum_type) else { + error!( + "complete_multipart_upload checksum type={checksum_type}, part_id={}, bucket={}, object={}", + p.part_num, bucket, object + ); + return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default())); }; - if part_crc.clone().unwrap_or_default() != crc { + if let Some(part_crc) = part_crc + && part_crc != crc + { error!("complete_multipart_upload checksum_type={checksum_type:?}, part_crc={part_crc:?}, crc={crc:?}"); error!( "complete_multipart_upload checksum mismatch part_id={}, bucket={}, object={}", @@ -4283,6 +4270,28 @@ fn get_complete_multipart_md5(parts: &[CompletePart]) -> String { format!("{}-{}", etag_hex, parts.len()) } +fn complete_part_checksum(part: &CompletePart, checksum_type: rustfs_rio::ChecksumType) -> Option> { + match checksum_type.base() { + rustfs_rio::ChecksumType::SHA256 => Some(part.checksum_sha256.clone()), + rustfs_rio::ChecksumType::SHA1 => Some(part.checksum_sha1.clone()), + rustfs_rio::ChecksumType::CRC32 => Some(part.checksum_crc32.clone()), + rustfs_rio::ChecksumType::CRC32C => Some(part.checksum_crc32c.clone()), + rustfs_rio::ChecksumType::CRC64_NVME => Some(part.checksum_crc64nvme.clone()), + _ => None, + } +} + +fn parts_after_marker(part_numbers: &[usize], part_number_marker: usize) -> Option<&[usize]> { + if part_number_marker == 0 { + return Some(part_numbers); + } + + part_numbers + .iter() + .position(|&part_number| part_number != 0 && part_number == part_number_marker) + .map(|index| &part_numbers[index + 1..]) +} + pub fn canonicalize_etag(etag: &str) -> String { let re = Regex::new("\"*?([^\"]*?)\"*?$").unwrap(); re.replace_all(etag, "$1").to_string() @@ -5547,6 +5556,39 @@ mod tests { assert!(!is_valid_storage_class("standard")); // lowercase } + #[test] + fn complete_part_checksum_accepts_missing_value_and_uses_base_type() { + let missing_checksum_part = CompletePart::default(); + assert_eq!( + complete_part_checksum(&missing_checksum_part, rustfs_rio::ChecksumType::CRC64_NVME), + Some(None) + ); + + let full_object_crc32 = + rustfs_rio::ChecksumType(rustfs_rio::ChecksumType::CRC32.0 | rustfs_rio::ChecksumType::FULL_OBJECT.0); + let part = CompletePart { + checksum_crc32: Some("AAAAAA==".to_string()), + ..Default::default() + }; + assert_eq!(complete_part_checksum(&part, full_object_crc32), Some(Some("AAAAAA==".to_string()))); + } + + #[test] + fn parts_after_marker_uses_marker_position() { + let part_numbers = (1..=1002).collect::>(); + + let remaining = parts_after_marker(&part_numbers, 1000).expect("marker should exist"); + + assert_eq!(remaining, &[1001, 1002]); + } + + #[test] + fn parts_after_marker_returns_none_for_missing_marker() { + let part_numbers = vec![1, 2, 3]; + + assert!(parts_after_marker(&part_numbers, 4).is_none()); + } + #[test] fn test_is_cold_storage_class() { // Test cold storage classes diff --git a/crates/rio/src/hash_reader.rs b/crates/rio/src/hash_reader.rs index 9af705641..744886a75 100644 --- a/crates/rio/src/hash_reader.rs +++ b/crates/rio/src/hash_reader.rs @@ -408,6 +408,24 @@ impl HashReader { Ok(()) } + pub fn add_calculated_checksum(&mut self, checksum_type: ChecksumType) -> Result<(), std::io::Error> { + if !checksum_type.is_set() { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid checksum type")); + } + + let Some(hasher) = checksum_type.hasher() else { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid checksum type")); + }; + + self.content_hash = Some(Checksum { + checksum_type, + ..Default::default() + }); + self.content_hasher = Some(hasher); + + Ok(()) + } + pub fn checksum(&self) -> Option { if self .content_hash @@ -569,7 +587,13 @@ impl AsyncRead for HashReader { let content_hash = hasher.finalize(); - if content_hash != expected_content_hash.raw { + if expected_content_hash.raw.is_empty() + && expected_content_hash.encoded.is_empty() + && !expected_content_hash.checksum_type.trailing() + { + expected_content_hash.raw = content_hash; + expected_content_hash.encoded = general_purpose::STANDARD.encode(&expected_content_hash.raw); + } else if content_hash != expected_content_hash.raw { let expected_hex = hex_simd::encode_to_string(&expected_content_hash.raw, hex_simd::AsciiCase::Lower); let actual_hex = hex_simd::encode_to_string(content_hash, hex_simd::AsciiCase::Lower); error!( @@ -742,6 +766,25 @@ mod tests { assert_eq!(buf, data); } + #[tokio::test] + async fn test_add_calculated_checksum_records_checksum() { + let data = b"server-side copy checksum"; + let reader = BufReader::new(Cursor::new(&data[..])); + let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false).unwrap(); + + hash_reader.add_calculated_checksum(ChecksumType::CRC64_NVME).unwrap(); + + let mut buf = Vec::new(); + hash_reader.read_to_end(&mut buf).await.unwrap(); + + let expected = Checksum::new_from_data(ChecksumType::CRC64_NVME, data).unwrap(); + let checksums = hash_reader.content_crc(); + + assert_eq!(buf, data); + assert_eq!(hash_reader.content_crc_type(), Some(ChecksumType::CRC64_NVME)); + assert_eq!(checksums.get("CRC64NVME"), Some(&expected.encoded)); + } + #[tokio::test] async fn test_hashreader_new_logic() { let data = b"test data"; diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 47a8eda55..4608b58e7 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -1225,10 +1225,11 @@ mod serial_tests { let object_name = "test/object.txt"; create_test_bucket(&ecstore, bucket_name.as_str()).await; + upload_test_object(&ecstore, bucket_name.as_str(), object_name, b"expire immediately").await; + set_bucket_lifecycle(bucket_name.as_str()) .await .expect("Failed to set lifecycle configuration"); - upload_test_object(&ecstore, bucket_name.as_str(), object_name, b"expire immediately").await; assert!(object_exists(&ecstore, bucket_name.as_str(), object_name).await); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index a03853c43..03f8f5955 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -1136,6 +1136,28 @@ impl DefaultMultipartUsecase { None => (None, None, mp_info.user_defined.clone()), }; + if let Some(checksum_algorithm) = mp_info + .user_defined + .get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM) + .filter(|checksum_algorithm| !checksum_algorithm.is_empty()) + { + let checksum_type = rustfs_rio::ChecksumType::from_string_with_obj_type( + checksum_algorithm, + mp_info + .user_defined + .get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE) + .map(String::as_str) + .unwrap_or_default(), + ); + if !checksum_type.is_set() { + return Err(ApiError::from(StorageError::other(format!( + "Invalid multipart checksum type: {checksum_algorithm}" + ))) + .into()); + } + reader.add_calculated_checksum(checksum_type).map_err(ApiError::from)?; + } + let mut reader = PutObjReader::new(reader); let dst_opts = ObjectOptions { @@ -1148,10 +1170,17 @@ impl DefaultMultipartUsecase { .await .map_err(ApiError::from)?; + let copy_checksums = reader.as_hash_reader().content_crc(); + let checksum_value = |checksum_type: rustfs_rio::ChecksumType| copy_checksums.get(&checksum_type.to_string()).cloned(); + let copy_part_result = CopyPartResult { + checksum_crc32: checksum_value(rustfs_rio::ChecksumType::CRC32), + checksum_crc32c: checksum_value(rustfs_rio::ChecksumType::CRC32C), + checksum_sha1: checksum_value(rustfs_rio::ChecksumType::SHA1), + checksum_sha256: checksum_value(rustfs_rio::ChecksumType::SHA256), + checksum_crc64nvme: checksum_value(rustfs_rio::ChecksumType::CRC64_NVME), e_tag: part_info.etag.map(|etag| to_s3s_etag(&etag)), last_modified: part_info.last_mod.map(Timestamp::from), - ..Default::default() }; let output = UploadPartCopyOutput { diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 631b07319..2711d9675 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -67,6 +67,7 @@ use rustfs_ecstore::bucket::{ }; use rustfs_ecstore::client::object_api_utils::to_s3s_etag; use rustfs_ecstore::compress::{MIN_COMPRESSIBLE_SIZE, is_compressible}; +use rustfs_ecstore::config::storageclass; use rustfs_ecstore::disk::{error::DiskError, error_reduce::is_all_buckets_not_found}; use rustfs_ecstore::error::{StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found}; use rustfs_ecstore::new_object_layer_fn; @@ -750,6 +751,14 @@ fn apply_put_request_metadata( Ok(()) } +fn response_storage_class(info: &ObjectInfo, metadata: &HashMap) -> Option { + info.storage_class + .clone() + .or_else(|| metadata.get(AMZ_STORAGE_CLASS).cloned()) + .filter(|storage_class| !storage_class.is_empty() && storage_class != storageclass::STANDARD) + .map(StorageClass::from) +} + async fn apply_put_request_object_lock_opts( bucket: &str, object_lock_legal_hold_status: Option, @@ -2007,6 +2016,7 @@ impl DefaultObjectUsecase { // x-amz-expiration: predict from lifecycle configuration let expiration = resolve_put_object_expiration(bucket, &info).await; + let storage_class = response_storage_class(&info, &info.user_defined); let output = GetObjectOutput { body, @@ -2031,6 +2041,7 @@ impl DefaultObjectUsecase { version_id: output_version_id, restore, expiration, + storage_class, ..Default::default() }; @@ -3449,13 +3460,7 @@ impl DefaultObjectUsecase { .map(|v| SSECustomerAlgorithm::from(v.clone())); let sse_customer_key_md5 = metadata_map.get("x-amz-server-side-encryption-customer-key-md5").cloned(); let sse_kms_key_id = metadata_map.get("x-amz-server-side-encryption-aws-kms-key-id").cloned(); - // Prefer explicit storage_class from object info; fall back to persisted metadata header. - let storage_class = info - .storage_class - .clone() - .or_else(|| metadata_map.get("x-amz-storage-class").cloned()) - .filter(|s| !s.is_empty()) - .map(StorageClass::from); + let storage_class = response_storage_class(&info, &metadata_map); let mut checksum_crc32 = None; let mut checksum_crc32c = None; let mut checksum_sha1 = None; @@ -4643,6 +4648,31 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidStorageClass); } + #[test] + fn response_storage_class_omits_standard_and_keeps_non_default() { + let metadata = HashMap::new(); + let standard_info = ObjectInfo { + storage_class: Some(storageclass::STANDARD.to_string()), + user_defined: metadata.clone(), + ..Default::default() + }; + assert!(response_storage_class(&standard_info, &metadata).is_none()); + + let mut metadata = HashMap::new(); + metadata.insert(AMZ_STORAGE_CLASS.to_string(), storageclass::STANDARD_IA.to_string()); + let infrequent_access_info = ObjectInfo { + storage_class: Some(storageclass::STANDARD_IA.to_string()), + user_defined: metadata.clone(), + ..Default::default() + }; + assert_eq!( + response_storage_class(&infrequent_access_info, &metadata) + .as_ref() + .map(StorageClass::as_str), + Some(storageclass::STANDARD_IA) + ); + } + #[tokio::test] async fn execute_get_object_rejects_zero_part_number() { let input = GetObjectInput::builder() diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 3fad55f40..1cb4a1e03 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -23,7 +23,7 @@ use crate::server::{ hybrid::hybrid, layer::{ AdminChunkedContentLengthCompatLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer, RedirectLayer, - RequestContextLayer, + RequestContextLayer, S3ErrorMessageCompatLayer, }, tls_material::{TlsAcceptorHolder, TlsHandshakeFailureKind, TlsMaterialSnapshot, spawn_reload_loop}, }; @@ -589,9 +589,10 @@ fn process_connection( // 11. PropagateRequestIdLayer — X-Request-ID → response // 12. CompressionLayer — response compression (whitelist, path-aware) // 13. PathCategoryInjectionLayer — injects path category for compression predicate - // 14. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes - // 15. ConditionalCorsLayer — S3 API CORS - // 16. RedirectLayer — console redirect (conditional) + // 14. S3ErrorMessageCompatLayer — missing S3 error message compatibility + // 15. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes + // 16. ConditionalCorsLayer — S3 API CORS + // 17. RedirectLayer — console redirect (conditional) // ───────────────────────────────────────────────────────────── let hybrid_service = ServiceBuilder::new() // NOTE: Both extension types are intentionally inserted to maintain compatibility: @@ -725,6 +726,7 @@ fn process_connection( // Only compresses when enabled and matches configured extensions/MIME types .layer(CompressionLayer::new().compress_when(PathAwareCompressionPredicate::new(compression_config))) .layer(PathCategoryInjectionLayer) + .layer(S3ErrorMessageCompatLayer) .layer(ObjectAttributesEtagFixLayer) // Conditional CORS layer: only applies to S3 API requests (not Admin, not Console) // Admin has its own CORS handling in router.rs diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index f9dfeb87a..e4b9c7743 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::admin::console::is_console_path; +use crate::error::ApiError; use crate::server::cors; use crate::server::hybrid::HybridBody; use crate::server::{ADMIN_PREFIX, CONSOLE_PREFIX, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, RUSTFS_ADMIN_PREFIX}; @@ -27,6 +28,7 @@ use opentelemetry::global; use opentelemetry::trace::TraceContextExt; use rustfs_utils::get_env_opt_str; use rustfs_utils::http::headers::AMZ_REQUEST_ID; +use s3s::S3ErrorCode; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -272,6 +274,68 @@ fn is_empty_body_admin_put_path(path: &str) -> bool { ) } +#[derive(Clone)] +pub struct S3ErrorMessageCompatLayer; + +impl Layer for S3ErrorMessageCompatLayer { + type Service = S3ErrorMessageCompatService; + + fn layer(&self, inner: S) -> Self::Service { + S3ErrorMessageCompatService { inner } + } +} + +#[derive(Clone)] +pub struct S3ErrorMessageCompatService { + inner: S, +} + +impl Service> for S3ErrorMessageCompatService +where + S: Service, Response = Response>> + Clone + Send + 'static, + S::Future: Send + 'static, + S::Error: Send + 'static, + RestBody: Body + From + Send + 'static, + RestBody::Error: Into + Send + 'static, + GrpcBody: Send + 'static, +{ + type Response = Response>; + type Error = S::Error; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, req: HttpRequest) -> Self::Future { + let mut inner = self.inner.clone(); + + Box::pin(async move { + let response = inner.call(req).await?; + let (parts, body) = response.into_parts(); + let should_fix = parts.status == StatusCode::FORBIDDEN && is_xml_response(&parts.headers); + + let response = match body { + HybridBody::Rest { rest_body } => { + if !should_fix { + Response::from_parts(parts, HybridBody::Rest { rest_body }) + } else { + let (rest_body, changed) = fix_s3_error_message_in_xml(rest_body).await.map_err(Into::into)?; + let mut parts = parts; + if changed { + parts.headers.remove(http::header::CONTENT_LENGTH); + } + Response::from_parts(parts, HybridBody::Rest { rest_body }) + } + } + HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }), + }; + + Ok(response) + }) + } +} + #[derive(Clone)] pub struct ObjectAttributesEtagFixLayer; @@ -364,6 +428,30 @@ where Ok(RestBody::from(Bytes::from(fixed))) } +async fn fix_s3_error_message_in_xml(body: RestBody) -> Result<(RestBody, bool), RestBody::Error> +where + RestBody: Body + From, +{ + let bytes = BodyExt::collect(body).await?.to_bytes(); + let xml = String::from_utf8(bytes.to_vec()).unwrap_or_else(|_| String::from_utf8_lossy(&bytes).into_owned()); + let (fixed, changed) = insert_missing_signature_error_message(xml); + Ok((RestBody::from(Bytes::from(fixed)), changed)) +} + +fn insert_missing_signature_error_message(mut xml: String) -> (String, bool) { + if !xml.contains("SignatureDoesNotMatch") || xml.contains("") { + return (xml, false); + } + + let Some(code_end) = xml.find("") else { + return (xml, false); + }; + + let message = ApiError::error_code_to_message(&S3ErrorCode::SignatureDoesNotMatch); + xml.insert_str(code_end + "".len(), &format!("{message}")); + (xml, true) +} + fn strip_quotes_from_first_etag(xml: String) -> String { let Some(start) = xml.find("") else { return xml; @@ -783,6 +871,24 @@ mod tests { ); } + #[test] + fn test_insert_missing_signature_error_message() { + let (fixed, changed) = + insert_missing_signature_error_message("SignatureDoesNotMatch".to_string()); + + assert!(changed); + assert!(fixed.contains("SignatureDoesNotMatchThe request signature we calculated does not match the signature you provided.")); + } + + #[test] + fn test_insert_missing_signature_error_message_preserves_existing_message() { + let input = "SignatureDoesNotMatchcustom".to_string(); + let (fixed, changed) = insert_missing_signature_error_message(input.clone()); + + assert!(!changed); + assert_eq!(fixed, input); + } + #[test] fn test_is_s3_path_excludes_admin_and_special_paths() { assert!(ConditionalCorsLayer::is_s3_path("/my-bucket/key"));