From 68f47b9219132559e97ac3b984cb6c76a66e7fd1 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 2 Sep 2026 16:45:47 +0800 Subject: [PATCH] fix(s3): preserve s3s v0.16 compatibility (#7052) --- crates/e2e_test/src/negative_sigv4_test.rs | 66 ++++++- rustfs/src/admin/router.rs | 19 ++ rustfs/src/error.rs | 209 +++++++++++++-------- rustfs/src/server/http.rs | 4 + 4 files changed, 216 insertions(+), 82 deletions(-) diff --git a/crates/e2e_test/src/negative_sigv4_test.rs b/crates/e2e_test/src/negative_sigv4_test.rs index ca4d1e63c..b88587e27 100644 --- a/crates/e2e_test/src/negative_sigv4_test.rs +++ b/crates/e2e_test/src/negative_sigv4_test.rs @@ -225,8 +225,8 @@ fn encode_unsigned_aws_chunked_with_sha256_trailer(decoded: &[u8]) -> Vec { let checksum = sha256_base64(decoded); let mut encoded = format!("{:x}\r\n", decoded.len()).into_bytes(); encoded.extend_from_slice(decoded); - encoded.extend_from_slice(b"\r\n0\r\n\r\n"); - encoded.extend_from_slice(format!("x-amz-checksum-sha256:{checksum}").as_bytes()); + encoded.extend_from_slice(b"\r\n0\r\n"); + encoded.extend_from_slice(format!("x-amz-checksum-sha256:{checksum}\r\n\r\n").as_bytes()); encoded } @@ -549,6 +549,68 @@ async fn tampered_upload_part_payload_is_rejected() -> Result<(), Box Result<(), Box> { + init_logging(); + let mut env = RustFSTestEnvironment::new().await?; + setup(&mut env).await?; + + for (key, declared_len) in [ + ("decoded-length-overrun.bin", 3_usize), + ("decoded-length-shortfall.bin", 9_usize), + ] { + let decoded = b"decoded"; + assert_ne!(declared_len, decoded.len(), "test case must exercise a mismatch"); + let encoded_body = encode_unsigned_aws_chunked_with_sha256_trailer(decoded); + let decoded_content_length = declared_len.to_string(); + let path = format!("/{BUCKET}/{key}"); + let signer = SigV4::new(&env); + let extra_signed_headers = [ + ("content-encoding", "aws-chunked"), + ("x-amz-decoded-content-length", decoded_content_length.as_str()), + ("x-amz-trailer", "x-amz-checksum-sha256"), + ]; + let headers = signer.sign_with_extra_headers("PUT", &path, "", UNSIGNED_PAYLOAD_TRAILER, &extra_signed_headers); + + let response = local_http_client() + .put(format!("{}{}", env.url, path)) + .header("authorization", &headers.authorization) + .header("content-encoding", "aws-chunked") + .header("x-amz-content-sha256", &headers.content_sha256) + .header("x-amz-date", &headers.amz_date) + .header("x-amz-decoded-content-length", &decoded_content_length) + .header("x-amz-trailer", "x-amz-checksum-sha256") + .body(encoded_body) + .timeout(std::time::Duration::from_secs(10)) + .send() + .await?; + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + assert_eq!( + status, + reqwest::StatusCode::BAD_REQUEST, + "decoded length mismatch must be a client error, body:\n{body}" + ); + assert_error_code(&body, "IncompleteBody"); + + let absent = env + .create_s3_client() + .get_object() + .bucket(BUCKET) + .key(key) + .send() + .await + .expect_err("decoded length mismatch 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")); + } + + env.stop_server(); + Ok(()) +} + /// (e) A request whose `x-amz-date` is skewed beyond the server's tolerance /// (s3s default 900s / 15 min) must be rejected with RequestTimeTooSkewed / /// 403. The signature is otherwise valid: the credential-scope date and diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 661795244..9c69babc5 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -4651,6 +4651,25 @@ mod tests { assert_eq!(listen_root_route, MiscExtRoute::ListenNotification { bucket: None }); } + #[test] + fn s3_route_claims_listen_notification_before_generated_s3_routes() { + let router: S3Router = S3Router::new(false); + let headers = HeaderMap::new(); + + for uri in [ + "/?events=s3:ObjectRemoved:*&x-id=ListenNotification", + "/demo-bucket?events=s3:ObjectCreated:*&x-id=ListenBucketNotification", + "/demo-bucket/?events=s3:ObjectCreated:*&events=s3:ObjectRemoved:Delete", + ] { + let uri: Uri = uri.parse().expect("uri should parse"); + let mut extensions = http::Extensions::new(); + assert!( + router.is_match(&Method::GET, &uri, &headers, &mut extensions), + "listen notification custom route must claim {uri}" + ); + } + } + #[test] fn parse_misc_extension_request_rejects_invalid_paths_or_methods() { let bucket_without_object: Uri = "/demo-bucket?lambdaArn=arn%3Atarget".parse().expect("uri should parse"); diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index c471457c0..d7e7e34b5 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -276,26 +276,49 @@ where false } -fn error_chain_has_upload_stream_sha256_mismatch(err: &(dyn std::error::Error + 'static)) -> bool { - if err.to_string() == "UploadStreamError: Sha256Mismatch" { - return true; +#[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 { + if let Some(classified) = classify_s3s_body_stream_error_display(err) { + return Some(classified); } if let Some(io_err) = err.downcast_ref::() && let Some(inner) = io_err.get_ref() - && error_chain_has_upload_stream_sha256_mismatch(inner) + && let Some(classified) = error_chain_s3s_body_stream_error(inner) { - return true; + return Some(classified); } let mut current = err.source(); while let Some(err) = current { - if err.to_string() == "UploadStreamError: Sha256Mismatch" { - return true; + if let Some(classified) = classify_s3s_body_stream_error_display(err) { + return Some(classified); + } + if let Some(io_err) = err.downcast_ref::() + && let Some(inner) = io_err.get_ref() + && let Some(classified) = error_chain_s3s_body_stream_error(inner) + { + return Some(classified); } current = err.source(); } - false + None } impl From for S3Error { @@ -310,69 +333,60 @@ impl From for S3Error { impl From for ApiError { fn from(err: StorageError) -> Self { - // Preserve typed client-provided digest failures across I/O boundaries. if let StorageError::Io(ref io_err) = err && let Some(inner) = io_err.get_ref() - && (inner.downcast_ref::().is_some() + { + 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() - || error_chain_has_upload_stream_sha256_mismatch(inner)) - { - return ApiError { - code: S3ErrorCode::BadDigest, - message: ApiError::error_code_to_message(&S3ErrorCode::BadDigest), - source: Some(Box::new(err)), - }; - } + || 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 let StorageError::Io(ref io_err) = err - && let Some(inner) = io_err.get_ref() - && 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::EntityTooLarge, + message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge), + source: Some(Box::new(err)), + }; + } - if let StorageError::Io(ref io_err) = err - && let Some(inner) = io_err.get_ref() - && 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::ServiceUnavailable, + message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), + source: Some(Box::new(err)), + }; + } - if let StorageError::Io(ref io_err) = err - && io_err - .get_ref() - .and_then(|inner| inner.downcast_ref::()) - .is_some() - { - return ApiError { - code: S3ErrorCode::ServiceUnavailable, - message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), - 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 let StorageError::Io(ref io_err) = err - && matches!( - io_err - .get_ref() - .and_then(|inner| inner.downcast_ref::()), - Some(rustfs_kms::KmsError::BackendError { .. }) - ) - { - return ApiError { - code: S3ErrorCode::ServiceUnavailable, - message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), - source: Some(Box::new(err)), - }; + if inner.downcast_ref::().is_some() + || matches!( + inner.downcast_ref::(), + Some(rustfs_kms::KmsError::BackendError { .. }) + ) + { + return ApiError { + code: S3ErrorCode::ServiceUnavailable, + message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), + source: Some(Box::new(err)), + }; + } } let code = match &err { @@ -462,10 +476,11 @@ 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) - || error_chain_has_upload_stream_sha256_mismatch(inner) + || matches!(s3s_body_stream_error, Some(S3sBodyStreamError::Sha256Mismatch)) { return ApiError { code: S3ErrorCode::BadDigest, @@ -487,6 +502,13 @@ impl From for ApiError { 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, @@ -568,6 +590,17 @@ mod tests { } } + #[derive(Debug)] + struct MockS3sBodyStreamError(&'static str); + + impl std::fmt::Display for MockS3sBodyStreamError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.0) + } + } + + impl std::error::Error for MockS3sBodyStreamError {} + #[test] fn test_api_error_from_io_error() { let io_error = IoError::new(ErrorKind::PermissionDenied, "permission denied"); @@ -656,27 +689,43 @@ mod tests { } #[test] - fn other_upload_stream_errors_do_not_map_to_bad_digest() { - let errors = [ - MockUploadStreamError::Underlying(IoError::other("underlying body error")), - MockUploadStreamError::LengthMismatch, - MockUploadStreamError::Incomplete, - ]; + fn underlying_upload_stream_errors_remain_internal() { + let api_error = + ApiError::from(IoError::other(MockUploadStreamError::Underlying(IoError::other("underlying body error")))); + assert_eq!(api_error.code, S3ErrorCode::InternalError); - for error in errors { + let api_error = ApiError::from(StorageError::Io(IoError::other(MockUploadStreamError::Underlying(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] { let api_error = ApiError::from(IoError::other(error)); - assert_eq!(api_error.code, S3ErrorCode::InternalError); + assert_eq!(api_error.code, S3ErrorCode::IncompleteBody); + assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); } - let errors = [ - MockUploadStreamError::Underlying(IoError::other("underlying body error")), - MockUploadStreamError::LengthMismatch, - MockUploadStreamError::Incomplete, - ]; - - for error in errors { + for error in [MockUploadStreamError::LengthMismatch, MockUploadStreamError::Incomplete] { let api_error = ApiError::from(StorageError::Io(IoError::other(error))); - assert_eq!(api_error.code, S3ErrorCode::InternalError); + 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)); + + 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)); } } diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 5cf0e9bf9..78ce81e5b 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -155,10 +155,13 @@ static HTTP_STATUS_CLASS_METRICS: std::sync::LazyLock<[HttpStatusClassMetrics; 6 static HTTP_TRANSPORT_FAILURES_COUNTER: std::sync::LazyLock = std::sync::LazyLock::new(|| counter!(METRIC_HTTP_SERVER_FAILURES_TOTAL, LABEL_HTTP_STATUS_CLASS => "transport")); +const RUSTFS_S3_PUT_OBJECT_MAX_SIZE: u64 = 5 * 1024 * 1024 * 1024; + fn rustfs_s3_config() -> S3Config { let mut s3_config = S3Config::default(); s3_config.normalize_forward_slash_path = true; s3_config.enable_sig_v2 = true; + s3_config.put_object_max_size = Some(RUSTFS_S3_PUT_OBJECT_MAX_SIZE); s3_config.sig_v4_allowed_services.push("s3tables".to_string()); s3_config } @@ -3005,6 +3008,7 @@ mod tests { assert!(s3_config.normalize_forward_slash_path); assert!(s3_config.normalize_content_length); assert!(s3_config.enable_sig_v2); + assert_eq!(s3_config.put_object_max_size, Some(RUSTFS_S3_PUT_OBJECT_MAX_SIZE)); assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3")); assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "sts")); assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3tables"));