form_net/multipart/
driver.rs

1//! The streaming driver: one implementation of the hook order over the
2//! parser, an inspector, and a sink.
3
4use alloc::vec::Vec;
5use core::fmt;
6
7use crate::error::{EncodingError, FormError, Location};
8use crate::hook::{Inspector, Sink, Verdict};
9use crate::limits::EmptyFilePolicy;
10use crate::multipart::owned_head::OwnedPartHead;
11use crate::multipart::parser::{Event, MultipartParser, PartHead, PartKind, PartSummary};
12use crate::wipe::WipeVec;
13
14/// What completed during a [`Driver::push`]. `Debug` shows a field's name
15/// and value length, never the value.
16#[non_exhaustive]
17pub enum Progress<'d, F> {
18    /// A text field ended: its name and value.
19    Field {
20        /// The field name.
21        name: &'d str,
22        /// The value bytes.
23        value: &'d [u8],
24    },
25    /// A file part ended: the sink's output.
26    File(F),
27    /// Every byte given was consumed; give more.
28    NeedInput,
29    /// The form is complete.
30    End,
31}
32
33impl<F> fmt::Debug for Progress<'_, F> {
34    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35        match self {
36            Progress::Field { name, value } => f
37                .debug_struct("Field")
38                .field("name", name)
39                .field("value_len", &value.len())
40                .finish(),
41            Progress::File(_) => f.write_str("File(..)"),
42            Progress::NeedInput => f.write_str("NeedInput"),
43            Progress::End => f.write_str("End"),
44        }
45    }
46}
47
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
49enum Part {
50    Idle,
51    Field {
52        skip: bool,
53        /// Its charset (its own, else the `_charset_` before it) is one
54        /// form-net reads; fixed at the part's start.
55        readable: bool,
56    },
57    File {
58        open: bool,
59        deferred: bool,
60        sniffing: bool,
61        skip: bool,
62    },
63}
64
65/// Everything but the parser, so its fields can be borrowed while an event
66/// borrows the parser.
67struct State<S> {
68    sink: S,
69    head: OwnedPartHead,
70    text: WipeVec,
71    sniff: WipeVec,
72    sniff_len: usize,
73    value_cap: usize,
74    skip_empty: bool,
75    part: Part,
76}
77
78/// What a part handler asks of the driver once the event is released.
79#[derive(Clone, Copy, PartialEq, Eq)]
80enum Next {
81    Continue,
82    SkipPart,
83}
84
85fn refused(v: Verdict, at: Location) -> Result<bool, FormError> {
86    match v {
87        Verdict::Refuse(refusal) => Err(FormError::Refused { refusal, at }),
88        Verdict::Skip => Ok(true),
89        _ => Ok(false),
90    }
91}
92
93impl<S: Sink> State<S> {
94    fn start_file<I: Inspector>(
95        &mut self,
96        head: &PartHead<'_>,
97        inspector: &mut I,
98        at: Location,
99    ) -> Result<Next, FormError> {
100        if refused(inspector.part_start(head), at)? {
101            self.part = Part::File {
102                open: false,
103                deferred: false,
104                sniffing: false,
105                skip: true,
106            };
107            return Ok(Next::SkipPart);
108        }
109        // The sink opens on the first byte the inspector has accepted (or at
110        // the end of an empty file), never before `sniff`: a refused or
111        // skipped part touches no storage at all.
112        crate::wipe::wipe(&mut self.sniff);
113        self.sniff.clear();
114        self.part = Part::File {
115            open: false,
116            deferred: false,
117            sniffing: self.sniff_len > 0,
118            skip: false,
119        };
120        Ok(Next::Continue)
121    }
122
123    fn on_start<I: Inspector>(
124        &mut self,
125        head: &PartHead<'_>,
126        declared_ok: bool,
127        inspector: &mut I,
128        at: Location,
129    ) -> Result<Next, FormError> {
130        self.head.copy_from(head);
131        match head.kind() {
132            PartKind::File => {
133                let empty = head.filename().is_some_and(|f| f.is_empty());
134                if empty && self.skip_empty {
135                    self.part = Part::File {
136                        open: false,
137                        deferred: true,
138                        sniffing: false,
139                        skip: false,
140                    };
141                    return Ok(Next::Continue);
142                }
143                self.start_file(head, inspector, at)
144            }
145            _ => {
146                crate::wipe::wipe(&mut self.text);
147                self.text.clear();
148                let readable = head.charset().readable(declared_ok);
149                if refused(inspector.part_start(head), at)? {
150                    self.part = Part::Field {
151                        skip: true,
152                        readable,
153                    };
154                    return Ok(Next::SkipPart);
155                }
156                self.part = Part::Field {
157                    skip: false,
158                    readable,
159                };
160                Ok(Next::Continue)
161            }
162        }
163    }
164
165    /// Opens the sink for the current file if it is not open yet.
166    fn ensure_open(&mut self) -> Result<(), FormError> {
167        if let Part::File { open: false, .. } = self.part {
168            self.sink.open(&self.head.view())?;
169            if let Part::File { open, .. } = &mut self.part {
170                *open = true;
171            }
172        }
173        Ok(())
174    }
175
176    /// Aborts the sink if this part opened it, and marks the part skipped.
177    fn skip_file(&mut self) {
178        if let Part::File { open: true, .. } = self.part {
179            self.sink.abort();
180        }
181        self.part = Part::File {
182            open: false,
183            deferred: false,
184            sniffing: false,
185            skip: true,
186        };
187    }
188
189    /// Passes file bytes to the inspector and the sink.
190    fn file_bytes<I: Inspector>(
191        &mut self,
192        index: u32,
193        bytes: &[u8],
194        inspector: &mut I,
195        at: Location,
196    ) -> Result<Next, FormError> {
197        if bytes.is_empty() {
198            return Ok(Next::Continue);
199        }
200        if refused(inspector.data(index, bytes), at)? {
201            self.skip_file();
202            return Ok(Next::SkipPart);
203        }
204        self.ensure_open()?;
205        self.sink.write(bytes).map(|()| Next::Continue)
206    }
207
208    /// Hands the sniff window to the inspector, then on as data.
209    fn flush_sniff<I: Inspector>(
210        &mut self,
211        index: u32,
212        inspector: &mut I,
213        at: Location,
214    ) -> Result<Next, FormError> {
215        if let Part::File { sniffing, .. } = &mut self.part {
216            *sniffing = false;
217        }
218        let window = core::mem::take(&mut self.sniff);
219        let verdict = inspector.sniff(&self.head.view(), &window);
220        let result = match refused(verdict, at) {
221            Err(e) => Err(e),
222            Ok(true) => {
223                self.skip_file();
224                Ok(Next::SkipPart)
225            }
226            Ok(false) => self.file_bytes(index, &window, inspector, at),
227        };
228        self.sniff = window;
229        crate::wipe::wipe(&mut self.sniff);
230        self.sniff.clear();
231        result
232    }
233
234    fn on_data<I: Inspector>(
235        &mut self,
236        bytes: &[u8],
237        inspector: &mut I,
238        at: Location,
239    ) -> Result<Next, FormError> {
240        let index = self.head.view().index();
241        match self.part {
242            Part::Field {
243                skip: false,
244                readable,
245            } => {
246                if refused(inspector.data(index, bytes), at)? {
247                    self.part = Part::Field {
248                        skip: true,
249                        readable,
250                    };
251                    return Ok(Next::SkipPart);
252                }
253                crate::wipe::append_wiping(&mut self.text, bytes, 256, self.value_cap);
254                Ok(Next::Continue)
255            }
256            Part::File { deferred: true, .. } => {
257                let head = self.head.clone();
258                if self.start_file(&head.view(), inspector, at)? == Next::SkipPart {
259                    return Ok(Next::SkipPart);
260                }
261                self.on_data(bytes, inspector, at)
262            }
263            Part::File {
264                skip: false,
265                sniffing: true,
266                ..
267            } => {
268                let room = self.sniff_len.saturating_sub(self.sniff.len());
269                let (head, rest) = bytes.split_at(room.min(bytes.len()));
270                self.sniff.extend_from_slice(head);
271                if self.sniff.len() < self.sniff_len {
272                    return Ok(Next::Continue);
273                }
274                if self.flush_sniff(index, inspector, at)? == Next::SkipPart {
275                    return Ok(Next::SkipPart);
276                }
277                self.file_bytes(index, rest, inspector, at)
278            }
279            Part::File { skip: false, .. } => self.file_bytes(index, bytes, inspector, at),
280            _ => Ok(Next::Continue),
281        }
282    }
283}
284
285/// What a step of the driver produced, without borrows.
286enum Done<F> {
287    Field,
288    File(F),
289    NeedInput,
290    End,
291}
292
293/// The one streaming implementation of the hook order: parser, inspector,
294/// sink, the text buffer, the sniff window, and the empty-file delay.
295///
296/// It takes its limits and policy from the parser. Because a `PartHead`
297/// cannot outlive the `parse` call that produced it, the driver copies each
298/// head once into an owned head and hands that copy to the delayed calls.
299///
300/// # Examples
301///
302/// ```
303/// use form_net::{Boundary, Driver, Limits, MemorySink, MultipartParser, NoInspector, Policy, Progress};
304///
305/// let parser = MultipartParser::new(Boundary::new(b"X")?, &Limits::DEFAULT, &Policy::DEFAULT)?;
306/// let mut driver = Driver::new(parser, MemorySink::new());
307/// let body = b"--X\r\nContent-Disposition: form-data; name=\"a\"\r\n\r\n1\r\n--X--";
308/// let (consumed, progress) = driver.push(body, true, &mut NoInspector)?;
309/// assert!(matches!(progress, Progress::Field { name: "a", value: b"1" }));
310/// let (_, progress) = driver.push(&body[consumed..], true, &mut NoInspector)?;
311/// assert!(matches!(progress, Progress::End));
312/// # Ok::<(), form_net::FormError>(())
313/// ```
314pub struct Driver<B: AsMut<[u8]> + AsRef<[u8]>, S: Sink> {
315    parser: MultipartParser<B>,
316    state: State<S>,
317    ended: bool,
318    failed: bool,
319}
320
321impl<B: AsMut<[u8]> + AsRef<[u8]>, S: Sink> fmt::Debug for Driver<B, S> {
322    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
323        f.debug_struct("Driver")
324            .field("parser", &self.parser)
325            .field("ended", &self.ended)
326            .finish_non_exhaustive()
327    }
328}
329
330impl<B: AsMut<[u8]> + AsRef<[u8]>, S: Sink> Driver<B, S> {
331    /// A driver over `parser` writing files to `sink`.
332    #[must_use]
333    pub fn new(parser: MultipartParser<B>, sink: S) -> Self {
334        let limits = *parser.limits();
335        let policy = *parser.policy();
336        let sniff_len = limits.sniff() as usize;
337        Driver {
338            state: State {
339                sink,
340                head: OwnedPartHead::with_capacity(limits.multipart_buffer_len(&policy)),
341                text: WipeVec::default(),
342                sniff: WipeVec(Vec::with_capacity(sniff_len)),
343                sniff_len,
344                value_cap: limits.value() as usize,
345                skip_empty: policy.empty_files == EmptyFilePolicy::Skip,
346                part: Part::Idle,
347            },
348            parser,
349            ended: false,
350            failed: false,
351        }
352    }
353
354    /// The parser.
355    #[must_use]
356    pub const fn parser(&self) -> &MultipartParser<B> {
357        &self.parser
358    }
359
360    /// The parser and the sink.
361    pub fn into_parts(self) -> (MultipartParser<B>, S) {
362        (self.parser, self.state.sink)
363    }
364
365    /// Feeds input and returns how much was consumed and what completed: a
366    /// text field, a closed file (the sink's output), a request for more
367    /// input, or the end. `eof = true` says `input` ends the body.
368    ///
369    /// # Errors
370    ///
371    /// Every multipart error, `Refused`, and the sink's errors. On error the
372    /// sink has already been aborted and `inspector.failed` called; later
373    /// calls return `Poisoned`.
374    pub fn push<'d, I: Inspector>(
375        &'d mut self,
376        input: &[u8],
377        eof: bool,
378        inspector: &mut I,
379    ) -> Result<(usize, Progress<'d, S::Output>), FormError> {
380        if self.failed {
381            return Err(FormError::Poisoned);
382        }
383        match self.run(input, eof, inspector) {
384            Ok((consumed, done)) => Ok((
385                consumed,
386                match done {
387                    Done::Field => Progress::Field {
388                        name: self.state.head_name(),
389                        value: &self.state.text,
390                    },
391                    Done::File(f) => Progress::File(f),
392                    Done::NeedInput => Progress::NeedInput,
393                    Done::End => Progress::End,
394                },
395            )),
396            Err(e) => {
397                self.failed = true;
398                self.state.sink.abort();
399                inspector.failed(&e);
400                Err(e)
401            }
402        }
403    }
404
405    fn at(&self) -> Location {
406        Location::in_part(
407            self.parser.parts_seen().checked_sub(1),
408            self.parser.bytes_seen(),
409        )
410    }
411
412    fn run<I: Inspector>(
413        &mut self,
414        input: &[u8],
415        eof: bool,
416        inspector: &mut I,
417    ) -> Result<(usize, Done<S::Output>), FormError> {
418        let mut pos = 0usize;
419        loop {
420            let rest = input.get(pos..).unwrap_or(&[]);
421            let seen = self.parser.bytes_seen();
422            let parts = self.parser.parts_seen();
423            // The `_charset_` declared before the event (a part start comes
424            // in a call of its own, after the part before it ended).
425            let declared_ok =
426                crate::multipart::parser::declared_readable(self.parser.declared_charset());
427            let step = self.parser.parse(rest, eof)?;
428            pos = pos.saturating_add(step.consumed);
429            let index = match &step.event {
430                Event::PartStart(head) => Some(head.index()),
431                _ => parts.checked_sub(1),
432            };
433            let at = Location::in_part(index, seen.saturating_add(step.consumed as u64));
434            let next = match step.event {
435                Event::PartStart(head) => self.state.on_start(&head, declared_ok, inspector, at)?,
436                Event::Data(bytes) => self.state.on_data(bytes, inspector, at)?,
437                Event::PartEnd(summary) => {
438                    if let Some(done) = self.on_end(summary, inspector)? {
439                        return Ok((pos, done));
440                    }
441                    Next::Continue
442                }
443                Event::End => {
444                    if !self.ended {
445                        self.ended = true;
446                        if let Verdict::Refuse(refusal) = inspector.end() {
447                            return Err(FormError::Refused { refusal, at });
448                        }
449                    }
450                    return Ok((pos, Done::End));
451                }
452                _ => return Ok((pos, Done::NeedInput)),
453            };
454            if next == Next::SkipPart {
455                self.parser.skip_part();
456            }
457        }
458    }
459
460    fn on_end<I: Inspector>(
461        &mut self,
462        summary: PartSummary,
463        inspector: &mut I,
464    ) -> Result<Option<Done<S::Output>>, FormError> {
465        let at = self.at();
466        let part = core::mem::replace(&mut self.state.part, Part::Idle);
467        let skipped = summary.skipped
468            || matches!(
469                part,
470                Part::Field { skip: true, .. } | Part::File { skip: true, .. }
471            );
472        let summary = PartSummary { skipped, ..summary };
473        match part {
474            Part::File { deferred: true, .. } => Ok(None),
475            Part::File {
476                skip: false,
477                sniffing,
478                ..
479            } => {
480                // The flush and the open read the part's state: give it back
481                // until the sink is closed.
482                self.state.part = part;
483                let flushed = if sniffing {
484                    self.state.flush_sniff(summary.index, inspector, at)
485                } else {
486                    Ok(Next::Continue)
487                };
488                let opened = match flushed {
489                    Ok(Next::Continue) => self.state.ensure_open().map(|()| true),
490                    Ok(Next::SkipPart) => Ok(false),
491                    Err(e) => Err(e),
492                };
493                self.state.part = Part::Idle;
494                if !opened? {
495                    let summary = PartSummary {
496                        skipped: true,
497                        ..summary
498                    };
499                    refused(inspector.part_end(&summary), at)?;
500                    return Ok(None);
501                }
502                let output = self.state.sink.close()?;
503                refused(inspector.part_end(&summary), at)?;
504                Ok(Some(Done::File(output)))
505            }
506            Part::Field {
507                skip: false,
508                readable,
509            } => {
510                // Checked as text before the inspector sees it as a field,
511                // as the buffered parser checks it: the charset fixed at the
512                // part's start, then the bytes as UTF-8.
513                let text_at = Location::in_part(Some(summary.index), at.offset);
514                if !readable {
515                    return Err(FormError::encoding(
516                        EncodingError::UnsupportedCharset,
517                        text_at,
518                    ));
519                }
520                if core::str::from_utf8(&self.state.text).is_err() {
521                    return Err(FormError::encoding(EncodingError::InvalidUtf8, text_at));
522                }
523                let name = self.state.head.view().name();
524                // The name is copied out of the head: the inspector borrows
525                // the text buffer.
526                let skip = refused(inspector.field(name, &self.state.text), at)?;
527                let summary = PartSummary {
528                    skipped: skip,
529                    ..summary
530                };
531                refused(inspector.part_end(&summary), at)?;
532                if skip {
533                    crate::wipe::wipe(&mut self.state.text);
534                    return Ok(None);
535                }
536                Ok(Some(Done::Field))
537            }
538            _ => {
539                refused(inspector.part_end(&summary), at)?;
540                Ok(None)
541            }
542        }
543    }
544}
545
546impl<S> State<S> {
547    fn head_name(&self) -> &str {
548        self.head.view().name()
549    }
550}
551
552impl<B: AsMut<[u8]> + AsRef<[u8]>, S: Sink> Driver<B, S> {
553    /// Whether a push failed (the driver is poisoned).
554    #[cfg(feature = "fs")]
555    pub(crate) const fn has_failed(&self) -> bool {
556        self.failed
557    }
558}
form_net/multipart/driver.rs

form_net/multipart/
driver.rs
