hpke_http/
sse.rs

1//! Byte-level SSE block splitting. Field parsing belongs to the caller.
2
3use crate::Error;
4
5/// One bounded partial SSE block. Line endings on the wire are LF.
6#[derive(Debug)]
7pub struct SseSplitter {
8    pending: Vec<u8>,
9    line_has_data: bool,
10    skip_lf: bool,
11    max_block_len: usize,
12    closed: bool,
13}
14
15impl SseSplitter {
16    /// Start a splitter with the same limit used for a response DATA record.
17    #[must_use]
18    pub fn new(max_block_len: usize) -> Self {
19        Self {
20            pending: Vec::new(),
21            line_has_data: false,
22            skip_lf: false,
23            max_block_len,
24            closed: false,
25        }
26    }
27
28    /// Read at most one complete block and report the number of input bytes used.
29    /// Pass the unread suffix to the next call.
30    ///
31    /// # Errors
32    /// Returns [`Error::LimitExceeded`] if one block exceeds its bound, or
33    /// [`Error::InvalidConfiguration`] if the splitter is closed.
34    pub fn feed(&mut self, input: &[u8]) -> Result<(usize, Option<Vec<u8>>), Error> {
35        if self.closed {
36            return Err(Error::InvalidConfiguration);
37        }
38        for (index, &byte) in input.iter().enumerate() {
39            if self.skip_lf {
40                self.skip_lf = false;
41                if byte == b'\n' {
42                    continue;
43                }
44            }
45            if byte == b'\r' || byte == b'\n' {
46                if byte == b'\r' {
47                    self.skip_lf = true;
48                }
49                self.push(b'\n')?;
50                if !self.line_has_data {
51                    return Ok((index + 1, Some(std::mem::take(&mut self.pending))));
52                }
53                self.line_has_data = false;
54            } else {
55                self.push(byte)?;
56                self.line_has_data = true;
57            }
58        }
59        Ok((input.len(), None))
60    }
61
62    /// Discard an incomplete final block and close the splitter.
63    pub fn finish(&mut self) {
64        self.pending.clear();
65        self.closed = true;
66    }
67
68    fn push(&mut self, byte: u8) -> Result<(), Error> {
69        if self.pending.len() >= self.max_block_len {
70            self.closed = true;
71            self.pending.clear();
72            return Err(Error::LimitExceeded);
73        }
74        self.pending.push(byte);
75        Ok(())
76    }
77}
78
79pub(crate) fn validate_block(block: &[u8], max_block_len: usize) -> Result<(), Error> {
80    if block.len() > max_block_len {
81        return Err(Error::LimitExceeded);
82    }
83    if block.is_empty() || block.contains(&b'\r') || !block.ends_with(b"\n") {
84        return Err(Error::MalformedEnvelope);
85    }
86    let mut line_start = 0;
87    for (index, &byte) in block.iter().enumerate() {
88        if byte == b'\n' {
89            if index == line_start {
90                return if index + 1 == block.len() {
91                    Ok(())
92                } else {
93                    Err(Error::MalformedEnvelope)
94                };
95            }
96            line_start = index + 1;
97        }
98    }
99    Err(Error::MalformedEnvelope)
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105
106    #[test]
107    fn splits_each_line_ending_without_losing_bytes() -> Result<(), Error> {
108        let mut splitter = SseSplitter::new(64);
109        let mut output = Vec::new();
110        for byte in b"data:\r\n\rid: 1\n\n" {
111            let (used, block) = splitter.feed(&[*byte])?;
112            assert_eq!(used, 1);
113            if let Some(block) = block {
114                output.push(block);
115            }
116        }
117        assert_eq!(output, vec![b"data:\n\n".to_vec(), b"id: 1\n\n".to_vec()]);
118        Ok(())
119    }
120
121    #[test]
122    fn keeps_raw_non_line_bytes_and_drops_tail() -> Result<(), Error> {
123        let mut splitter = SseSplitter::new(64);
124        let input = b"\xef\xbb\xbfdata:\xff\n\npartial";
125        let (used, block) = splitter.feed(input)?;
126        assert_eq!(used, 11);
127        assert_eq!(
128            block.ok_or(Error::MalformedEnvelope)?,
129            b"\xef\xbb\xbfdata:\xff\n\n"
130        );
131        splitter.feed(&input[used..])?;
132        splitter.finish();
133        assert!(splitter.feed(b"\n").is_err());
134        Ok(())
135    }
136
137    #[test]
138    fn rejects_non_single_blocks() {
139        assert!(validate_block(b"\n", 1).is_ok());
140        for value in [b"data:x\n".as_slice(), b"data:x\n\n\n", b"data:x\r\n\r\n"] {
141            assert_eq!(validate_block(value, 100), Err(Error::MalformedEnvelope));
142        }
143    }
144
145    #[test]
146    fn exact_block_limit_passes_and_one_more_byte_fails() -> Result<(), Error> {
147        let mut exact = SseSplitter::new(3);
148        assert_eq!(exact.feed(b"x\n\n")?, (3, Some(b"x\n\n".to_vec())));
149        let mut short = SseSplitter::new(2);
150        assert_eq!(short.feed(b"x\n\n"), Err(Error::LimitExceeded));
151        assert!(short.feed(b"\n").is_err());
152        Ok(())
153    }
154}
hpke_http/sse.rs

hpke_http/
sse.rs
