gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 03:59:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(s3): preserve s3s v0.16 compatibility (#7052)

Browse Source
This commit is contained in:
houseme
2026-09-02 16:45:47 +08:00
committed by GitHub
parent de7422b508
commit 68f47b9219
4 changed files with 216 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/negative_sigv4_test.rs
+64 -2
View File
@@ -225,8 +225,8 @@ fn encode_unsigned_aws_chunked_with_sha256_trailer(decoded: &[u8]) -> Vec<u8> {
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<dyn std::e
Ok(())
}
/// s3s v0.16 validates the aws-chunked decoded length while RustFS consumes the
/// body stream. Mismatches are client body errors and must not leak as 500s.
#[tokio::test]
async fn aws_chunked_decoded_length_mismatch_returns_incomplete_body() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
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
rustfs/src/admin/router.rs
+19
View File
@@ -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<StatusOperation> = 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");
rustfs/src/error.rs
+129 -80
View File
@@ -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<S3sBodyStreamError> {
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<S3sBodyStreamError> {
if let Some(classified) = classify_s3s_body_stream_error_display(err) {
return Some(classified);
}
if let Some(io_err) = err.downcast_ref::<std::io::Error>()
&& 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::<std::io::Error>()
&& 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<ApiError> for S3Error {
@@ -310,69 +333,60 @@ impl From<ApiError> for S3Error {
impl From<StorageError> 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::<rustfs_rio::ChecksumMismatch>().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::<rustfs_rio::ChecksumMismatch>().is_some()
|| inner.downcast_ref::<rustfs_rio::BadDigest>().is_some()
|| inner.downcast_ref::<rustfs_rio::Sha256Mismatch>().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::<UploadLimitExceeded>(inner)
{
return ApiError {
code: S3ErrorCode::EntityTooLarge,
message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
source: Some(Box::new(err)),
};
}
if error_chain_has_type::<UploadLimitExceeded>(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::<ServerSideSourceReadError>(inner)
{
return ApiError {
code: S3ErrorCode::ServiceUnavailable,
message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
source: Some(Box::new(err)),
};
}
if error_chain_has_type::<ServerSideSourceReadError>(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::<KmsUnavailableError>())
.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::<rustfs_kms::KmsError>()),
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::<KmsUnavailableError>().is_some()
|| matches!(
inner.downcast_ref::<rustfs_kms::KmsError>(),
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<std::io::Error> 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::<rustfs_rio::ChecksumMismatch>(inner)
|| error_chain_has_type::<rustfs_rio::BadDigest>(inner)
|| error_chain_has_type::<rustfs_rio::Sha256Mismatch>(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<std::io::Error> 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::<rustfs_rio::IncompleteBody>(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));
}
}
rustfs/src/server/http.rs
+4
View File
@@ -155,10 +155,13 @@ static HTTP_STATUS_CLASS_METRICS: std::sync::LazyLock<[HttpStatusClassMetrics; 6
static HTTP_TRANSPORT_FAILURES_COUNTER: std::sync::LazyLock<metrics::Counter> =
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"));