Accept and discard optional transport-padding.

'transport-padding' is defined as any amount of linear whitespace (space
or horizontal tab) between a boundary and its corresponding newline.

See also https://tools.ietf.org/html/rfc2046#section-5.1.1

Fixes #25.

src/buffer.rs

@@ -65,6 +65,10 @@ impl StreamBuffer {
         }
     }
 
+    pub fn peek_exact(&mut self, size: usize) -> Option<&[u8]> {
+        self.buf.get(..size)
+    }
+
     pub fn read_until(&mut self, pattern: &[u8]) -> Option<Bytes> {
         twoway::find_bytes(&self.buf, pattern).map(|idx| self.buf.split_to(idx + pattern.len()).freeze())
     }
@@ -73,6 +77,19 @@ impl StreamBuffer {
         twoway::find_bytes(&self.buf, pattern).map(|idx| self.buf.split_to(idx).freeze())
     }
 
+    pub fn advance_past_transport_padding(&mut self) -> bool {
+        match self.buf.iter().position(|b| *b != b' ' && *b != b'\t') {
+            Some(pos) => {
+                self.buf.advance(pos);
+                true
+            }
+            None => {
+                self.buf.clear();
+                false
+            }
+        }
+    }
+
     pub fn read_field_data(
         &mut self,
         boundary: &str,

src/multipart.rs

@@ -307,7 +307,7 @@ impl Stream for Multipart {
 
         if state.stage == StreamingStage::ReadingBoundary {
             let boundary = &state.boundary;
-            let boundary_deriv_len = constants::BOUNDARY_EXT.len() + boundary.len() + 2;
+            let boundary_deriv_len = constants::BOUNDARY_EXT.len() + boundary.len();
 
             let boundary_bytes = match stream_buffer.read_exact(boundary_deriv_len) {
                 Some(bytes) => bytes,
@@ -320,17 +320,59 @@ impl Stream for Multipart {
                 }
             };
 
-            if &boundary_bytes[..]
-                == format!("{}{}{}", constants::BOUNDARY_EXT, boundary, constants::BOUNDARY_EXT).as_bytes()
-            {
+            if &boundary_bytes[..] == format!("{}{}", constants::BOUNDARY_EXT, boundary).as_bytes() {
+                state.stage = StreamingStage::DeterminingBoundaryType;
+            } else {
+                return Poll::Ready(Some(Err(crate::Error::IncompleteStream)));
+            }
+        }
+
+        if state.stage == StreamingStage::DeterminingBoundaryType {
+            let ext_len = constants::BOUNDARY_EXT.len();
+            let next_bytes = match stream_buffer.peek_exact(ext_len) {
+                Some(bytes) => bytes,
+                None => {
+                    return if stream_buffer.eof {
+                        Poll::Ready(Some(Err(crate::Error::IncompleteStream)))
+                    } else {
+                        Poll::Pending
+                    };
+                }
+            };
+
+            if next_bytes == constants::BOUNDARY_EXT.as_bytes() {
                 state.stage = StreamingStage::Eof;
                 return Poll::Ready(None);
+            } else {
+                state.stage = StreamingStage::ReadingTransportPadding;
             }
+        }
 
-            if &boundary_bytes[..] != format!("{}{}{}", constants::BOUNDARY_EXT, boundary, constants::CRLF).as_bytes() {
-                return Poll::Ready(Some(Err(crate::Error::IncompleteStream)));
-            } else {
+        if state.stage == StreamingStage::ReadingTransportPadding {
+            if !stream_buffer.advance_past_transport_padding() {
+                return if stream_buffer.eof {
+                    Poll::Ready(Some(Err(crate::Error::IncompleteStream)))
+                } else {
+                    Poll::Pending
+                };
+            }
+
+            let crlf_len = constants::CRLF.len();
+            let crlf_bytes = match stream_buffer.read_exact(crlf_len) {
+                Some(bytes) => bytes,
+                None => {
+                    return if stream_buffer.eof {
+                        Poll::Ready(Some(Err(crate::Error::IncompleteStream)))
+                    } else {
+                        Poll::Pending
+                    };
+                }
+            };
+
+            if &crlf_bytes[..] == constants::CRLF.as_bytes() {
                 state.stage = StreamingStage::ReadingFieldHeaders;
+            } else {
+                return Poll::Ready(Some(Err(crate::Error::IncompleteStream)));
             }
         }
 

src/state.rs

@@ -20,6 +20,8 @@ pub(crate) enum StreamingStage {
     CleaningPrevFieldData,
     FindingFirstBoundary,
     ReadingBoundary,
+    DeterminingBoundaryType,
+    ReadingTransportPadding,
     ReadingFieldHeaders,
     ReadingFieldData,
     Eof,

tests/integration.rs

@@ -63,6 +63,34 @@ async fn test_multipart_clean_field() {
     assert!(m.next_field().await.unwrap().is_none());
 }
 
+#[tokio::test]
+async fn test_multipart_transport_padding() {
+    let data = "--X-BOUNDARY \t \r\nContent-Disposition: form-data; name=\"my_text_field\"\r\n\r\nabcd\r\n--X-BOUNDARY     \r\nContent-Disposition: form-data; name=\"my_file_field\"; filename=\"a-text-file.txt\"\r\nContent-Type: text/plain\r\n\r\nHello world\nHello\r\nWorld\rAgain\r\n--X-BOUNDARY--\t\t\t\t\t\r\n";
+    let stream = stream::iter(
+        data.chars()
+            .map(|ch| ch.to_string())
+            .map(|part| multer::Result::Ok(Bytes::copy_from_slice(part.as_bytes()))),
+    );
+
+    let mut m = Multipart::new(stream, "X-BOUNDARY");
+
+    assert!(m.next_field().await.unwrap().is_some());
+    assert!(m.next_field().await.unwrap().is_some());
+    assert!(m.next_field().await.unwrap().is_none());
+
+    let bad_data = "--X-BOUNDARY \t \r\nContent-Disposition: form-data; name=\"my_text_field\"\r\n\r\nabcd\r\n--X-BOUNDARYzz     \r\nContent-Disposition: form-data; name=\"my_file_field\"; filename=\"a-text-file.txt\"\r\nContent-Type: text/plain\r\n\r\nHello world\nHello\r\nWorld\rAgain\r\n--X-BOUNDARY--\t\t\t\t\t\r\n";
+    let bad_stream = stream::iter(
+        bad_data
+            .chars()
+            .map(|ch| ch.to_string())
+            .map(|part| multer::Result::Ok(Bytes::copy_from_slice(part.as_bytes()))),
+    );
+
+    let mut m = Multipart::new(bad_stream, "X-BOUNDARY");
+    assert!(m.next_field().await.unwrap().is_some());
+    assert!(m.next_field().await.is_err());
+}
+
 #[tokio::test]
 async fn test_multipart_header() {
     let should_pass = [