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

form_net/multipart/
driver.rs
