fix: reject premature EOF in file request bodies

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<std::fs::File, MultipartError> {
     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<tokio::fs::File, MultipartError> {
     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,15 +460,19 @@ 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);
                             return Poll::Ready(Some(Ok(Bytes::copy_from_slice(
                                 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();

src/http/request.rs

@@ -384,7 +384,7 @@ pub(crate) fn request_body_sha256_hex(body: &RequestBody) -> Result<String, Fetc
         RequestBodySource::Bytes(bytes) => 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<String, FetchError> {
     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<R> {
+    inner: tokio::io::Take<R>,
+    terminal: bool,
+}
+
+impl<R: AsyncRead> ExactLengthReader<R> {
+    fn new(inner: R, expected_len: u64) -> Self {
+        Self {
+            inner: inner.take(expected_len),
+            terminal: false,
+        }
+    }
+}
+
+impl<R: AsyncRead + Unpin> AsyncRead for ExactLengthReader<R> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        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<Body, FetchError> {
     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();