netget/utils/
line_reader.rs

1//! One bounded, `\n`-terminated read, shared by the line-oriented text protocols.
2//!
3//! `tokio::io::AsyncBufReadExt::read_line` grows its `String` until it finds a newline and
4//! caps nothing. A peer that connects and streams printable bytes without ever sending `\n`
5//! therefore makes the server allocate without bound — a one-connection out-of-memory, from an
6//! unauthenticated socket, before any model call. That defect was found in `ftp`, `irc` and
7//! `smtp` and fixed three separate times, then found again in `imap`, `nntp` and `pop3`.
8//!
9//! Three copies of a `fill_buf`/`consume` loop is how the control-character filters ended up
10//! with six different semantics before `utils::sanitize` existed, so this is one
11//! implementation rather than a sixth. Each protocol still owns its own `MAX_*` constant and
12//! its own refusal wording — the bound is a protocol decision, the reading is not.
13//!
14//! **The cap is on what is buffered, not on what is consumed.** A line longer than the cap is
15//! reported as [`BoundedLine::TooLong`] with the offending bytes left unconsumed, because there
16//! is no resynchronisation point: the peer is mid-line and the next byte is not the start of a
17//! command. Every caller answers in its own vocabulary and closes.
18//!
19//! Decoding is lossy. `read_line` returns `ErrorKind::InvalidData` for any non-UTF-8 byte and
20//! kills the session on it, which is wrong for protocols that advertise 8-bit transports
21//! (`smtp`'s `8BITMIME`, `imap`'s literals). These bytes become an event payload and a log
22//! line; nothing here is re-emitted on the wire, so a lossy decode loses nothing a peer can
23//! observe.
24
25use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader};
26
27/// Read one complete byte-delimited frame, preserving binary payloads and enforcing
28/// the cap before allocation. EOF with a fragment is an error. Keep this future alive
29/// across unrelated commands: like `read_until`, cancellation discards its partial frame.
30pub async fn read_bounded_delimited<R: AsyncRead + Unpin>(
31    reader: &mut BufReader<R>,
32    delimiter: u8,
33    max_len: usize,
34) -> std::io::Result<Option<Vec<u8>>> {
35    let mut frame = Vec::new();
36    loop {
37        let available = reader.fill_buf().await?;
38        if available.is_empty() {
39            return if frame.is_empty() {
40                Ok(None)
41            } else {
42                Err(std::io::Error::new(
43                    std::io::ErrorKind::UnexpectedEof,
44                    "partial delimited frame",
45                ))
46            };
47        }
48        let end = available.iter().position(|&b| b == delimiter);
49        let take = end.map_or(available.len(), |n| n + 1);
50        if take > max_len.saturating_sub(frame.len()) {
51            return Err(std::io::Error::new(
52                std::io::ErrorKind::InvalidData,
53                "delimited frame exceeds cap",
54            ));
55        }
56        frame.extend_from_slice(&available[..take]);
57        reader.consume(take);
58        if end.is_some() {
59            return Ok(Some(frame));
60        }
61    }
62}
63
64/// The outcome of one bounded line read.
65#[derive(Debug)]
66pub enum BoundedLine {
67    /// A complete line. The trailing `\n` (and `\r`, if any) is still attached; callers trim.
68    Line(String),
69    /// The peer closed, or sent a partial line and then closed. Half a command is not a
70    /// command, so a trailing fragment at EOF is reported here rather than as a `Line`.
71    Eof,
72    /// `max_len` bytes went by with no `\n` in them.
73    TooLong,
74}
75
76/// Read one `\n`-terminated line, buffering at most `max_len` bytes.
77///
78/// Returns the line and the number of bytes consumed from the socket, so a caller can feed
79/// `update_connection_stats` exactly as `read_line`'s return value allowed.
80///
81/// `max_len` bounds the buffer the peer can make this function allocate. It is compared
82/// *before* each `extend_from_slice`, so the high-water mark is `max_len`, not `max_len` plus
83/// however much happened to be in the `BufReader` when the limit was crossed.
84pub async fn read_bounded_line<R>(
85    reader: &mut BufReader<R>,
86    max_len: usize,
87) -> std::io::Result<(BoundedLine, usize)>
88where
89    R: AsyncRead + Unpin,
90{
91    let mut buf: Vec<u8> = Vec::new();
92    let mut consumed = 0usize;
93
94    loop {
95        // `fill_buf` borrows the reader, so decide what to take and drop the borrow before
96        // calling `consume`.
97        let (newline_at, available_len) = {
98            let available = reader.fill_buf().await?;
99            if available.is_empty() {
100                return Ok((BoundedLine::Eof, consumed));
101            }
102            (available.iter().position(|&b| b == b'\n'), available.len())
103        };
104
105        match newline_at {
106            Some(idx) => {
107                if buf.len() + idx + 1 > max_len {
108                    return Ok((BoundedLine::TooLong, consumed));
109                }
110                let take = idx + 1;
111                buf.extend_from_slice(&reader.buffer()[..take]);
112                reader.consume(take);
113                consumed += take;
114                return Ok((
115                    BoundedLine::Line(String::from_utf8_lossy(&buf).into_owned()),
116                    consumed,
117                ));
118            }
119            None => {
120                // No newline in what is buffered. Refuse before growing past the cap: the
121                // check has to come first, or the peer decides the allocation.
122                if buf.len() + available_len > max_len {
123                    return Ok((BoundedLine::TooLong, consumed));
124                }
125                buf.extend_from_slice(&reader.buffer()[..available_len]);
126                reader.consume(available_len);
127                consumed += available_len;
128            }
129        }
130    }
131}
netget/utils/line_reader.rs

netget/utils/
line_reader.rs
