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

form_net/multipart/
driver.rs
