From 0d00b886ac5d447b6feaf6e2749a97917c2b4ec0 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 2 Jun 2026 19:31:51 +0800 Subject: [PATCH] fix(rio): map truncated put bodies to incompletebody (#3168) * fix(rio): surface incomplete put bodies Propagate incomplete PUT request bodies as IncompleteBody instead of allowing erasure encode to treat truncated input as a normal EOF.\n\n- mark premature EOFs in HardLimitReader with an explicit IncompleteBody error\n- preserve EOF error chains through read_full and map them to S3 IncompleteBody\n- stop erasure encode from swallowing UnexpectedEof on truncated input\n- add regression tests for reader, erasure encode, and API error mapping\n\nRefs: rustfs/backlog#654 * fix(s3): honor decoded length for aws chunked put Use x-amz-decoded-content-length for aws-chunked PutObject requests so trailer-checksum uploads are sized against the decoded payload instead of the wire-encoded content-length.\n\n- prefer decoded content length for aws-chunked put bodies\n- add a regression test covering the size selection logic\n- keeps the incomplete body fix working for truly truncated uploads while restoring checksum trailer compatibility\n\nRefs: rustfs/backlog#654 * fix(io): follow up review comments on incompletebody handling Address PR review feedback by restoring read_full's existing EOF contract, adding a dedicated read_full_or_eof helper for erasure encoding, covering nested incomplete-body error chains, and documenting plus hardening aws-chunked size selection.\n\n- keep read_full returning early EOF on empty reads\n- use read_full_or_eof only in erasure encoding paths\n- detect aws-chunked via content-encoding or transfer-encoding\n- add nested error-chain and aws-chunked regression tests\n\nRefs: rustfs/backlog#654 * fix(rio): surface incomplete put bodies Propagate incomplete PUT request bodies as IncompleteBody instead of allowing erasure encode to treat truncated input as a normal EOF.\n\n- mark premature EOFs in HardLimitReader with an explicit IncompleteBody error\n- preserve EOF error chains through read_full and map them to S3 IncompleteBody\n- stop erasure encode from swallowing UnexpectedEof on truncated input\n- add regression tests for reader, erasure encode, and API error mapping\n\nRefs: rustfs/backlog#654 * fix(s3): honor decoded length for aws chunked put Use x-amz-decoded-content-length for aws-chunked PutObject requests so trailer-checksum uploads are sized against the decoded payload instead of the wire-encoded content-length.\n\n- prefer decoded content length for aws-chunked put bodies\n- add a regression test covering the size selection logic\n- keeps the incomplete body fix working for truly truncated uploads while restoring checksum trailer compatibility\n\nRefs: rustfs/backlog#654 * fix(io): follow up review comments on incompletebody handling Address PR review feedback by restoring read_full's existing EOF contract, adding a dedicated read_full_or_eof helper for erasure encoding, covering nested incomplete-body error chains, and documenting plus hardening aws-chunked size selection.\n\n- keep read_full returning early EOF on empty reads\n- use read_full_or_eof only in erasure encoding paths\n- detect aws-chunked via content-encoding or transfer-encoding\n- add nested error-chain and aws-chunked regression tests\n\nRefs: rustfs/backlog#654 * fix(rio): reject bytes beyond hard limit * fix(ecstore): reject zero-sized erasure blocks --- crates/ecstore/src/erasure_coding/encode.rs | 60 ++++++++++++++-- crates/ecstore/src/erasure_coding/erasure.rs | 45 +++++++++++- crates/rio/src/errors.rs | 7 ++ crates/rio/src/hardlimit_reader.rs | 73 +++++++++++++++++++- crates/utils/src/io.rs | 45 +++++++++--- rustfs/src/app/object_usecase.rs | 60 ++++++++++++---- rustfs/src/error.rs | 57 +++++++++++++-- 7 files changed, 310 insertions(+), 37 deletions(-) diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index 386ecff28..2be77ef3f 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/crates/ecstore/src/erasure_coding/encode.rs @@ -209,6 +209,13 @@ impl Erasure { where R: AsyncRead + Send + Sync + Unpin + 'static, { + if self.block_size == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "erasure block_size must be non-zero", + )); + } + // Bound queued encoded blocks by memory budget to avoid per-request spikes. let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count()); let max_inflight_bytes = rustfs_utils::get_env_usize( @@ -223,8 +230,9 @@ impl Erasure { let mut total = 0; let mut buf = vec![0u8; block_size]; loop { - match rustfs_utils::read_full(&mut reader, &mut buf).await { - Ok(n) if n > 0 => { + match rustfs_utils::read_full_or_eof(&mut reader, &mut buf).await { + Ok(Some(n)) => { + debug_assert!(n > 0, "non-zero block_size prevents zero-length reads"); total += n; let erasure = self.clone(); let encode_buf = std::mem::take(&mut buf); @@ -243,7 +251,7 @@ impl Erasure { return Err(std::io::Error::other(format!("Failed to send encoded data : {err}"))); } } - Ok(_) => { + Ok(None) => { break; } Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { @@ -253,7 +261,7 @@ impl Erasure { { return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())); } - break; + return Err(e); } Err(e) => { return Err(e); @@ -327,7 +335,9 @@ impl Erasure { mod tests { use super::*; use crate::erasure_coding::{BitrotWriterWrapper, CustomWriter}; + use rustfs_rio::HardLimitReader; use rustfs_utils::HashAlgorithm; + use std::io::Cursor; use std::pin::Pin; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; @@ -384,6 +394,48 @@ mod tests { assert!(!committed.lock().unwrap().is_empty()); } + #[tokio::test] + async fn encode_returns_unexpected_eof_for_truncated_limited_reader() { + let committed = Arc::new(Mutex::new(Vec::new())); + let writer = DeferredCommitWriter::new(committed); + let mut writers = vec![Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(writer), + 16, + HashAlgorithm::HighwayHash256S, + ))]; + + let erasure = Arc::new(Erasure::new(1, 0, 16)); + let truncated = HardLimitReader::new(Cursor::new(b"short".to_vec()), 10); + + let err = match erasure.encode(truncated, &mut writers, 1).await { + Ok(_) => panic!("truncated input must fail"), + Err(err) => err, + }; + + assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); + } + + #[tokio::test] + async fn encode_rejects_zero_block_size() { + let committed = Arc::new(Mutex::new(Vec::new())); + let writer = DeferredCommitWriter::new(committed); + let mut writers = vec![Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(writer), + 16, + HashAlgorithm::HighwayHash256S, + ))]; + + let erasure = Arc::new(Erasure::new(1, 0, 0)); + let reader = tokio::io::BufReader::new(std::io::Cursor::new(b"payload".to_vec())); + let err = erasure + .encode(reader, &mut writers, 1) + .await + .expect_err("zero block size must be rejected"); + + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); + assert!(err.to_string().contains("block_size")); + } + /// encode_inline_small: empty reader returns (reader, 0) without writing to any shard. #[tokio::test] async fn encode_inline_small_empty_stream_returns_zero() { diff --git a/crates/ecstore/src/erasure_coding/erasure.rs b/crates/ecstore/src/erasure_coding/erasure.rs index 2a07fb686..6d2dd962d 100644 --- a/crates/ecstore/src/erasure_coding/erasure.rs +++ b/crates/ecstore/src/erasure_coding/erasure.rs @@ -579,12 +579,22 @@ impl Erasure { F: FnMut(std::io::Result>) -> Fut + Send, Fut: std::future::Future> + Send, { + if self.block_size == 0 { + on_block(Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "erasure block_size must be non-zero", + ))) + .await?; + return Ok(0); + } + let block_size = self.block_size; let mut total = 0; let mut buf = vec![0u8; block_size]; loop { - match rustfs_utils::read_full(&mut *reader, &mut buf).await { - Ok(n) if n > 0 => { + match rustfs_utils::read_full_or_eof(&mut *reader, &mut buf).await { + Ok(Some(n)) => { + debug_assert!(n > 0, "non-zero block_size prevents zero-length reads"); warn!("encode_stream_callback_async read n={}", n); total += n; let erasure = self.clone(); @@ -604,7 +614,7 @@ impl Erasure { buf = returned_buf; on_block(res).await? } - Ok(_) => { + Ok(None) => { warn!("encode_stream_callback_async read unexpected ok"); break; } @@ -1032,6 +1042,35 @@ mod tests { assert_eq!(&recovered, &data_clone); } + #[tokio::test] + async fn test_encode_stream_callback_async_reports_zero_block_size() { + use std::io::Cursor; + use std::sync::{Arc, Mutex}; + + let erasure = Arc::new(Erasure::new(1, 0, 0)); + let mut reader = Cursor::new(b"payload".to_vec()); + let observed = Arc::new(Mutex::new(None)); + let observed_clone = observed.clone(); + + let total = erasure + .encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| { + let observed = observed_clone.clone(); + async move { + let err = res.expect_err("zero block size should report an error"); + *observed.lock().unwrap() = Some((err.kind(), err.to_string())); + Ok(()) + } + }) + .await + .expect("callback should handle the zero block size error"); + + assert_eq!(total, 0); + let observed = observed.lock().unwrap(); + let (kind, message) = observed.as_ref().expect("callback should be invoked once"); + assert_eq!(*kind, std::io::ErrorKind::InvalidInput); + assert!(message.contains("block_size")); + } + // SIMD mode specific tests mod simd_tests { use super::*; diff --git a/crates/rio/src/errors.rs b/crates/rio/src/errors.rs index 7be06b964..1f87c4da4 100644 --- a/crates/rio/src/errors.rs +++ b/crates/rio/src/errors.rs @@ -62,6 +62,13 @@ pub struct ChecksumMismatch { pub got: String, } +/// Request body ended before the declared size was fully read. +#[derive(Error, Debug, Clone, PartialEq)] +#[error("Incomplete body: {remaining} bytes were still expected")] +pub struct IncompleteBody { + pub remaining: i64, +} + /// Invalid checksum error #[derive(Error, Debug, Clone, PartialEq)] #[error("invalid checksum")] diff --git a/crates/rio/src/hardlimit_reader.rs b/crates/rio/src/hardlimit_reader.rs index e50b052f5..8149e9fe9 100644 --- a/crates/rio/src/hardlimit_reader.rs +++ b/crates/rio/src/hardlimit_reader.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::IncompleteBody; use pin_project_lite::pin_project; use std::io::{Error, Result}; use std::pin::Pin; @@ -40,8 +41,25 @@ where if self.remaining < 0 { return Poll::Ready(Err(Error::other("input provided more bytes than specified"))); } + let original_filled = buf.filled().len(); + if self.remaining == 0 { + let mut discard = [0u8; 8192]; + let mut discard_buf = ReadBuf::new(&mut discard); + return match self.as_mut().project().inner.poll_read(cx, &mut discard_buf) { + Poll::Pending => Poll::Pending, + Poll::Ready(Ok(())) => { + if discard_buf.filled().is_empty() { + debug_assert_eq!(buf.filled().len(), original_filled); + Poll::Ready(Ok(())) + } else { + Poll::Ready(Err(Error::other("input provided more bytes than specified"))) + } + } + Poll::Ready(Err(err)) => Poll::Ready(Err(err)), + }; + } // Save the initial length - let before = buf.filled().len(); + let before = original_filled; // Poll the inner reader let this = self.as_mut().project(); @@ -50,6 +68,14 @@ where if let Poll::Ready(Ok(())) = &poll { let after = buf.filled().len(); let read = (after - before) as i64; + if read == 0 && *this.remaining > 0 { + return Poll::Ready(Err(Error::new( + std::io::ErrorKind::UnexpectedEof, + IncompleteBody { + remaining: *this.remaining, + }, + ))); + } *this.remaining -= read; if *this.remaining < 0 { return Poll::Ready(Err(Error::other("input provided more bytes than specified"))); @@ -73,7 +99,7 @@ mod tests { async fn test_hardlimit_reader_normal() { let data = b"hello world"; let reader = BufReader::new(&data[..]); - let hardlimit = HardLimitReader::new(reader, 20); + let hardlimit = HardLimitReader::new(reader, data.len() as i64); let mut r = hardlimit; let mut buf = Vec::new(); let n = r.read_to_end(&mut buf).await.unwrap(); @@ -121,11 +147,52 @@ mod tests { async fn test_hardlimit_reader_empty() { let data = b""; let reader = BufReader::new(&data[..]); - let hardlimit = HardLimitReader::new(reader, 5); + let hardlimit = HardLimitReader::new(reader, 0); let mut r = hardlimit; let mut buf = Vec::new(); let n = r.read_to_end(&mut buf).await.unwrap(); assert_eq!(n, 0); assert_eq!(&buf, data); } + + #[tokio::test] + async fn test_hardlimit_reader_short_input_returns_unexpected_eof() { + let data = b"abc"; + let reader = BufReader::new(&data[..]); + let mut r = HardLimitReader::new(reader, 5); + let mut buf = [0u8; 8]; + + let err = read_full(&mut r, &mut buf) + .await + .expect_err("short input must surface unexpected eof"); + + assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); + assert!( + err.get_ref() + .and_then(|inner| inner.downcast_ref::()) + .and_then(|inner| inner.get_ref()) + .and_then(|inner| inner.downcast_ref::()) + .is_some(), + "error should retain the incomplete body marker" + ); + } + + #[tokio::test] + async fn test_hardlimit_reader_rejects_extra_bytes_after_limit() { + let data = b"abcdef"; + let reader = BufReader::new(&data[..]); + let mut r = HardLimitReader::new(reader, 3); + + let mut first = [0u8; 3]; + let n = read_full(&mut r, &mut first).await.expect("first read should consume limit"); + assert_eq!(n, 3); + assert_eq!(&first, b"abc"); + + let mut second = [0u8; 1]; + let err = read_full(&mut r, &mut second) + .await + .expect_err("bytes beyond the declared limit must be rejected"); + assert_eq!(err.kind(), std::io::ErrorKind::Other); + assert!(err.to_string().contains("more bytes than specified")); + } } diff --git a/crates/utils/src/io.rs b/crates/utils/src/io.rs index 94777b8dc..92e69e5db 100644 --- a/crates/utils/src/io.rs +++ b/crates/utils/src/io.rs @@ -29,10 +29,15 @@ pub async fn write_all(writer: &mut W, buf: Ok(total) } -/// Read exactly buf.len() bytes into buf, or return an error if EOF is reached before. -/// Like Go's io.ReadFull. -#[allow(dead_code)] -pub async fn read_full(mut reader: R, mut buf: &mut [u8]) -> std::io::Result { +/// Read up to buf.len() bytes into buf and distinguish a clean EOF from a short read. +/// +/// Returns `Ok(None)` when EOF is reached before any bytes are read, `Ok(Some(n))` when +/// at least one byte is read, and preserves the underlying error chain when the reader +/// fails after a partial fill. +pub async fn read_full_or_eof( + mut reader: R, + mut buf: &mut [u8], +) -> std::io::Result> { let mut total = 0; while !buf.is_empty() { let n = match reader.read(buf).await { @@ -46,22 +51,29 @@ pub async fn read_full(mut reader: R, mut bu if e.kind() == std::io::ErrorKind::InvalidData { return Err(e); } - return Err(std::io::Error::new( - std::io::ErrorKind::UnexpectedEof, - format!("read {total} bytes, error: {e}"), - )); + return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, e)); } }; if n == 0 { if total > 0 { - return Ok(total); + return Ok(Some(total)); } - return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "early EOF")); + return Ok(None); } buf = &mut buf[n..]; total += n; } - Ok(total) + Ok(Some(total)) +} + +/// Read exactly buf.len() bytes into buf, or return an error if EOF is reached before any bytes are read. +/// Like Go's io.ReadFull. +#[allow(dead_code)] +pub async fn read_full(reader: R, buf: &mut [u8]) -> std::io::Result { + match read_full_or_eof(reader, buf).await? { + Some(n) => Ok(n), + None => Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "early EOF")), + } } /// Encodes a u64 into buf and returns the number of bytes written. @@ -163,6 +175,17 @@ mod tests { assert_eq!(buf, data[..size / 3]); } + #[tokio::test] + async fn test_read_full_or_eof_returns_none_for_empty_reader() { + let data = b""; + let mut reader = BufReader::new(&data[..]); + let mut buf = [0u8; 8]; + + let n = read_full_or_eof(&mut reader, &mut buf).await.unwrap(); + + assert_eq!(n, None); + } + #[test] fn test_put_uvarint_and_uvarint_zero() { let mut buf = [0u8; 16]; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 4d4b9727d..64d6859fb 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -129,6 +129,28 @@ const ACCEPT_RANGES_BYTES: &str = "bytes"; const MAX_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 64 * 1024 * 1024; static GET_OBJECT_BUFFER_THRESHOLD_WARNED: AtomicBool = AtomicBool::new(false); +fn decoded_content_length_from_headers(headers: &HeaderMap) -> S3Result> { + let Some(val) = headers.get(AMZ_DECODED_CONTENT_LENGTH) else { + return Ok(None); + }; + + match atoi::atoi::(val.as_bytes()) { + Some(x) => Ok(Some(x)), + None => Err(s3_error!(UnexpectedContent)), + } +} + +fn request_uses_aws_chunked(headers: &HeaderMap) -> bool { + let has_aws_chunked = |header_name: &str| { + headers + .get(header_name) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.split(',').any(|part| part.trim().eq_ignore_ascii_case("aws-chunked"))) + }; + + has_aws_chunked("content-encoding") || has_aws_chunked("transfer-encoding") +} + struct DeadlockRequestGuard { deadlock_detector: Arc, request_id: String, @@ -1728,18 +1750,12 @@ impl DefaultObjectUsecase { let Some(body) = body else { return Err(s3_error!(IncompleteBody)) }; - let mut size = match content_length { - Some(c) => c, - None => { - if let Some(val) = req.headers.get(AMZ_DECODED_CONTENT_LENGTH) { - match atoi::atoi::(val.as_bytes()) { - Some(x) => x, - None => return Err(s3_error!(UnexpectedContent)), - } - } else { - return Err(s3_error!(UnexpectedContent)); - } - } + let decoded_content_length = decoded_content_length_from_headers(&req.headers)?; + let mut size = match (request_uses_aws_chunked(&req.headers), decoded_content_length, content_length) { + (true, Some(decoded), _) => decoded, + (_, _, Some(c)) => c, + (_, Some(decoded), None) => decoded, + _ => return Err(s3_error!(UnexpectedContent)), }; if size == -1 { @@ -4633,6 +4649,26 @@ mod tests { assert!(!should_use_zero_copy(2 * 1024 * 1024, &headers)); } + #[test] + fn aws_chunked_put_prefers_decoded_content_length() { + let mut headers = HeaderMap::new(); + headers.insert("content-encoding", HeaderValue::from_static("aws-chunked")); + headers.insert(AMZ_DECODED_CONTENT_LENGTH, HeaderValue::from_static("71680")); + + let decoded = decoded_content_length_from_headers(&headers).expect("decoded content length should parse"); + assert!(request_uses_aws_chunked(&headers)); + assert_eq!(decoded, Some(71680)); + + let resolved = match (request_uses_aws_chunked(&headers), decoded, Some(99999)) { + (true, Some(decoded), _) => decoded, + (_, _, Some(c)) => c, + (_, Some(decoded), None) => decoded, + _ => unreachable!("test provides a valid size source"), + }; + + assert_eq!(resolved, 71680); + } + #[test] fn should_buffer_get_object_in_memory_respects_hard_safety_cap() { let info = ObjectInfo::default(); diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index dfc40fe51..d5268e5c6 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -181,6 +181,31 @@ impl ApiError { } } +fn error_chain_has_type(err: &(dyn std::error::Error + 'static)) -> bool +where + T: std::error::Error + 'static, +{ + if err.downcast_ref::().is_some() { + return true; + } + + if let Some(io_err) = err.downcast_ref::() + && let Some(inner) = io_err.get_ref() + && error_chain_has_type::(inner) + { + return true; + } + + let mut current = Some(err); + while let Some(err) = current { + if err.downcast_ref::().is_some() { + return true; + } + current = err.source(); + } + false +} + impl From for S3Error { fn from(err: ApiError) -> Self { let mut s3e = S3Error::with_message(err.code, err.message); @@ -260,17 +285,18 @@ impl From for ApiError { fn from(err: std::io::Error) -> Self { // Check if the error is a ChecksumMismatch (BadDigest) if let Some(inner) = err.get_ref() { - if inner.downcast_ref::().is_some() { + if error_chain_has_type::(inner) || error_chain_has_type::(inner) + { return ApiError { code: S3ErrorCode::BadDigest, message: ApiError::error_code_to_message(&S3ErrorCode::BadDigest), source: Some(Box::new(err)), }; } - if inner.downcast_ref::().is_some() { + if error_chain_has_type::(inner) { return ApiError { - code: S3ErrorCode::BadDigest, - message: ApiError::error_code_to_message(&S3ErrorCode::BadDigest), + code: S3ErrorCode::IncompleteBody, + message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody), source: Some(Box::new(err)), }; } @@ -447,6 +473,29 @@ mod tests { } } + #[test] + fn test_api_error_from_unexpected_eof_maps_to_incomplete_body() { + let io_error = IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 7 }); + let api_error: ApiError = io_error.into(); + + assert_eq!(api_error.code, S3ErrorCode::IncompleteBody); + assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); + assert!(api_error.source.is_some()); + } + + #[test] + fn test_api_error_from_nested_unexpected_eof_maps_to_incomplete_body() { + let nested = IoError::new( + ErrorKind::UnexpectedEof, + IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 7 }), + ); + let api_error: ApiError = nested.into(); + + assert_eq!(api_error.code, S3ErrorCode::IncompleteBody); + assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody)); + assert!(api_error.source.is_some()); + } + #[test] fn test_api_error_from_iam_error() { let iam_error = rustfs_iam::error::Error::other("IAM test error");