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

form_net/
io.rs
