diff --git a/crates/e2e_test/src/negative_sigv4_test.rs b/crates/e2e_test/src/negative_sigv4_test.rs index d0b52a4e9..75004340e 100644 --- a/crates/e2e_test/src/negative_sigv4_test.rs +++ b/crates/e2e_test/src/negative_sigv4_test.rs @@ -566,47 +566,54 @@ async fn tampered_payload_is_rejected() -> Result<(), Box { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); + let client = env.create_s3_client(); + // Empty, small-eager, zero-copy, and streaming PUTs must share the error contract. + for size in [0, 24, 2 * 1024 * 1024, 17 * 1024 * 1024] { + let actual_body = vec![b'a'; size]; + let wrong_hash = sha256_hex(b"a different payload"); + let content_md5 = base64_simd::STANDARD.encode_to_string(Md5::digest(&actual_body)); + for with_md5 in [false, true] { + let key = format!("tampered-payload-{size}-{with_md5}.txt"); + let path = format!("/{BUCKET}/{key}"); + let extra = if with_md5 { + vec![("content-md5", content_md5.as_str())] + } else { + vec![] + }; + let headers = signer.sign_with_extra_headers("PUT", &path, "", &wrong_hash, &extra); + let mut request = local_http_client() + .put(format!("{}{}", env.url, path)) + .header("x-amz-date", &headers.amz_date) + .header("x-amz-content-sha256", &headers.content_sha256) + .header("authorization", &headers.authorization); + for (name, value) in extra { + request = request.header(name, value); + } + let response = request.body(actual_body.clone()).send().await?; + let status = response.status(); + let body = response.text().await?; + assert_eq!(status, reqwest::StatusCode::BAD_REQUEST, "payload mismatch: {body}"); + assert_error_code(&body, "XAmzContentSHA256Mismatch"); assert!( - status.is_client_error(), - "payload mismatch must be rejected with a client error, got {status}, body:\n{body}" + body.replace("'", "'") + .contains("The provided 'x-amz-content-sha256' header does not match what was computed."), + "wrong message: {body}" ); - info!(%status, "tampered payload rejected with error status"); - } - // A mid-stream hash-mismatch abort surfacing as a transport error is - // also a valid rejection (definitely not a 200 success). - Err(err) => { - assert!(!err.is_connect(), "connection failure is not proof of payload rejection: {err}"); - assert!(!err.is_timeout(), "request timeout is not proof of payload rejection: {err}"); - info!(%err, "tampered payload rejected via mid-stream transport error"); + + let absent = client + .get_object() + .bucket(BUCKET) + .key(&key) + .send() + .await + .expect_err("a tampered payload must not publish an object"); + assert_eq!(absent.raw_response().map(|response| response.status().as_u16()), Some(404)); + assert_eq!(absent.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey")); } } - - let absent = env - .create_s3_client() - .get_object() - .bucket(BUCKET) - .key(key) - .send() - .await - .expect_err("a tampered payload must not publish an object"); - assert_eq!(absent.raw_response().map(|response| response.status().as_u16()), Some(404)); - assert_eq!(absent.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey")); Ok(()) } @@ -638,9 +645,9 @@ async fn tampered_upload_part_payload_is_rejected() -> Result<(), Box, is_dir: boo } fn map_extract_archive_error(err: std::io::Error) -> S3Error { - let message = err.to_string(); let api_error = ApiError::from(err); - if matches!(api_error.code, S3ErrorCode::BadDigest | S3ErrorCode::IncompleteBody) { + if api_error.code != S3ErrorCode::InternalError { return api_error.into(); } - let mut archive_error = s3_error!(InvalidArgument, "Failed to process archive entry: {}", message); + let mut archive_error = s3_error!(InvalidArgument, "Failed to process archive entry: {}", api_error.message); archive_error.set_source(Box::new(api_error)); archive_error } @@ -2239,14 +2238,7 @@ impl DefaultObjectUsecase { } while let Some(entry) = entries.next().await { - let mut f = match entry { - Ok(f) => f, - Err(error) => { - error!(error = %error, "Archive entry read failed"); - flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors).await?; - return Err(s3_error!(InvalidArgument, "Failed to read archive entry: {:?}", error)); - } - }; + let mut f = extract_try!(entry.map_err(map_extract_archive_error)); extracted_entry_count = extracted_entry_count.saturating_add(1); extract_try!(validate_put_object_extract_entry_count(extracted_entry_count, extract_limits,)); let entry_size = f.effective_size(); @@ -3475,6 +3467,123 @@ mod tests { } } + #[test] + fn archive_error_mapping_preserves_classified_body_errors() { + use crate::error::{ClientBodyReadTimeout, ServerSideSourceReadError, UploadLimitExceeded}; + for (error, code) in [ + ( + io::Error::other(ClientBodyReadTimeout { + timeout: Duration::from_secs(1), + raw_bytes_received: 16, + }), + S3ErrorCode::RequestTimeout, + ), + ( + io::Error::other(s3s::BodySizeLimitExceeded { size: 6, limit: 5 }), + S3ErrorCode::EntityTooLarge, + ), + (io::Error::other(UploadLimitExceeded { limit: 5 }), S3ErrorCode::EntityTooLarge), + ( + io::Error::other(ServerSideSourceReadError::new("CopyObject", io::Error::other("source unavailable"))), + S3ErrorCode::ServiceUnavailable, + ), + ] { + let error = map_extract_archive_error(error); + assert_eq!(error.code(), &code); + assert_eq!(error.message(), Some(ApiError::error_code_to_message(&code).as_str())); + assert!(error.source().is_some()); + } + let error = map_extract_archive_error(io::Error::new(io::ErrorKind::InvalidData, "invalid TAR header")); + assert_eq!(error.code(), &S3ErrorCode::InvalidArgument); + assert_eq!(error.message(), Some("Failed to process archive entry: invalid TAR header")); + assert!(error.source().is_some()); + } + + #[test] + fn archive_body_errors_do_not_format_discarded_diagnostics() { + #[derive(Debug)] + struct DisplayProbe { + source: io::Error, + displays: Arc, + } + impl std::fmt::Display for DisplayProbe { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.displays.fetch_add(1, Ordering::Relaxed); + f.write_str("archive transport diagnostics") + } + } + impl std::error::Error for DisplayProbe { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&self.source) + } + } + let displays = Arc::new(AtomicUsize::new(0)); + let error = map_extract_archive_error(io::Error::other(DisplayProbe { + source: io::Error::other(UploadStreamError::Sha256Mismatch), + displays: Arc::clone(&displays), + })); + assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch"); + assert_eq!( + displays.load(Ordering::Relaxed), + 0, + "classified body failures need only their public message" + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn snowball_extract_preserves_sha256_failure_after_sniffing() { + use crate::app::storage_api::s3::UploadStream; + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + use http::Method; + + let (store, context) = real_store_test_context().await; + let bucket = format!("extract-body-error-{}", Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create extract test bucket"); + let mut header = Header::new_gnu(); + header.set_path("member.txt").expect("valid member path"); + header.set_size(0); + header.set_mode(0o644); + header.set_cksum(); + let mut archive = header.as_bytes().to_vec(); + archive.extend_from_slice(&[0; 1024]); + let size = archive.len(); + // Sniffing consumes only the prefix; the final frame fails while TAR + // reads its first header, before a member can be prepared or committed. + let frames = futures::stream::iter([ + Ok::<_, StdError>(Bytes::copy_from_slice(&archive[..16])), + Ok(Bytes::copy_from_slice(&archive[16..])), + ]); + let body = StreamingBlob::wrap(UploadStream::new(frames, size, s3s_sigv4::Sha256Sum::from_bytes([0; 32]))); + let input = PutObjectInput::builder() + .bucket(bucket.clone()) + .key("archive.tar".to_string()) + .content_length(Some(i64::try_from(size).expect("fixture size"))) + .body(Some(body)) + .build() + .expect("extract input"); + let mut request = build_request(input, Method::PUT); + request + .headers + .insert("x-amz-meta-snowball-auto-extract", HeaderValue::from_static("true")); + let error = DefaultObjectUsecase::with_context(Some(context)) + .execute_put_object_extract(request) + .await + .expect_err("a bad payload digest must reject the archive request"); + assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch"); + assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST)); + for key in ["archive.tar", "member.txt"] { + let missing = store + .get_object_info(&bucket, key, &ObjectOptions::default()) + .await + .expect_err("rejected archive cannot publish a member"); + assert!(is_err_object_not_found(&missing), "{missing}"); + } + } + #[tokio::test] async fn archive_etag_reader_validates_sha256_before_completion() { let payload = b"archive-with-wrong-sha256".to_vec(); @@ -3502,6 +3611,9 @@ mod tests { assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); assert!(!state.lock().expect("archive state lock must remain healthy").body_complete); + let error = map_extract_archive_error(err); + assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch"); + assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST)); } fn pax_record(key: &str, value: &[u8]) -> Vec { @@ -3648,6 +3760,9 @@ mod tests { assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); assert!(!state.lock().expect("archive state lock must remain healthy").body_complete); + let error = map_extract_archive_error(err); + assert_eq!(error.code(), &S3ErrorCode::BadDigest); + assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST)); } #[tokio::test] @@ -4777,7 +4892,7 @@ mod tests { ); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn snowball_extract_body_guard_aborts_stalled_upload() { let body = StreamingBlob::wrap(futures::stream::pending::>()); let mut guarded = guard_put_object_body_read_timeout( @@ -4798,6 +4913,9 @@ mod tests { .downcast_ref::() .expect("stall error should retain its I/O kind"); assert_eq!(io_err.kind(), std::io::ErrorKind::TimedOut); + let error = map_extract_archive_error(s3s_body_error_to_io(err)); + assert_eq!(error.code(), &S3ErrorCode::RequestTimeout); + assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST)); } #[test] diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index e88e4f49a..a0c2f1849 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -18,7 +18,7 @@ use super::*; use crate::app::trailer_adapter::trailer_source; use crate::auth::{RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, parse_presigned_put_max_content_length}; -use crate::error::{UploadLimitExceeded, normalize_put_read_quorum_error}; +use crate::error::{ClientBodyReadTimeout, UploadLimitExceeded, normalize_put_read_quorum_error}; static PUT_FAILURE_LOGS: rustfs_utils::LogThrottle = rustfs_utils::LogThrottle::new(5_000); #[cfg(test)] @@ -293,12 +293,10 @@ impl Stream for RequestBodyReadTimeout { ); return Poll::Ready(Some(Err(Box::new(std::io::Error::new( std::io::ErrorKind::TimedOut, - format!( - "request body read stalled: received {} of {} bytes, no data for {}s", - this.received, - expected_display, - this.timeout.as_secs() - ), + ClientBodyReadTimeout { + timeout: this.timeout, + raw_bytes_received: this.received, + }, )) as StdError))); } @@ -2970,7 +2968,7 @@ mod tests { assert!(!object_traffic_health.snapshot().write_stalled); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn put_object_body_read_timeout_guard_aborts_on_stall() { // Inner stream never yields and never reports EOF (a proxy that forwarded // a partial body then went silent while holding the connection open). @@ -2993,6 +2991,9 @@ mod tests { .downcast_ref::() .expect("stall error should wrap an io::Error"); assert_eq!(io_err.kind(), std::io::ErrorKind::TimedOut); + let error = S3Error::from(ApiError::from(s3s_body_error_to_io(err))); + assert_eq!(error.code(), &S3ErrorCode::RequestTimeout); + assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST)); // After a stall the guard terminates the stream instead of re-polling the // abandoned inner stream. @@ -3260,28 +3261,28 @@ mod tests { } #[tokio::test] - async fn read_small_put_body_maps_upload_stream_sha256_mismatch_to_bad_digest() { + async fn read_small_put_body_maps_upload_stream_sha256_mismatch_to_content_sha256_mismatch() { let body = StreamReader::new(futures::stream::iter(vec![Err::(s3s_body_error_to_io(Box::new( - MockUploadStreamSha256Mismatch, + UploadStreamError::Sha256Mismatch, )))])); let error = read_small_put_body_exact_direct(body, 1) .await .expect_err("SHA256 mismatch should reject the small PUT body"); - assert_eq!(error.code(), &S3ErrorCode::BadDigest); + assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch"); } #[tokio::test] - async fn read_zero_copy_put_body_maps_upload_stream_sha256_mismatch_to_bad_digest() { - let body = futures::stream::iter(vec![Err::(MockUploadStreamSha256Mismatch)]); + async fn read_zero_copy_put_body_maps_upload_stream_sha256_mismatch_to_content_sha256_mismatch() { + let body = futures::stream::iter(vec![Err::(UploadStreamError::Sha256Mismatch)]); let error = match read_zero_copy_put_body_exact(body, 1).await { Ok(_) => panic!("SHA256 mismatch should reject the zero-copy PUT body"), Err(error) => error, }; - assert_eq!(error.code(), &S3ErrorCode::BadDigest); + assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch"); } struct FragmentedBody { diff --git a/rustfs/src/app/object/shared.rs b/rustfs/src/app/object/shared.rs index e4d7cc295..97b43d0b4 100644 --- a/rustfs/src/app/object/shared.rs +++ b/rustfs/src/app/object/shared.rs @@ -1783,13 +1783,11 @@ mod tests { #[test] fn s3s_body_error_to_io_preserves_upload_stream_error_source() { - let error = s3s_body_error_to_io(Box::new(MockUploadStreamSha256Mismatch)); + let error = s3s_body_error_to_io(Box::new(UploadStreamError::Sha256Mismatch)); assert!(matches!( - error - .get_ref() - .and_then(|source| source.downcast_ref::()), - Some(MockUploadStreamSha256Mismatch) + error.get_ref().and_then(|source| source.downcast_ref::()), + Some(UploadStreamError::Sha256Mismatch) )); } diff --git a/rustfs/src/app/object/test_support.rs b/rustfs/src/app/object/test_support.rs index 55ebb28b2..48118d871 100644 --- a/rustfs/src/app/object/test_support.rs +++ b/rustfs/src/app/object/test_support.rs @@ -18,16 +18,7 @@ use super::*; use http::{Extensions, HeaderMap, Method, Uri}; use std::sync::Arc; -#[derive(Debug)] -pub(super) struct MockUploadStreamSha256Mismatch; - -impl std::fmt::Display for MockUploadStreamSha256Mismatch { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str("UploadStreamError: Sha256Mismatch") - } -} - -impl std::error::Error for MockUploadStreamSha256Mismatch {} +pub(super) use crate::app::storage_api::s3::UploadStreamError; pub(super) fn build_request(input: T, method: Method) -> S3Request { S3Request { diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index a79b96433..bcc848e99 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -54,6 +54,8 @@ pub(crate) mod s3 { #[cfg(test)] pub(crate) use s3s::service::{S3Service, S3ServiceBuilder}; #[cfg(test)] + pub(crate) use s3s::stream::upload_stream::{UploadStream, UploadStreamError}; + #[cfg(test)] pub(crate) use s3s::xml::{Serialize as XmlSerialize, Serializer as XmlSerializer}; #[cfg(test)] pub(crate) use s3s::{Body, S3, S3Response}; diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 57afc8208..f726826fb 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -17,8 +17,12 @@ use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError, unr use http::StatusCode; use rustfs_kms::KmsUnavailableError; use rustfs_lock::LockError; +use s3s::stream::{aws_chunked_stream::AwsChunkedStreamError, upload_stream::UploadStreamError}; use s3s::{S3Error, S3ErrorCode}; +const CONTENT_SHA256_MISMATCH_CODE: &str = "XAmzContentSHA256Mismatch"; +const CONTENT_SHA256_MISMATCH_MESSAGE: &str = "The provided 'x-amz-content-sha256' header does not match what was computed."; + const MAX_VERSIONS_EXCEEDED_CODE: &str = "MaxVersionsExceeded"; const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the number of versions you can create on this object"; const SLOW_DOWN_READ_CODE: &str = "SlowDownRead"; @@ -164,7 +168,12 @@ fn data_plane_kms_error(error: &rustfs_kms::KmsError) -> Option<(S3ErrorCode, St /// exist" has to say so itself. fn custom_error_status(code: &S3ErrorCode) -> Option { match code { - S3ErrorCode::Custom(custom) if &**custom == KMS_KEY_NOT_FOUND_ERROR_CODE || &**custom == MAX_VERSIONS_EXCEEDED_CODE => { + S3ErrorCode::Custom(custom) + if matches!( + &**custom, + KMS_KEY_NOT_FOUND_ERROR_CODE | MAX_VERSIONS_EXCEEDED_CODE | CONTENT_SHA256_MISMATCH_CODE + ) => + { Some(StatusCode::BAD_REQUEST) } S3ErrorCode::Custom(custom) if &**custom == SLOW_DOWN_READ_CODE => Some(StatusCode::SERVICE_UNAVAILABLE), @@ -398,6 +407,9 @@ impl ApiError { S3ErrorCode::InvalidArgument => "Invalid argument".to_string(), S3ErrorCode::InvalidStorageClass => "Invalid storage class.".to_string(), S3ErrorCode::AccessDenied => "Access Denied.".to_string(), + S3ErrorCode::Custom(code) if &**code == CONTENT_SHA256_MISMATCH_CODE => { + CONTENT_SHA256_MISMATCH_MESSAGE.to_string() + } S3ErrorCode::BadDigest => "The Content-Md5 you specified did not match what we received.".to_string(), S3ErrorCode::EntityTooSmall => "Your proposed upload is smaller than the minimum allowed object size.".to_string(), S3ErrorCode::EntityTooLarge => "Your proposed upload exceeds the maximum allowed object size.".to_string(), @@ -534,32 +546,55 @@ impl ApiError { } } -fn error_chain_has_type(err: &(dyn std::error::Error + 'static)) -> bool -where - T: std::error::Error + 'static, -{ - error_chain_find(err, |error| error.is::().then_some(())).is_some() -} +/// Classify body failures in one bounded pass, retaining precedence across wrappers. +/// A digest failure wins immediately; other markers must wait for deeper causes. +fn body_read_error_code(err: &(dyn std::error::Error + 'static)) -> Option { + let mut upload_limit_exceeded = false; + let mut server_source_failed = false; + let mut client_body_timed_out = false; + let mut body_size_exceeded = false; + let mut incomplete_body = false; + let digest = error_chain_find(err, |error| { + if error.is::() + || matches!(error.downcast_ref::(), Some(UploadStreamError::Sha256Mismatch)) + { + return Some(S3ErrorCode::Custom(CONTENT_SHA256_MISMATCH_CODE.into())); + } + if error.is::() || error.is::() { + return Some(S3ErrorCode::BadDigest); + } -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -enum S3sBodyStreamError { - Sha256Mismatch, - IncompleteBody, -} - -fn classify_s3s_body_stream_error_display(err: &(dyn std::error::Error + 'static)) -> Option { - match err.to_string().as_str() { - "UploadStreamError: Sha256Mismatch" => Some(S3sBodyStreamError::Sha256Mismatch), - "UploadStreamError: LengthMismatch" - | "UploadStreamError: Incomplete" - | "AwsChunkedStreamError: LengthMismatch" - | "AwsChunkedStreamError: Incomplete" => Some(S3sBodyStreamError::IncompleteBody), - _ => None, - } -} - -fn error_chain_s3s_body_stream_error(err: &(dyn std::error::Error + 'static)) -> Option { - error_chain_find(err, classify_s3s_body_stream_error_display) + upload_limit_exceeded |= error.is::(); + server_source_failed |= error.is::(); + client_body_timed_out |= error.is::(); + body_size_exceeded |= error.is::(); + incomplete_body |= error.is::() + || matches!( + error.downcast_ref::(), + Some(UploadStreamError::LengthMismatch | UploadStreamError::Incomplete) + ) + || matches!( + error.downcast_ref::(), + Some(AwsChunkedStreamError::LengthMismatch | AwsChunkedStreamError::Incomplete) + ) + || is_hyper_body_eof(error); + None + }); + digest.or_else(|| { + if upload_limit_exceeded { + Some(S3ErrorCode::EntityTooLarge) + } else if server_source_failed { + Some(S3ErrorCode::ServiceUnavailable) + } else if client_body_timed_out { + Some(S3ErrorCode::RequestTimeout) + } else if body_size_exceeded { + Some(S3ErrorCode::EntityTooLarge) + } else if incomplete_body { + Some(S3ErrorCode::IncompleteBody) + } else { + None + } + }) } /// `io::Error::source` skips its custom payload itself. Visit that payload @@ -584,10 +619,6 @@ fn error_chain_find( None } -fn error_chain_has_body_size_limit_exceeded(err: &(dyn std::error::Error + 'static)) -> bool { - error_chain_has_type::(err) -} - /// hyper reports a request body whose connection hit EOF before /// `Content-Length` bytes arrived as a `Kind::Body` error carrying an /// `UnexpectedEof` `io::Error` (its `IncompleteBody` marker is private). @@ -599,10 +630,6 @@ fn is_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool { .is_some_and(|io_err| io_err.kind() == std::io::ErrorKind::UnexpectedEof) } -fn error_chain_has_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool { - error_chain_find(err, |error| is_hyper_body_eof(error).then_some(())).is_some() -} - impl From for S3Error { fn from(err: ApiError) -> Self { let status = custom_error_status(&err.code); @@ -674,65 +701,10 @@ impl From for ApiError { if let StorageError::Io(ref io_err) = err && let Some(inner) = io_err.get_ref() { - let s3s_body_stream_error = error_chain_s3s_body_stream_error(inner); - - // Preserve client-provided body and digest failures across I/O boundaries. - if inner.downcast_ref::().is_some() - || inner.downcast_ref::().is_some() - || inner.downcast_ref::().is_some() - || matches!(s3s_body_stream_error, Some(S3sBodyStreamError::Sha256Mismatch)) - { + if let Some(code) = body_read_error_code(inner) { return ApiError { - code: S3ErrorCode::BadDigest, - message: ApiError::error_code_to_message(&S3ErrorCode::BadDigest), - source: Some(Box::new(err)), - }; - } - - if error_chain_has_type::(inner) { - return ApiError { - code: S3ErrorCode::EntityTooLarge, - message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge), - source: Some(Box::new(err)), - }; - } - - if error_chain_has_type::(inner) { - return ApiError { - code: S3ErrorCode::ServiceUnavailable, - message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), - source: Some(Box::new(err)), - }; - } - - if error_chain_has_type::(inner) { - return ApiError { - code: S3ErrorCode::RequestTimeout, - message: ApiError::error_code_to_message(&S3ErrorCode::RequestTimeout), - source: Some(Box::new(err)), - }; - } - - if error_chain_has_body_size_limit_exceeded(inner) { - return ApiError { - code: S3ErrorCode::EntityTooLarge, - message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge), - source: Some(Box::new(err)), - }; - } - - if error_chain_has_hyper_body_eof(inner) { - return ApiError { - code: S3ErrorCode::IncompleteBody, - message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody), - source: Some(Box::new(err)), - }; - } - - if matches!(s3s_body_stream_error, Some(S3sBodyStreamError::IncompleteBody)) { - return ApiError { - code: S3ErrorCode::IncompleteBody, - message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody), + message: ApiError::error_code_to_message(&code), + code, source: Some(Box::new(err)), }; } @@ -852,72 +824,14 @@ impl From for ApiError { impl From for ApiError { fn from(err: std::io::Error) -> Self { - // Map client-provided digest mismatches to BadDigest. - if let Some(inner) = err.get_ref() { - let s3s_body_stream_error = error_chain_s3s_body_stream_error(inner); - if error_chain_has_type::(inner) - || error_chain_has_type::(inner) - || error_chain_has_type::(inner) - || matches!(s3s_body_stream_error, Some(S3sBodyStreamError::Sha256Mismatch)) - { - return ApiError { - code: S3ErrorCode::BadDigest, - message: ApiError::error_code_to_message(&S3ErrorCode::BadDigest), - source: Some(Box::new(err)), - }; - } - if error_chain_has_type::(inner) { - return ApiError { - code: S3ErrorCode::EntityTooLarge, - message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge), - source: Some(Box::new(err)), - }; - } - if error_chain_has_type::(inner) { - return ApiError { - code: S3ErrorCode::ServiceUnavailable, - message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), - source: Some(Box::new(err)), - }; - } - if error_chain_has_type::(inner) { - return ApiError { - code: S3ErrorCode::RequestTimeout, - message: ApiError::error_code_to_message(&S3ErrorCode::RequestTimeout), - source: Some(Box::new(err)), - }; - } - - if error_chain_has_body_size_limit_exceeded(inner) { - return ApiError { - code: S3ErrorCode::EntityTooLarge, - message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge), - source: Some(Box::new(err)), - }; - } - - if error_chain_has_hyper_body_eof(inner) { - return ApiError { - code: S3ErrorCode::IncompleteBody, - message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody), - source: Some(Box::new(err)), - }; - } - - if matches!(s3s_body_stream_error, Some(S3sBodyStreamError::IncompleteBody)) { - return ApiError { - code: S3ErrorCode::IncompleteBody, - message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody), - source: Some(Box::new(err)), - }; - } - if error_chain_has_type::(inner) { - return ApiError { - code: S3ErrorCode::IncompleteBody, - message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody), - source: Some(Box::new(err)), - }; - } + if let Some(inner) = err.get_ref() + && let Some(code) = body_read_error_code(inner) + { + return ApiError { + message: ApiError::error_code_to_message(&code), + code, + source: Some(Box::new(err)), + }; } ApiError { code: S3ErrorCode::InternalError, @@ -965,9 +879,17 @@ mod tests { use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard; use rustfs_s3_ops::S3Operation; + use s3s::stream::{aws_chunked_stream::AwsChunkedStreamError, upload_stream::UploadStreamError}; use s3s::{S3Error, S3ErrorCode}; use std::io::{Error as IoError, ErrorKind}; + fn error_chain_has_type(err: &(dyn std::error::Error + 'static)) -> bool + where + T: std::error::Error + 'static, + { + error_chain_find(err, |error| error.is::().then_some(())).is_some() + } + #[tokio::test] async fn lock_timeout_response_advises_retry_without_exposing_resource() { use http_body_util::BodyExt as _; @@ -1238,34 +1160,6 @@ mod tests { assert!(!format!("{:?}", error.diagnostic()).contains("secret")); } - #[derive(Debug)] - enum MockUploadStreamError { - Underlying(IoError), - Sha256Mismatch, - LengthMismatch, - Incomplete, - } - - impl std::fmt::Display for MockUploadStreamError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Underlying(err) => write!(f, "UploadStreamError: Underlying: {err}"), - Self::Sha256Mismatch => f.write_str("UploadStreamError: Sha256Mismatch"), - Self::LengthMismatch => f.write_str("UploadStreamError: LengthMismatch"), - Self::Incomplete => f.write_str("UploadStreamError: Incomplete"), - } - } - } - - impl std::error::Error for MockUploadStreamError { - fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { - match self { - Self::Underlying(err) => Some(err), - Self::Sha256Mismatch | Self::LengthMismatch | Self::Incomplete => None, - } - } - } - #[derive(Debug)] struct MockS3sBodyStreamError(&'static str); @@ -1314,94 +1208,124 @@ mod tests { } #[test] - fn sha256_mismatch_io_errors_map_to_bad_digest() { - let io_error = IoError::new( - ErrorKind::InvalidData, - rustfs_rio::Sha256Mismatch { - expected_sha256: "expected".to_string(), - calculated_sha256: "calculated".to_string(), - }, - ); - let api_error = ApiError::from(io_error); - assert_eq!(api_error.code, S3ErrorCode::BadDigest); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest)); - - let storage_error = StorageError::Io(IoError::new( - ErrorKind::InvalidData, - rustfs_rio::Sha256Mismatch { - expected_sha256: "expected".to_string(), - calculated_sha256: "calculated".to_string(), - }, - )); - let api_error = ApiError::from(storage_error); - assert_eq!(api_error.code, S3ErrorCode::BadDigest); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest)); + fn payload_sha256_mismatch_io_errors_map_to_content_sha256_mismatch() { + for storage in [false, true] { + for nested in [false, true] { + for error in [ + IoError::other(rustfs_rio::Sha256Mismatch { + expected_sha256: "expected".to_string(), + calculated_sha256: "calculated".to_string(), + }), + IoError::other(UploadStreamError::Sha256Mismatch), + ] { + let error = if nested { IoError::other(error) } else { error }; + let api_error = if storage { + ApiError::from(StorageError::Io(error)) + } else { + ApiError::from(error) + }; + assert_eq!(api_error.code.as_str(), "XAmzContentSHA256Mismatch"); + assert_eq!( + api_error.message, + "The provided 'x-amz-content-sha256' header does not match what was computed." + ); + assert!( + error_chain_has_type::(&api_error) + || error_chain_has_type::(&api_error) + ); + let s3_error = S3Error::from(api_error); + assert_eq!(s3_error.status_code(), Some(StatusCode::BAD_REQUEST)); + assert!(s3_error.source().is_some()); + } + } + } } #[test] - fn content_md5_mismatch_io_errors_map_to_bad_digest() { - let bad_digest = || rustfs_rio::BadDigest { - expected_md5: "expected".to_string(), - calculated_md5: "calculated".to_string(), - }; - let api_error = ApiError::from(IoError::new(ErrorKind::InvalidData, bad_digest())); - assert_eq!(api_error.code, S3ErrorCode::BadDigest); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest)); - - let api_error = ApiError::from(StorageError::Io(IoError::new(ErrorKind::InvalidData, bad_digest()))); - assert_eq!(api_error.code, S3ErrorCode::BadDigest); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest)); - } - - #[test] - fn upload_stream_sha256_mismatch_maps_to_bad_digest() { - let api_error = ApiError::from(IoError::other(MockUploadStreamError::Sha256Mismatch)); - assert_eq!(api_error.code, S3ErrorCode::BadDigest); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest)); - - let api_error = ApiError::from(StorageError::Io(IoError::other(MockUploadStreamError::Sha256Mismatch))); - assert_eq!(api_error.code, S3ErrorCode::BadDigest); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest)); + fn content_md5_and_additional_checksum_mismatches_map_to_bad_digest() { + for storage in [false, true] { + for nested in [false, true] { + for error in [ + IoError::other(rustfs_rio::BadDigest { + expected_md5: "expected".to_string(), + calculated_md5: "calculated".to_string(), + }), + IoError::other(rustfs_rio::ChecksumMismatch { + want: "expected".into(), + got: "calculated".into(), + }), + ] { + let error = if nested { IoError::other(error) } else { error }; + let api_error = if storage { + ApiError::from(StorageError::Io(error)) + } else { + ApiError::from(error) + }; + assert_eq!(api_error.code, S3ErrorCode::BadDigest); + assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest)); + assert_eq!(S3Error::from(api_error).status_code(), Some(StatusCode::BAD_REQUEST)); + } + } + } } #[test] fn underlying_upload_stream_errors_remain_internal() { - let api_error = - ApiError::from(IoError::other(MockUploadStreamError::Underlying(IoError::other("underlying body error")))); + let api_error = ApiError::from(IoError::other(UploadStreamError::Underlying(Box::new(IoError::other( + "underlying body error", + ))))); assert_eq!(api_error.code, S3ErrorCode::InternalError); - let api_error = ApiError::from(StorageError::Io(IoError::other(MockUploadStreamError::Underlying(IoError::other( - "underlying body error", + let api_error = ApiError::from(StorageError::Io(IoError::other(UploadStreamError::Underlying(Box::new( + IoError::other("underlying body error"), ))))); assert_eq!(api_error.code, S3ErrorCode::InternalError); } #[test] fn s3s_incomplete_body_stream_errors_map_to_incomplete_body() { - for error in [MockUploadStreamError::LengthMismatch, MockUploadStreamError::Incomplete] { + for error in [UploadStreamError::LengthMismatch, UploadStreamError::Incomplete] { let api_error = ApiError::from(IoError::other(error)); assert_eq!(api_error.code, S3ErrorCode::IncompleteBody); assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); } - for error in [MockUploadStreamError::LengthMismatch, MockUploadStreamError::Incomplete] { + for error in [UploadStreamError::LengthMismatch, UploadStreamError::Incomplete] { let api_error = ApiError::from(StorageError::Io(IoError::other(error))); assert_eq!(api_error.code, S3ErrorCode::IncompleteBody); assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); } - for message in ["AwsChunkedStreamError: LengthMismatch", "AwsChunkedStreamError: Incomplete"] { - let api_error = ApiError::from(IoError::other(MockS3sBodyStreamError(message))); - assert_eq!(api_error.code, S3ErrorCode::IncompleteBody, "{message}"); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); + for storage in [false, true] { + for nested in [false, true] { + for error in [AwsChunkedStreamError::LengthMismatch, AwsChunkedStreamError::Incomplete] { + let error = IoError::other(error); + let error = if nested { IoError::other(error) } else { error }; + let api_error = if storage { + ApiError::from(StorageError::Io(error)) + } else { + ApiError::from(error) + }; + assert_eq!(api_error.code, S3ErrorCode::IncompleteBody); + assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); + } + } + } + } - let api_error = ApiError::from(IoError::other(IoError::other(MockS3sBodyStreamError(message)))); - assert_eq!(api_error.code, S3ErrorCode::IncompleteBody, "{message}"); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); - - let api_error = ApiError::from(StorageError::Io(IoError::other(MockS3sBodyStreamError(message)))); - assert_eq!(api_error.code, S3ErrorCode::IncompleteBody, "{message}"); - assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); + #[test] + fn body_error_display_text_does_not_impersonate_typed_failures() { + for message in [ + "UploadStreamError: Sha256Mismatch", + "UploadStreamError: Incomplete", + "AwsChunkedStreamError: LengthMismatch", + ] { + for error in [ + ApiError::from(IoError::other(MockS3sBodyStreamError(message))), + ApiError::from(StorageError::Io(IoError::other(MockS3sBodyStreamError(message)))), + ] { + assert_eq!(error.code, S3ErrorCode::InternalError); + } } } @@ -1477,8 +1401,12 @@ mod tests { #[test] fn body_error_classification_bounds_cyclic_source_chains() { + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; #[derive(Debug)] - struct Cycle; + struct Cycle(Arc); impl std::fmt::Display for Cycle { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.write_str("cycle") @@ -1486,11 +1414,70 @@ mod tests { } impl std::error::Error for Cycle { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.0.fetch_add(1, Ordering::Relaxed); Some(self) } } - assert!(!error_chain_has_type::(&Cycle)); - assert_eq!(ApiError::from(IoError::other(Cycle)).code, S3ErrorCode::InternalError); + let visits = Arc::new(AtomicUsize::new(0)); + let error = ApiError::from(IoError::other(Cycle(Arc::clone(&visits)))); + assert_eq!(error.code, S3ErrorCode::InternalError); + assert!(visits.load(Ordering::Relaxed) <= 64, "body classification must share one chain budget"); + } + + #[test] + fn body_error_classification_preserves_nested_marker_precedence() { + for storage in [false, true] { + for (inner, expected) in [ + (IoError::other(UploadStreamError::Sha256Mismatch), CONTENT_SHA256_MISMATCH_CODE), + ( + IoError::other(rustfs_rio::BadDigest { + expected_md5: "expected".into(), + calculated_md5: "calculated".into(), + }), + "BadDigest", + ), + (IoError::other(UploadLimitExceeded { limit: 5 }), "EntityTooLarge"), + ( + IoError::other(ClientBodyReadTimeout { + timeout: std::time::Duration::from_secs(300), + raw_bytes_received: 0, + }), + "ServiceUnavailable", + ), + (IoError::other(s3s::BodySizeLimitExceeded { size: 6, limit: 5 }), "ServiceUnavailable"), + ] { + let error = IoError::other(ServerSideSourceReadError::new("CopyObject", inner)); + let error = if storage { + ApiError::from(StorageError::Io(error)) + } else { + ApiError::from(error) + }; + assert_eq!(error.code.as_str(), expected); + assert!(error.source.is_some()); + } + } + } + + #[tokio::test] + async fn short_hash_reader_body_maps_to_incomplete_body_across_io_boundaries() { + use tokio::io::AsyncReadExt; + for storage in [false, true] { + let mut reader = + rustfs_rio::HashReader::from_stream(std::io::Cursor::new(b"short".to_vec()), 10, 10, None, None, false) + .expect("valid declared body length"); + let error = reader + .read_to_end(&mut Vec::new()) + .await + .expect_err("five bytes cannot satisfy declared ten"); + let error = if storage { + ApiError::from(StorageError::Io(error)) + } else { + ApiError::from(error) + }; + assert_eq!(error.code, S3ErrorCode::IncompleteBody); + assert!(error_chain_has_type::(&error)); + assert_eq!(S3Error::from(error).status_code(), Some(StatusCode::BAD_REQUEST)); + } } /// Exercise the public error type through the actual body budget. diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index 3cb2e031d..9378b7fd7 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -1065,7 +1065,7 @@ pub fn parse_copy_source_range(range_str: &str) -> S3Result { /// digest, for both header-signed and presigned SigV4 requests, and rejects a /// malformed header outright. Every request that reaches a handler with that /// header has therefore already had its body verified (or is about to fail in -/// the body stream with the same `BadDigest` mapping), so hashing it again in +/// the body stream with `XAmzContentSHA256Mismatch`), so hashing it again in /// `HashReader` would cost a second full SHA-256 pass per PUT for no extra /// protection. Streaming payloads are verified chunk by chunk by s3s and never /// produced a digest here. diff --git a/scripts/check_s3s_footprint.sh b/scripts/check_s3s_footprint.sh index 98ed78f30..6413d65e5 100755 --- a/scripts/check_s3s_footprint.sh +++ b/scripts/check_s3s_footprint.sh @@ -65,7 +65,7 @@ cd "$(dirname "$0")/.." # other reductions. The merged tree measures 209 import files and 1586 # invocation lines, tightening main's reviewed 1592-line baseline by six. S3S_IMPORT_FILES_BASELINE=209 -S3_ERROR_LINES_BASELINE=1586 +S3_ERROR_LINES_BASELINE=1572 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming* # client was extracted to crates/s3-client, where s3s usage is legitimate;