diff --git a/src/http/multipart.rs b/src/http/multipart.rs index 2977144..e17cd30 100644 --- a/src/http/multipart.rs +++ b/src/http/multipart.rs @@ -111,9 +111,12 @@ impl Multipart { let remaining = limit - out.len(); let mut input = open_file_part(file)?; let read_limit = file.len.min(u64::try_from(remaining).unwrap_or(u64::MAX)); - Read::by_ref(&mut input) + let copied = Read::by_ref(&mut input) .take(read_limit) .read_to_end(&mut out)?; + if u64::try_from(copied).unwrap_or(u64::MAX) != read_limit { + return Err(premature_file_eof().into()); + } } append_preview(&mut out, limit, b"\r\n"); } @@ -145,7 +148,7 @@ impl Multipart { FieldValue::File(file) => { out.write_all(field.header.as_bytes())?; let input = open_file_part(file)?; - std::io::copy(&mut input.take(file.len), &mut out)?; + copy_file_exact(input, &mut out, file.len)?; out.write_all(b"\r\n")?; } } @@ -256,6 +259,25 @@ fn open_file_part(file: &FilePart) -> Result { Ok(opened) } +fn premature_file_eof() -> std::io::Error { + std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "multipart file ended before its expected length", + ) +} + +fn copy_file_exact( + input: impl Read, + out: &mut impl Write, + expected_len: u64, +) -> Result<(), MultipartError> { + let copied = std::io::copy(&mut input.take(expected_len), out)?; + if copied != expected_len { + return Err(premature_file_eof().into()); + } + Ok(()) +} + async fn open_file_part_async(file: &FilePart) -> Result { let opened = tokio::fs::File::open(&file.path).await?; validate_opened_file_part(file, &opened.metadata().await?)?; @@ -409,7 +431,10 @@ impl Stream for MultipartStream { MultipartStreamState::OpeningFile { header, open } => { let file = match open.as_mut().poll(cx) { Poll::Ready(Ok(file)) => file, - Poll::Ready(Err(err)) => return Poll::Ready(Some(Err(err))), + Poll::Ready(Err(err)) => { + self.state = MultipartStreamState::Done; + return Poll::Ready(Some(Err(err))); + } Poll::Pending => return Poll::Pending, }; let header = std::mem::replace(header, Bytes::new()); @@ -435,7 +460,8 @@ impl Stream for MultipartStream { let mut read_buf = ReadBuf::new(&mut bytes[..read_len]); match Pin::new(file).poll_read(cx, &mut read_buf) { Poll::Ready(Ok(())) if read_buf.filled().is_empty() => { - self.state = MultipartStreamState::FileCrlf; + self.state = MultipartStreamState::Done; + return Poll::Ready(Some(Err(premature_file_eof().into()))); } Poll::Ready(Ok(())) => { *remaining = remaining.saturating_sub(read_buf.filled().len() as u64); @@ -443,7 +469,10 @@ impl Stream for MultipartStream { read_buf.filled(), )))); } - Poll::Ready(Err(err)) => return Poll::Ready(Some(Err(err.into()))), + Poll::Ready(Err(err)) => { + self.state = MultipartStreamState::Done; + return Poll::Ready(Some(Err(err.into()))); + } Poll::Pending => return Poll::Pending, } } @@ -461,6 +490,7 @@ impl Stream for MultipartStream { #[cfg(test)] mod tests { use super::*; + use futures_util::StreamExt; #[test] fn multipart_small_json_file_uses_detected_content_type() { @@ -560,6 +590,54 @@ mod tests { assert!(err.to_string().contains("changed size")); } + #[test] + fn copy_file_exact_rejects_a_short_reader() { + let mut out = Vec::new(); + + let err = copy_file_exact(std::io::Cursor::new(b"short"), &mut out, 10).unwrap_err(); + + assert!(matches!( + err, + MultipartError::Io(ref err) if err.kind() == std::io::ErrorKind::UnexpectedEof + )); + } + + #[tokio::test] + async fn multipart_stream_rejects_file_truncated_after_open() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("payload.txt"); + std::fs::write(&path, b"expected payload").unwrap(); + let multipart = Multipart::from_cli_fields(&[format!("file=@{}", path.display())]) + .unwrap() + .unwrap(); + let mut stream = multipart.stream(); + + assert!(stream.next().await.unwrap().is_ok()); + std::fs::write(&path, b"").unwrap(); + let err = stream.next().await.unwrap().unwrap_err(); + + assert!(matches!( + err, + MultipartError::Io(ref err) if err.kind() == std::io::ErrorKind::UnexpectedEof + )); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn multipart_stream_is_terminal_after_file_open_error() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("payload.txt"); + std::fs::write(&path, b"payload").unwrap(); + let multipart = Multipart::from_cli_fields(&[format!("file=@{}", path.display())]) + .unwrap() + .unwrap(); + std::fs::remove_file(path).unwrap(); + let mut stream = multipart.stream(); + + assert!(stream.next().await.unwrap().is_err()); + assert!(stream.next().await.is_none()); + } + #[test] fn multipart_content_len_reports_body_too_large_on_overflow() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/http/request.rs b/src/http/request.rs index ca718ff..aaa48ed 100644 --- a/src/http/request.rs +++ b/src/http/request.rs @@ -384,7 +384,7 @@ pub(crate) fn request_body_sha256_hex(body: &RequestBody) -> Result hasher.update(bytes), RequestBodySource::File { path, len } => { let file = open_regular_file_at_len(path, *len)?; - hash_reader(&mut hasher, file.take(*len))?; + hash_reader_exact(&mut hasher, file, *len)?; } RequestBodySource::Multipart(multipart) => { let mut writer = Sha256Writer(&mut hasher); @@ -423,6 +423,25 @@ pub(super) fn hash_reader(hasher: &mut Sha256, mut reader: impl Read) -> Result< } } +fn hash_reader_exact( + hasher: &mut Sha256, + reader: impl Read, + expected_len: u64, +) -> Result<(), FetchError> { + let mut reader = reader.take(expected_len); + let mut read = 0_u64; + let mut buf = [0_u8; 8192]; + while read < expected_len { + let len = reader.read(&mut buf)?; + if len == 0 { + return Err(premature_request_file_eof().into()); + } + read += len as u64; + hasher.update(&buf[..len]); + } + Ok(()) +} + pub(super) fn hex_sha256_stream(reader: impl Read) -> Result { let mut hasher = Sha256::new(); hash_reader(&mut hasher, reader)?; @@ -498,13 +517,59 @@ pub(super) fn inferred_request_body_content_len( request_body_content_len(body) } +fn premature_request_file_eof() -> std::io::Error { + std::io::Error::new( + ErrorKind::UnexpectedEof, + "request body file ended before its expected length", + ) +} + +struct ExactLengthReader { + inner: tokio::io::Take, + terminal: bool, +} + +impl ExactLengthReader { + fn new(inner: R, expected_len: u64) -> Self { + Self { + inner: inner.take(expected_len), + terminal: false, + } + } +} + +impl AsyncRead for ExactLengthReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.terminal || self.inner.limit() == 0 || buf.remaining() == 0 { + return Poll::Ready(Ok(())); + } + + let filled_before = buf.filled().len(); + match Pin::new(&mut self.inner).poll_read(cx, buf) { + Poll::Ready(Ok(())) if buf.filled().len() == filled_before => { + self.terminal = true; + Poll::Ready(Err(premature_request_file_eof())) + } + Poll::Ready(Err(err)) => { + self.terminal = true; + Poll::Ready(Err(err)) + } + result => result, + } + } +} + pub(crate) fn request_body_to_transport_body(body: RequestBodyPayload) -> Result { match body.source { RequestBodySource::Bytes(bytes) => Ok(Body::from(bytes)), RequestBodySource::File { path, len } => { let file = open_regular_file_at_len(&path, len)?; Ok(Body::wrap_stream(ReaderStream::new( - tokio::fs::File::from_std(file).take(len), + ExactLengthReader::new(tokio::fs::File::from_std(file), len), ))) } RequestBodySource::Stdin => Ok(Body::wrap_stream(ReaderStream::new(tokio::io::stdin()))), @@ -530,7 +595,10 @@ pub(crate) fn request_body_source_to_async_reader( RequestBodySource::Bytes(bytes) => Ok(Box::pin(Cursor::new(bytes))), RequestBodySource::File { path, len } => { let file = open_regular_file_at_len(&path, len)?; - Ok(Box::pin(tokio::fs::File::from_std(file).take(len))) + Ok(Box::pin(ExactLengthReader::new( + tokio::fs::File::from_std(file), + len, + ))) } RequestBodySource::Stdin => Ok(Box::pin(tokio::io::stdin())), RequestBodySource::Multipart(multipart) => Ok(Box::pin(StreamReader::new( @@ -579,7 +647,10 @@ pub(crate) fn request_body_source_to_bytes( RequestBodySource::File { path, len } => { let file = open_regular_file_at_len(&path, len)?; let mut out = Vec::with_capacity(usize::try_from(len).unwrap_or(0)); - file.take(len).read_to_end(&mut out)?; + let copied = file.take(len).read_to_end(&mut out)?; + if u64::try_from(copied).unwrap_or(u64::MAX) != len { + return Err(premature_request_file_eof().into()); + } Ok(out) } RequestBodySource::Stdin => { @@ -611,7 +682,11 @@ fn request_body_source_to_bytes_limited( RequestBodySource::File { path, len } => { ensure_materialized_len_u64(len, max_bytes, limit_error)?; let file = open_regular_file_at_len(&path, len)?; - read_to_end_limited(file.take(len), max_bytes, limit_error) + let out = read_to_end_limited(file.take(len), max_bytes, limit_error)?; + if u64::try_from(out.len()).unwrap_or(u64::MAX) != len { + return Err(premature_request_file_eof().into()); + } + Ok(out) } RequestBodySource::Stdin => { let stdin = std::io::stdin(); @@ -730,9 +805,12 @@ fn dry_run_source_preview( }), RequestBodySource::File { path, len } => { let file = open_regular_file_at_len(path, *len)?; + let expected = (*len).min(u64::try_from(limit).unwrap_or(u64::MAX)); let mut bytes = Vec::new(); - file.take(u64::try_from(limit).unwrap_or(u64::MAX)) - .read_to_end(&mut bytes)?; + let copied = file.take(expected).read_to_end(&mut bytes)?; + if u64::try_from(copied).unwrap_or(u64::MAX) != expected { + return Err(premature_request_file_eof().into()); + } Ok(DryRunBodyPreview { bytes, truncated: *len > u64::try_from(limit).unwrap_or(u64::MAX), @@ -1092,6 +1170,19 @@ mod tests { assert_eq!(body.1.as_deref(), Some("application/x-www-form-urlencoded")); } + #[tokio::test] + async fn exact_length_reader_rejects_premature_eof_and_becomes_terminal() { + let mut reader = ExactLengthReader::new(Cursor::new(b"short".to_vec()), 10); + let mut out = Vec::new(); + + let err = reader.read_to_end(&mut out).await.unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::UnexpectedEof); + assert_eq!(out, b"short"); + let mut after_error = [0_u8; 1]; + assert_eq!(reader.read(&mut after_error).await.unwrap(), 0); + } + #[test] fn request_body_file_errors_match_go_cli_surface() { let dir = tempfile::tempdir().unwrap();