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    ///
30    /// # Errors
31    /// Returns [`Error::LimitExceeded`] if one block exceeds its bound, or
32    /// [`Error::InvalidConfiguration`] if the splitter is closed.
33    pub fn feed(&mut self, input: &[u8]) -> Result<(usize, Option<Vec<u8>>), Error> {
34        if self.closed {
35            return Err(Error::InvalidConfiguration);
36        }
37        for (index, &byte) in input.iter().enumerate() {
38            if self.skip_lf {
39                self.skip_lf = false;
40                if byte == b'\n' {
41                    continue;
42                }
43            }
44            if byte == b'\r' || byte == b'\n' {
45                if byte == b'\r' {
46                    self.skip_lf = true;
47                }
48                self.push(b'\n')?;
49                if !self.line_has_data {
50                    return Ok((index + 1, Some(std::mem::take(&mut self.pending))));
51                }
52                self.line_has_data = false;
53            } else {
54                self.push(byte)?;
55                self.line_has_data = true;
56            }
57        }
58        Ok((input.len(), None))
59    }
60
61    /// Discard an incomplete final block and close the splitter.
62    pub fn finish(&mut self) {
63        self.pending.clear();
64        self.closed = true;
65    }
66
67    fn push(&mut self, byte: u8) -> Result<(), Error> {
68        if self.pending.len() >= self.max_block_len {
69            self.closed = true;
70            self.pending.clear();
71            return Err(Error::LimitExceeded);
72        }
73        self.pending.push(byte);
74        Ok(())
75    }
76}
77
78pub(crate) fn validate_block(block: &[u8], max_block_len: usize) -> Result<(), Error> {
79    if block.len() > max_block_len {
80        return Err(Error::LimitExceeded);
81    }
82    if block.is_empty() || block.contains(&b'\r') || !block.ends_with(b"\n") {
83        return Err(Error::MalformedEnvelope);
84    }
85    let mut line_start = 0;
86    for (index, &byte) in block.iter().enumerate() {
87        if byte == b'\n' {
88            if index == line_start {
89                return if index + 1 == block.len() {
90                    Ok(())
91                } else {
92                    Err(Error::MalformedEnvelope)
93                };
94            }
95            line_start = index + 1;
96        }
97    }
98    Err(Error::MalformedEnvelope)
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104
105    #[test]
106    fn splits_each_line_ending_without_losing_bytes() -> Result<(), Error> {
107        let mut splitter = SseSplitter::new(64);
108        let mut output = Vec::new();
109        for byte in b"data:\r\n\rid: 1\n\n" {
110            let (used, block) = splitter.feed(&[*byte])?;
111            assert_eq!(used, 1);
112            if let Some(block) = block {
113                output.push(block);
114            }
115        }
116        assert_eq!(output, vec![b"data:\n\n".to_vec(), b"id: 1\n\n".to_vec()]);
117        Ok(())
118    }
119
120    #[test]
121    fn keeps_raw_non_line_bytes_and_drops_tail() -> Result<(), Error> {
122        let mut splitter = SseSplitter::new(64);
123        let input = b"\xef\xbb\xbfdata:\xff\n\npartial";
124        let (used, block) = splitter.feed(input)?;
125        assert_eq!(used, 11);
126        assert_eq!(
127            block.ok_or(Error::MalformedEnvelope)?,
128            b"\xef\xbb\xbfdata:\xff\n\n"
129        );
130        splitter.feed(&input[used..])?;
131        splitter.finish();
132        assert!(splitter.feed(b"\n").is_err());
133        Ok(())
134    }
135
136    #[test]
137    fn rejects_non_single_blocks() {
138        assert!(validate_block(b"\n", 1).is_ok());
139        for value in [b"data:x\n".as_slice(), b"data:x\n\n\n", b"data:x\r\n\r\n"] {
140            assert_eq!(validate_block(value, 100), Err(Error::MalformedEnvelope));
141        }
142    }
143
144    #[test]
145    fn exact_block_limit_passes_and_one_more_byte_fails() -> Result<(), Error> {
146        let mut exact = SseSplitter::new(3);
147        assert_eq!(exact.feed(b"x\n\n")?, (3, Some(b"x\n\n".to_vec())));
148        let mut short = SseSplitter::new(2);
149        assert_eq!(short.feed(b"x\n\n"), Err(Error::LimitExceeded));
150        assert!(short.feed(b"\n").is_err());
151        Ok(())
152    }
153}
hpke_http/sse.rs

hpke_http/
sse.rs
