mail_parser/mailbox/
mbox.rs

1/*
2 * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
3 *
4 * SPDX-License-Identifier: Apache-2.0 OR MIT
5 */
6
7//! Mbox files.
8
9use crate::DateTime;
10use memchr::{memchr, memmem::Finder, memrchr};
11use std::io::{self, BufRead};
12
13const SEPARATOR: &[u8] = b"From ";
14const QUOTE: u8 = b'>';
15const CONTENTS_CAPACITY: usize = 1024;
16
17/// Iterates the messages of an mbox read from a buffered reader.
18///
19/// A message starts at each line beginning with `From ` (the separator line,
20/// which gives [`Message::from`] and [`Message::internal_date`]) and runs to
21/// the next one. Lines before the first separator are skipped. Lines quoted
22/// as `>From `, `>>From `, ... lose one `>`, as described in the
23/// [qmail mbox specification](http://qmail.org/qmail-manual-html/man5/mbox.html).
24/// Separators are found with `memchr`, and the contents are copied from the
25/// reader's buffer in bulk.
26///
27/// ```
28/// use mail_parser::mailbox::mbox::MessageIterator;
29///
30/// let mbox = b"From ann@example.com Sat Jan  3 01:05:34 1996\n\
31/// Subject: hello\n\n>From the start\n";
32/// let message = MessageIterator::new(&mbox[..])
33///     .next()
34///     .expect("one message")
35///     .expect("readable mbox");
36/// assert_eq!(message.from(), "ann@example.com");
37/// assert_eq!(message.internal_date(), 820_631_134);
38/// assert_eq!(message.contents(), b"Subject: hello\n\nFrom the start\n");
39/// ```
40#[derive(Debug)]
41pub struct MessageIterator<T> {
42    reader: T,
43    message: Option<Message>,
44    line: Vec<u8>,
45    complete: usize,
46    separator: Finder<'static>,
47}
48
49/// A message of an mbox: its contents and the metadata of its separator
50/// line.
51#[derive(Debug, PartialEq, Eq, Clone, PartialOrd, Ord)]
52pub struct Message {
53    internal_date: u64,
54    from: String,
55    contents: Vec<u8>,
56}
57
58impl<T> MessageIterator<T>
59where
60    T: BufRead,
61{
62    /// Iterates over the messages of the mbox read from `reader`.
63    pub fn new(reader: T) -> MessageIterator<T> {
64        MessageIterator {
65            reader,
66            message: None,
67            line: Vec::new(),
68            complete: 0,
69            separator: Finder::new(SEPARATOR),
70        }
71    }
72}
73
74impl<T> Iterator for MessageIterator<T>
75where
76    T: BufRead,
77{
78    type Item = io::Result<Message>;
79
80    fn next(&mut self) -> Option<Self::Item> {
81        loop {
82            if self.line.is_empty() {
83                let buf = match self.reader.fill_buf() {
84                    Ok(buf) => buf,
85                    Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
86                    Err(err) => return Some(Err(err)),
87                };
88                if buf.is_empty() {
89                    return self.message.take().map(Ok);
90                }
91                if self.complete == 0 || self.complete > buf.len() {
92                    self.complete = memrchr(b'\n', buf).map_or(0, |last| last + 1);
93                }
94                if let Some(lines) = buf.get(..self.complete).filter(|lines| !lines.is_empty()) {
95                    let (consumed, finished) = split(&self.separator, &mut self.message, lines);
96                    self.reader.consume(consumed);
97                    self.complete -= consumed;
98                    if finished.is_some() {
99                        return finished.map(Ok);
100                    }
101                    continue;
102                }
103            }
104            if let Err(err) = self.reader.read_until(b'\n', &mut self.line) {
105                return Some(Err(err));
106            }
107            let (_, finished) = split(&self.separator, &mut self.message, &self.line);
108            self.line.clear();
109            if finished.is_some() {
110                return finished.map(Ok);
111            }
112        }
113    }
114}
115
116fn split(
117    separator: &Finder<'_>,
118    message: &mut Option<Message>,
119    lines: &[u8],
120) -> (usize, Option<Message>) {
121    let mut copied = 0;
122    let mut offset = 0;
123    while let Some(hit) = lines
124        .get(offset..)
125        .and_then(|rest| separator.find(rest))
126        .map(|found| offset + found)
127    {
128        offset = hit + SEPARATOR.len();
129        let before = lines.get(..hit).unwrap_or_default();
130        let quotes = before
131            .iter()
132            .rev()
133            .take_while(|&&byte| byte == QUOTE)
134            .count();
135        let Some((head, _)) = before.split_at_checked(before.len() - quotes) else {
136            continue;
137        };
138        if !matches!(head.last(), None | Some(b'\n')) {
139            continue;
140        }
141        let line_start = head.len();
142        append(message, lines.get(copied..line_start));
143        if quotes > 0 {
144            copied = line_start + 1;
145            continue;
146        }
147        let line_end = lines
148            .get(hit..)
149            .and_then(|rest| memchr(b'\n', rest))
150            .map_or(lines.len(), |end| hit + end + 1);
151        let finished = message.replace(Message::new(
152            lines.get(line_start..line_end).unwrap_or_default(),
153        ));
154        if finished.is_some() {
155            return (line_end, finished);
156        }
157        copied = line_end;
158        offset = line_end;
159    }
160    append(message, lines.get(copied..));
161    (lines.len(), None)
162}
163
164fn append(message: &mut Option<Message>, bytes: Option<&[u8]>) {
165    if let (Some(message), Some(bytes)) = (message, bytes) {
166        message.contents.extend_from_slice(bytes);
167    }
168}
169
170fn internal_date(date: &str) -> u64 {
171    let mut dt = DateTime {
172        year: u16::MAX,
173        month: u8::MAX,
174        day: u8::MAX,
175        hour: u8::MAX,
176        minute: u8::MAX,
177        second: u8::MAX,
178        tz_before_gmt: false,
179        tz_hour: 0,
180        tz_minute: 0,
181    };
182    for (pos, part) in date.split_whitespace().enumerate() {
183        match pos {
184            1 => {
185                dt.month = hashify::map_ignore_case!(part.as_bytes(), u8,
186                    "jan" => 1,
187                    "feb" => 2,
188                    "mar" => 3,
189                    "apr" => 4,
190                    "may" => 5,
191                    "jun" => 6,
192                    "jul" => 7,
193                    "aug" => 8,
194                    "sep" => 9,
195                    "oct" => 10,
196                    "nov" => 11,
197                    "dec" => 12,
198                )
199                .copied()
200                .unwrap_or(u8::MAX);
201            }
202            2 => dt.day = part.parse().unwrap_or(u8::MAX),
203            3 => {
204                let fields = [&mut dt.hour, &mut dt.minute, &mut dt.second];
205                for (field, value) in fields.into_iter().zip(part.split(':')) {
206                    *field = value.parse().unwrap_or(u8::MAX);
207                }
208            }
209            4 => dt.year = part.parse().unwrap_or(u16::MAX),
210            _ => (),
211        }
212    }
213    if dt.is_valid() {
214        dt.to_timestamp() as u64
215    } else {
216        0
217    }
218}
219
220impl Message {
221    fn new(line: &[u8]) -> Self {
222        let (internal_date, from) = line
223            .strip_prefix(SEPARATOR)
224            .and_then(|rest| std::str::from_utf8(rest).ok())
225            .and_then(|rest| rest.split_once(' '))
226            .map_or((0, String::new()), |(from, date)| {
227                (internal_date(date), from.trim().to_string())
228            });
229        Message {
230            internal_date,
231            from,
232            contents: Vec::with_capacity(CONTENTS_CAPACITY),
233        }
234    }
235
236    /// The date of the separator line, in seconds since the UNIX epoch; 0
237    /// when it is missing or does not parse.
238    pub fn internal_date(&self) -> u64 {
239        self.internal_date
240    }
241
242    /// The sender address of the separator line; empty when missing.
243    pub fn from(&self) -> &str {
244        &self.from
245    }
246
247    /// The message, ready for [`crate::MessageParser::parse`].
248    pub fn contents(&self) -> &[u8] {
249        &self.contents
250    }
251
252    /// Takes the message contents.
253    pub fn unwrap_contents(self) -> Vec<u8> {
254        self.contents
255    }
256}
257
258#[cfg(test)]
259mod tests {
260    use super::{Message, MessageIterator};
261    use crate::scan::tests::Rng;
262    use std::io::{self, BufRead, BufReader, Read};
263
264    const MBOX: &[u8] = include_bytes!("../../resources/mbox/sample.mbox");
265
266    fn expected() -> Vec<Message> {
267        vec![
268            Message {
269                internal_date: 820631134,
270                from: "god@heaven.af.mil".to_string(),
271                contents: b"Message 1\n\n".to_vec(),
272            },
273            Message {
274                internal_date: 1027453163,
275                from: "cras@irccrew.org".to_string(),
276                contents: b"Message 2\n\n".to_vec(),
277            },
278            Message {
279                internal_date: 1028640874,
280                from: "test@test.com".to_string(),
281                contents: b"Message 3\nFrom hello\n>From world\n>>From test\n\n".to_vec(),
282            },
283            Message {
284                internal_date: 1516030200,
285                from: "other@domain.com".to_string(),
286                contents: b"Message 4\n> From\n>F\n".to_vec(),
287            },
288        ]
289    }
290
291    fn read_all(reader: impl BufRead) -> Vec<Message> {
292        MessageIterator::new(reader)
293            .map(|message| message.expect("message is readable"))
294            .collect()
295    }
296
297    fn unquoted(line: &[u8]) -> &[u8] {
298        let quotes = line.iter().take_while(|&&byte| byte == b'>').count();
299        match (line.get(quotes..), line.get(1..)) {
300            (Some(rest), Some(unquoted)) if quotes > 0 && rest.starts_with(b"From ") => unquoted,
301            _ => line,
302        }
303    }
304
305    fn reference(mbox: &[u8]) -> Vec<Message> {
306        let mut messages = Vec::new();
307        let mut current: Option<Message> = None;
308        for line in mbox.split_inclusive(|&byte| byte == b'\n') {
309            if line.starts_with(b"From ") {
310                messages.extend(current.replace(Message::new(line)));
311            } else if let Some(message) = &mut current {
312                message.contents.extend_from_slice(unquoted(line));
313            }
314        }
315        messages.extend(current);
316        messages
317    }
318
319    struct Chunked<'x> {
320        data: &'x [u8],
321        sizes: Vec<usize>,
322        next: usize,
323    }
324
325    impl Read for Chunked<'_> {
326        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
327            let size = self.sizes.get(self.next).copied().unwrap_or(1).max(1);
328            self.next = (self.next + 1) % self.sizes.len().max(1);
329            let len = size.min(buf.len()).min(self.data.len());
330            let (head, tail) = self.data.split_at(len);
331            buf.get_mut(..len)
332                .expect("length fits")
333                .copy_from_slice(head);
334            self.data = tail;
335            Ok(len)
336        }
337    }
338
339    #[test]
340    fn parse_mbox() {
341        assert_eq!(read_all(MBOX), expected());
342        assert_eq!(reference(MBOX), expected());
343    }
344
345    #[test]
346    fn buffer_boundaries() {
347        for capacity in 1..=64 {
348            assert_eq!(
349                read_all(BufReader::with_capacity(capacity, MBOX)),
350                expected(),
351                "capacity {capacity}"
352            );
353        }
354    }
355
356    #[test]
357    fn same_as_line_reader() {
358        const PIECES: [&[u8]; 14] = [
359            b"From ",
360            b">",
361            b">>",
362            b"From a@b Sat Jan  3 01:05:34 1996",
363            b"\n",
364            b"\r\n",
365            b"text",
366            b" ",
367            b"From",
368            b"F",
369            b"rom ",
370            b">From x",
371            b"x@y",
372            b"From \xff Tue Jul 23 19:39:23 2002",
373        ];
374        let mut rng = Rng(0x6d62_6f78_0000_0001);
375        for round in 0..4_000 {
376            let mbox: Vec<u8> = (0..rng.below(40))
377                .flat_map(|_| {
378                    PIECES
379                        .get(rng.below(PIECES.len()))
380                        .copied()
381                        .unwrap_or_default()
382                })
383                .copied()
384                .collect();
385            let expected = reference(&mbox);
386            assert_eq!(read_all(&mbox[..]), expected, "round {round}: {mbox:?}");
387            let capacity = 1 + rng.below(12);
388            assert_eq!(
389                read_all(BufReader::with_capacity(capacity, &mbox[..])),
390                expected,
391                "round {round} capacity {capacity}: {mbox:?}"
392            );
393            let sizes = (0..4).map(|_| 1 + rng.below(9)).collect();
394            let chunked = Chunked {
395                data: &mbox,
396                sizes,
397                next: 0,
398            };
399            assert_eq!(
400                read_all(BufReader::with_capacity(16, chunked)),
401                expected,
402                "round {round} chunked: {mbox:?}"
403            );
404        }
405    }
406
407    #[test]
408    fn long_lines_and_missing_final_newline() {
409        let long = "x".repeat(10_000);
410        let mbox = format!(
411            "{long}\nFrom a@b Sat Jan  3 01:05:34 1996\n{long}\n>From {long}\nFrom c@d {long}"
412        );
413        let expected = reference(mbox.as_bytes());
414        assert_eq!(expected.len(), 2);
415        assert_eq!(read_all(mbox.as_bytes()), expected);
416        assert_eq!(
417            read_all(BufReader::with_capacity(64, mbox.as_bytes())),
418            expected
419        );
420    }
421}
mail_parser/mailbox/mbox.rs

mail_parser/mailbox/
mbox.rs
