form_net/
io.rs

1//! Blocking readers: a pull-style API over `std::io::Read`, free of any
2//! runtime. Ask for the next part, read it like a file, ask for the next.
3
4use alloc::string::String;
5use alloc::vec::Vec;
6use core::fmt;
7use std::io::{self, Read};
8
9use crate::error::{EncodingError, FormError, Location};
10use crate::limits::{Limits, Policy};
11use crate::multipart::owned_head::OwnedPartHead;
12use crate::multipart::parser::{Charset, Event, MultipartParser, PartSummary};
13
14const READ_BUFFER: usize = 64 * 1024;
15
16/// Reads a multipart body from a blocking reader, one part at a time.
17///
18/// It has no hooks: the caller sees each part itself. Dropping a
19/// [`PartReader`] before its end skips the rest of the part at the next
20/// [`next_part`](Self::next_part).
21///
22/// # Examples
23///
24/// ```
25/// use std::io::Read;
26///
27/// let body = b"--X\r\nContent-Disposition: form-data; name=note\r\n\r\nhi\r\n--X\r\nContent-Disposition: form-data; name=f; filename=a.bin\r\n\r\n0123\r\n--X--";
28/// let mut reader = form_net::MultipartReader::new(b"multipart/form-data; boundary=X", &body[..])?;
29/// let mut seen = Vec::new();
30/// while let Some(mut part) = reader.next_part()? {
31///     if part.head().view().is_file() {
32///         let mut bytes = Vec::new();
33///         part.read_to_end(&mut bytes).unwrap();
34///         seen.push(bytes.len());
35///     } else {
36///         seen.push(part.text()?.len());
37///     }
38/// }
39/// assert_eq!(seen, [2, 4]);
40/// # Ok::<(), form_net::FormError>(())
41/// ```
42pub struct MultipartReader<R: Read> {
43    reader: R,
44    parser: MultipartParser<Vec<u8>>,
45    buf: Vec<u8>,
46    start: usize,
47    end: usize,
48    eof: bool,
49    head: OwnedPartHead,
50    pending: Vec<u8>,
51    pending_at: usize,
52    in_part: bool,
53    last: Option<PartSummary>,
54    done: bool,
55    declared: Option<Vec<u8>>,
56    /// Bytes `save` has put on disk for this request.
57    #[cfg(feature = "fs")]
58    disk_used: u64,
59}
60
61impl<R: Read> Drop for MultipartReader<R> {
62    fn drop(&mut self) {
63        crate::wipe::wipe_vec(&mut self.buf);
64        crate::wipe::wipe_vec(&mut self.pending);
65        if let Some(d) = self.declared.as_mut() {
66            crate::wipe::wipe_vec(d);
67        }
68    }
69}
70
71impl<R: Read> fmt::Debug for MultipartReader<R> {
72    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
73        f.debug_struct("MultipartReader")
74            .field("parser", &self.parser)
75            .field("in_part", &self.in_part)
76            .finish_non_exhaustive()
77    }
78}
79
80/// What the reader's loop produced.
81enum Got {
82    Start,
83    Data,
84    End(PartSummary),
85    Done,
86}
87
88impl<R: Read> MultipartReader<R> {
89    /// A reader with the default limits and policy.
90    ///
91    /// # Errors
92    ///
93    /// `UnsupportedMediaType` and the boundary errors.
94    pub fn new(content_type: &[u8], reader: R) -> Result<Self, FormError> {
95        Self::with_limits(content_type, reader, &Limits::DEFAULT, &Policy::DEFAULT)
96    }
97
98    /// A reader with explicit limits and policy.
99    ///
100    /// # Errors
101    ///
102    /// As [`new`](Self::new); `InvalidLimits`.
103    pub fn with_limits(
104        content_type: &[u8],
105        reader: R,
106        limits: &Limits,
107        policy: &Policy,
108    ) -> Result<Self, FormError> {
109        let boundary = crate::Boundary::parse(content_type)?;
110        let parser = MultipartParser::new(boundary, limits, policy)?;
111        Ok(MultipartReader {
112            reader,
113            parser,
114            buf: alloc::vec![0u8; READ_BUFFER],
115            start: 0,
116            end: 0,
117            eof: false,
118            head: OwnedPartHead::with_capacity(limits.multipart_buffer_len(policy)),
119            pending: Vec::new(),
120            pending_at: 0,
121            in_part: false,
122            last: None,
123            done: false,
124            declared: None,
125            #[cfg(feature = "fs")]
126            disk_used: 0,
127        })
128    }
129
130    /// The token field's value, once its part has ended.
131    #[must_use]
132    pub fn captured_token(&self) -> Option<&str> {
133        self.parser.captured_token()
134    }
135
136    fn fill(&mut self) -> Result<(), FormError> {
137        if self.start < self.end || self.eof {
138            return Ok(());
139        }
140        loop {
141            match self.reader.read(&mut self.buf) {
142                Ok(0) => {
143                    self.eof = true;
144                    self.start = 0;
145                    self.end = 0;
146                    return Ok(());
147                }
148                Ok(n) => {
149                    self.start = 0;
150                    self.end = n.min(self.buf.len());
151                    return Ok(());
152                }
153                Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
154                Err(e) => return Err(FormError::Io { kind: e.kind() }),
155            }
156        }
157    }
158
159    /// One parser event; data goes to `pending`.
160    fn step(&mut self) -> Result<Got, FormError> {
161        loop {
162            self.fill()?;
163            let input = self.buf.get(self.start..self.end).unwrap_or(&[]);
164            let step = self.parser.parse(input, self.eof)?;
165            self.start = self.start.saturating_add(step.consumed);
166            match step.event {
167                Event::PartStart(head) => {
168                    self.head.copy_from(&head);
169                    return Ok(Got::Start);
170                }
171                Event::Data(bytes) => {
172                    crate::wipe::wipe(&mut self.pending);
173                    self.pending.clear();
174                    crate::wipe::append_wiping(&mut self.pending, bytes, 4096, usize::MAX);
175                    self.pending_at = 0;
176                    return Ok(Got::Data);
177                }
178                Event::PartEnd(s) => return Ok(Got::End(s)),
179                Event::End => return Ok(Got::Done),
180                _ => {}
181            }
182        }
183    }
184
185    /// The next part, or `None` after the close delimiter. Skips whatever
186    /// is left of the previous part.
187    ///
188    /// # Errors
189    ///
190    /// Every multipart error; `Io { kind }` from the reader.
191    pub fn next_part(&mut self) -> Result<Option<PartReader<'_, R>>, FormError> {
192        if self.in_part {
193            self.parser.skip_part();
194            self.pending.clear();
195            while self.in_part {
196                if let Got::End(s) = self.step()? {
197                    self.end_part(s);
198                }
199            }
200        }
201        while !self.done {
202            match self.step()? {
203                Got::Start => {
204                    self.in_part = true;
205                    return Ok(Some(PartReader { inner: self }));
206                }
207                Got::Done => self.done = true,
208                _ => {}
209            }
210        }
211        Ok(None)
212    }
213
214    fn end_part(&mut self, summary: PartSummary) {
215        self.in_part = false;
216        self.last = Some(summary);
217        self.pending.clear();
218    }
219}
220
221/// One part of a [`MultipartReader`]: read it like a file.
222///
223/// `Read` reports a parse error as `std::io::Error` with kind `InvalidData`
224/// and the `FormError` as its source.
225pub struct PartReader<'r, R: Read> {
226    inner: &'r mut MultipartReader<R>,
227}
228
229impl<R: Read> fmt::Debug for PartReader<'_, R> {
230    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
231        f.debug_struct("PartReader")
232            .field("head", &self.inner.head)
233            .finish_non_exhaustive()
234    }
235}
236
237impl<R: Read> PartReader<'_, R> {
238    /// The part's head, copied when the part started.
239    #[must_use]
240    pub fn head(&self) -> &OwnedPartHead {
241        &self.inner.head
242    }
243
244    /// Reads the rest of the part and checks it as text (its charset, else
245    /// the `_charset_` field before it, else UTF-8), bounded by
246    /// `max_value_bytes` for file parts too.
247    ///
248    /// # Errors
249    ///
250    /// Every multipart error; `Encoding { InvalidUtf8 | UnsupportedCharset }`;
251    /// `Limit { ValueLength }`.
252    pub fn text(&mut self) -> Result<String, FormError> {
253        let own = self.inner.head.view().charset();
254        let is_charset = self.inner.head.view().name() == "_charset_";
255        let cap = self.inner.parser.limits().value() as usize;
256        let mut out = Vec::new();
257        let read = (|| {
258            while let Some(chunk) = self.next_chunk()? {
259                if out.len().saturating_add(chunk.len()) > cap {
260                    return Err(FormError::limit(
261                        crate::Limit::ValueLength,
262                        Location::in_part(
263                            Some(self.inner.head.view().index()),
264                            self.inner.parser.bytes_seen(),
265                        ),
266                    ));
267                }
268                crate::wipe::append_wiping(&mut out, chunk, 4096, cap);
269            }
270            Ok(())
271        })();
272        if let Err(e) = read {
273            crate::wipe::wipe_vec(&mut out);
274            return Err(e);
275        }
276        let at = Location::in_part(
277            Some(self.inner.head.view().index()),
278            self.inner.parser.bytes_seen(),
279        );
280        let readable = match own {
281            Charset::Utf8 | Charset::Ascii => true,
282            Charset::Other => false,
283            Charset::Unspecified => self.inner.declared.as_deref().is_none_or(|l| {
284                l.eq_ignore_ascii_case(b"utf-8") || l.eq_ignore_ascii_case(b"us-ascii")
285            }),
286        };
287        if !readable {
288            crate::wipe::wipe_vec(&mut out);
289            return Err(FormError::encoding(EncodingError::UnsupportedCharset, at));
290        }
291        if is_charset {
292            self.inner.declared = Some(out.clone());
293        }
294        String::from_utf8(out).map_err(|e| {
295            let mut bytes = e.into_bytes();
296            crate::wipe::wipe_vec(&mut bytes);
297            FormError::encoding(EncodingError::InvalidUtf8, at)
298        })
299    }
300
301    /// Discards the rest of the part and returns its data length.
302    ///
303    /// # Errors
304    ///
305    /// Every multipart error.
306    pub fn skip(&mut self) -> Result<u64, FormError> {
307        self.inner.parser.skip_part();
308        self.inner.pending.clear();
309        while self.inner.in_part {
310            if let Got::End(s) = self.inner.step()? {
311                self.inner.end_part(s);
312            }
313        }
314        Ok(self.inner.last.map_or(0, |s| s.bytes))
315    }
316
317    /// Streams a file part into `store`.
318    ///
319    /// # Errors
320    ///
321    /// Every multipart error; `Io`, `Store`, `Limit { Disk }`.
322    #[cfg(feature = "fs")]
323    #[cfg_attr(docsrs, doc(cfg(feature = "fs")))]
324    pub fn save(&mut self, store: &crate::Store) -> Result<crate::StoredFile, FormError> {
325        use crate::hook::Sink;
326        // One request, one disk limit: the sink starts from what earlier
327        // saves of this body already put on disk.
328        let mut sink = store
329            .sink(self.inner.parser.limits())
330            .with_used_disk(self.inner.disk_used);
331        let head = self.inner.head.clone();
332        sink.open(&head.view())?;
333        let r = (|| {
334            while let Some(chunk) = self.next_chunk()? {
335                sink.write(chunk)?;
336            }
337            sink.close()
338        })();
339        if r.is_err() {
340            sink.abort();
341        } else {
342            self.inner.disk_used = sink.used_disk();
343        }
344        r
345    }
346
347    /// The next piece of data, or `None` at the end of the part.
348    fn next_chunk(&mut self) -> Result<Option<&[u8]>, FormError> {
349        let inner = &mut *self.inner;
350        if inner.pending_at < inner.pending.len() {
351            let chunk = inner.pending.get(inner.pending_at..).unwrap_or(&[]);
352            inner.pending_at = inner.pending.len();
353            return Ok(Some(chunk));
354        }
355        while inner.in_part {
356            match inner.step()? {
357                Got::Data => {
358                    inner.pending_at = inner.pending.len();
359                    return Ok(Some(&inner.pending));
360                }
361                Got::End(s) => inner.end_part(s),
362                _ => {}
363            }
364        }
365        Ok(None)
366    }
367}
368
369impl<R: Read> Read for PartReader<'_, R> {
370    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
371        let inner = &mut *self.inner;
372        loop {
373            if inner.pending_at < inner.pending.len() {
374                let src = inner.pending.get(inner.pending_at..).unwrap_or(&[]);
375                let n = src.len().min(out.len());
376                if let (Some(dst), Some(src)) = (out.get_mut(..n), src.get(..n)) {
377                    dst.copy_from_slice(src);
378                }
379                inner.pending_at = inner.pending_at.saturating_add(n);
380                return Ok(n);
381            }
382            if !inner.in_part {
383                return Ok(0);
384            }
385            match inner.step() {
386                Ok(Got::End(s)) => inner.end_part(s),
387                Ok(_) => {}
388                Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidData, e)),
389            }
390        }
391    }
392}
form_net/io.rs

form_net/
io.rs
