diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index b2366a264..81e78e99e 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -1129,7 +1129,7 @@ impl DefaultMultipartUsecase { #[instrument(level = "debug", skip(self, req))] #[hotpath::measure(impl_type = "MultipartUsecase")] - pub async fn execute_upload_part(&self, req: S3Request) -> S3Result> { + pub async fn execute_upload_part(&self, mut req: S3Request) -> S3Result> { reject_presigned_multipart_max_total_object_size_for_other_operation( &req.headers, req.uri.query(), @@ -1140,6 +1140,7 @@ impl DefaultMultipartUsecase { req.uri.query(), req.extensions.get::().is_some(), )?; + normalize_presigned_part_checksums(&mut req)?; let mut opts = ObjectOptions::default(); apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?; let input = req.input; @@ -1932,6 +1933,53 @@ fn passthrough_part_actual_size(headers: &HeaderMap) -> Option { .filter(|size| *size > 0) } +// Hoisted values are signed query parameters, not HTTP headers. Normalize only +// after access control has verified the presigned request. +fn normalize_presigned_part_checksums(req: &mut S3Request) -> S3Result<()> { + if req.extensions.get::().is_none() { + return Ok(()); + } + let Some(query) = req.uri.query() else { + return Ok(()); + }; + let mut seen = std::collections::HashSet::new(); + for (name, value) in url::form_urlencoded::parse(query.as_bytes()) { + let field = match name.as_ref() { + "x-amz-checksum-crc32" => Some(&mut req.input.checksum_crc32), + "x-amz-checksum-crc32c" => Some(&mut req.input.checksum_crc32c), + "x-amz-checksum-crc64nvme" => Some(&mut req.input.checksum_crc64nvme), + "x-amz-checksum-sha1" => Some(&mut req.input.checksum_sha1), + "x-amz-checksum-sha256" => Some(&mut req.input.checksum_sha256), + "x-amz-sdk-checksum-algorithm" => None, + _ => continue, + }; + if !seen.insert(name.clone()) { + return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "Duplicate checksum query parameter")); + } + if let Some(header) = req.headers.get(name.as_ref()) + && header.as_bytes() != value.as_bytes() + { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + "Conflicting checksum header and query parameter", + )); + } + let header = http::HeaderValue::from_str(&value) + .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid checksum query parameter"))?; + req.headers.insert( + http::header::HeaderName::from_bytes(name.as_bytes()) + .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid checksum query parameter"))?, + header, + ); + if let Some(field) = field { + *field = Some(value.into_owned()); + } else { + req.input.checksum_algorithm = Some(ChecksumAlgorithm::from(value.into_owned())); + } + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -1947,6 +1995,102 @@ mod tests { use temp_env::async_with_vars; use tokio::io::AsyncReadExt; + fn presigned_checksum_request(query: &str) -> S3Request { + let mut req = build_request(UploadPartInput::default(), Method::PUT); + req.uri = format!("/bucket/object?{query}").parse().unwrap(); + req.extensions.insert(VerifiedPresignedRequest); + req + } + + #[tokio::test] + async fn normalize_presigned_part_checksums_validates_body() { + // SHA256("abc"), including percent-encoded base64 punctuation. + let checksum = "ungWv48Bz+pBQUDeXa4iI7ADYaOWF3qctBD/YfIAFa0="; + let query = + "x-amz-checksum-sha256=ungWv48Bz%2BpBQUDeXa4iI7ADYaOWF3qctBD%2FYfIAFa0%3D&x-amz-sdk-checksum-algorithm=SHA256"; + for payload in [b"abc", b"abd"] { + let mut req = presigned_checksum_request(query); + normalize_presigned_part_checksums(&mut req).unwrap(); + assert_eq!(req.input.checksum_sha256.as_deref(), Some(checksum)); + assert_eq!( + req.input.checksum_algorithm.as_ref().map(|algorithm| algorithm.as_str()), + Some(ChecksumAlgorithm::SHA256) + ); + let mut reader = HashReader::from_stream(Cursor::new(payload), 3, 3, None, None, false).unwrap(); + reader.add_checksum_from_s3s(&req.headers, None, false).unwrap(); + assert_eq!(reader.content_crc_type(), Some(rustfs_rio::ChecksumType::SHA256)); + let mut bytes = Vec::new(); + let result = reader.read_to_end(&mut bytes).await; + if payload == b"abc" { + result.unwrap(); + assert_eq!(bytes, payload); + assert_eq!(reader.content_crc().get("SHA256").map(String::as_str), Some(checksum)); + } else { + let err = S3Error::from(ApiError::from(result.unwrap_err())); + assert_eq!(*err.code(), S3ErrorCode::BadDigest); + } + } + } + + #[tokio::test] + async fn normalize_presigned_part_checksums_accepts_sdk_crc32_default() { + let mut req = presigned_checksum_request("x-amz-checksum-crc32=y%2FQ5Jg%3D%3D&x-amz-sdk-checksum-algorithm=CRC32"); + normalize_presigned_part_checksums(&mut req).unwrap(); + assert_eq!(req.input.checksum_crc32.as_deref(), Some("y/Q5Jg==")); + let mut reader = HashReader::from_stream(Cursor::new(b"123456789"), 9, 9, None, None, false).unwrap(); + reader.add_checksum_from_s3s(&req.headers, None, false).unwrap(); + let mut bytes = Vec::new(); + reader.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"123456789"); + assert_eq!(reader.content_crc().get("CRC32").map(String::as_str), Some("y/Q5Jg==")); + } + + #[test] + fn normalize_presigned_part_checksums_rejects_ambiguous_values() { + for query in [ + "x-amz-checksum-sha256=one&x-amz-checksum-sha256=two", + "x-amz-sdk-checksum-algorithm=SHA256&x-amz-sdk-checksum-algorithm=CRC32", + ] { + let mut req = presigned_checksum_request(query); + assert_eq!( + *normalize_presigned_part_checksums(&mut req).unwrap_err().code(), + S3ErrorCode::InvalidRequest + ); + } + let mut req = presigned_checksum_request("x-amz-checksum-sha256=one"); + req.headers.insert("x-amz-checksum-sha256", HeaderValue::from_static("two")); + assert_eq!( + *normalize_presigned_part_checksums(&mut req).unwrap_err().code(), + S3ErrorCode::InvalidRequest + ); + } + + #[test] + fn normalize_presigned_part_checksums_preserves_headers_and_rejects_invalid_values() { + let mut req = presigned_checksum_request("partNumber=1&uploadId=test"); + req.input.checksum_sha256 = Some("existing".to_owned()); + req.headers + .insert("x-amz-checksum-sha256", HeaderValue::from_static("existing")); + normalize_presigned_part_checksums(&mut req).unwrap(); + assert_eq!(req.input.checksum_sha256.as_deref(), Some("existing")); + assert_eq!(req.headers["x-amz-checksum-sha256"], "existing"); + + let mut req = presigned_checksum_request("x-amz-checksum-sha256=%0D%0A"); + assert_eq!( + *normalize_presigned_part_checksums(&mut req).unwrap_err().code(), + S3ErrorCode::InvalidArgument + ); + } + + #[test] + fn normalize_presigned_part_checksums_requires_verified_presign() { + let mut req = presigned_checksum_request("x-amz-checksum-sha256=one"); + req.extensions.remove::(); + normalize_presigned_part_checksums(&mut req).unwrap(); + assert!(req.input.checksum_sha256.is_none()); + assert!(req.headers.is_empty()); + } + fn upload_metadata_with_checksum_type(recorded: &str) -> HashMap { HashMap::from([(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE.to_string(), recorded.to_string())]) } diff --git a/rustfs/src/app/multipart_usecase/tests/body_read_tests.rs b/rustfs/src/app/multipart_usecase/tests/body_read_tests.rs index e0cdb37d8..6826bab7a 100644 --- a/rustfs/src/app/multipart_usecase/tests/body_read_tests.rs +++ b/rustfs/src/app/multipart_usecase/tests/body_read_tests.rs @@ -313,3 +313,72 @@ async fn assert_foreground_queue() { .expect("upload task") .expect("queue time is not client inactivity"); } + +#[test] +#[serial_test::serial] +fn presigned_part_checksum_survives_completion_and_rejects_bad_retry() { + crate::app::gating_test_env::run_large_stack_test("presigned-part-checksum", || async { + let (_, store) = crate::app::gating_test_env::shared_gating_ecstore_and_disk_paths().await; + let ambient = crate::app::gating_test_env::shared_gating_ambient().await; + let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms())); + let usecase = + DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), Arc::new(ConcurrencyManager::default())); + let bucket = format!("presigned-checksum-{}", Uuid::new_v4().simple()); + store.make_bucket(&bucket, &MakeBucketOptions::default()).await.unwrap(); + let checksum = "ungWv48Bz+pBQUDeXa4iI7ADYaOWF3qctBD/YfIAFa0="; + for compressed in [false, true] { + let mut options = ObjectOptions { + want_checksum: Some(rustfs_rio::Checksum { + checksum_type: rustfs_rio::ChecksumType::SHA256, + ..Default::default() + }), + ..Default::default() + }; + if compressed { + insert_str( + &mut options.user_defined, + rustfs_utils::http::SUFFIX_COMPRESSION, + compression_metadata_value(CompressionAlgorithm::default()), + ); + } + let upload = store.new_multipart_upload(&bucket, "object", &options).await.unwrap(); + let mut etag = None; + for payload in [b"abc", b"abd"] { + let mut req = part_request(&bucket, &upload.upload_id, StreamingBlob::from(Bytes::copy_from_slice(payload)), 3); + req.uri = "/bucket/object?x-amz-checksum-sha256=ungWv48Bz%2BpBQUDeXa4iI7ADYaOWF3qctBD%2FYfIAFa0%3D&x-amz-sdk-checksum-algorithm=SHA256".parse().unwrap(); + req.extensions.insert(VerifiedPresignedRequest); + let result = usecase.execute_upload_part(req).await; + if payload == b"abc" { + let response = result.expect("hoisted SHA256 part must be accepted"); + assert_eq!(response.output.checksum_sha256.as_deref(), Some(checksum)); + etag = response.output.e_tag.map(|etag| etag.value().to_owned()); + } else { + assert_eq!(*result.unwrap_err().code(), S3ErrorCode::BadDigest); + } + } + store + .clone() + .complete_multipart_upload( + &bucket, + "object", + &upload.upload_id, + vec![CompletePart { + part_num: 1, + etag, + checksum_sha256: Some(checksum.to_owned()), + ..Default::default() + }], + &ObjectOptions::default(), + ) + .await + .expect("stored checksum must survive failed replacement and complete"); + let mut object = store + .get_object_reader(&bucket, "object", None, HeaderMap::new(), &ObjectOptions::default()) + .await + .unwrap(); + let mut bytes = Vec::new(); + object.stream.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"abc"); + } + }); +}