uart_fpl/
parser.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Streaming frame parser with preamble resynchronization, header integrity verification,
6//! and bounded memory allocation.
7
8use zerocopy::FromBytes;
9use zerocopy::byteorder::big_endian::U32 as BeU32;
10
11use crate::frame::{
12    CHECKSUM_LEN, Frame, FrameType, HEADER_CHECKSUM_OFFSET, HEADER_LEN, MAX_PAYLOAD_SIZE,
13    SYNC_WORD, SYNC_WORD_LEN, WireHeader, crc8, crc32,
14};
15
16/// Maximum capacity for the internal parser buffer (64 KiB) to prevent unbounded memory growth.
17pub(crate) const MAX_BUFFER_CAPACITY: usize = 64 * 1024;
18
19/// Internal buffer compaction threshold in bytes before draining consumed bytes.
20const COMPACTION_THRESHOLD_BYTES: usize = 4096;
21
22/// Streaming parser for continuous byte streams from serial interfaces.
23#[derive(Debug)]
24pub struct FrameParser {
25    buffer: Vec<u8>,
26    cursor: usize,
27    checksum_errors: u64,
28}
29
30impl Default for FrameParser {
31    fn default() -> Self {
32        Self::new()
33    }
34}
35
36impl FrameParser {
37    /// Creates a new streaming frame parser.
38    pub fn new() -> Self {
39        Self { buffer: Vec::new(), cursor: 0, checksum_errors: 0 }
40    }
41
42    /// Resets the parser state, clearing all buffered bytes and resetting cursor.
43    pub fn reset(&mut self) {
44        self.buffer.clear();
45        self.cursor = 0;
46        self.checksum_errors = 0;
47    }
48
49    /// Returns true if there are no unconsumed bytes in the buffer.
50    pub fn is_empty(&self) -> bool {
51        self.unconsumed_len() == 0
52    }
53
54    /// Drains and returns all unconsumed bytes from the internal buffer, resetting the parser.
55    pub fn take_unconsumed(&mut self) -> Vec<u8> {
56        let remaining = self.buffer[self.cursor..].to_vec();
57        self.buffer.clear();
58        self.cursor = 0;
59        remaining
60    }
61
62    /// Returns the cumulative number of CRC32 / header checksum mismatch errors encountered.
63    pub fn checksum_errors(&self) -> u64 {
64        self.checksum_errors
65    }
66
67    /// Returns a slice of the unparsed bytes currently remaining in the buffer.
68    #[inline]
69    pub fn unconsumed(&self) -> &[u8] {
70        &self.buffer[self.cursor..]
71    }
72
73    /// Returns the current number of unparsed bytes waiting in the internal stream buffer.
74    #[inline]
75    pub fn unconsumed_len(&self) -> usize {
76        self.buffer.len() - self.cursor
77    }
78
79    #[inline]
80    fn consume(&mut self, count: usize) {
81        self.cursor += count;
82        if self.cursor >= self.buffer.len() {
83            self.buffer.clear();
84            self.cursor = 0;
85        }
86    }
87
88    /// Appends incoming byte chunks to the internal streaming buffer.
89    ///
90    /// Memory consumption is strictly bounded to [`MAX_BUFFER_CAPACITY`]; if incoming data
91    /// exceeds this limit, the oldest un-parsed bytes are discarded to prevent memory exhaustion.
92    pub fn feed(&mut self, data: &[u8]) {
93        let unconsumed_len = self.unconsumed_len();
94        let total_len = unconsumed_len.saturating_add(data.len());
95
96        // Compact consumed bytes so cursor is at 0 before evaluating overflow trimming.
97        if self.cursor > 0
98            && (self.cursor >= COMPACTION_THRESHOLD_BYTES
99                || self.cursor >= self.buffer.len() / 2
100                || self.buffer.len().saturating_add(data.len()) > MAX_BUFFER_CAPACITY)
101        {
102            self.buffer.drain(..self.cursor);
103            self.cursor = 0;
104        }
105
106        if total_len > MAX_BUFFER_CAPACITY {
107            let overflow = total_len - MAX_BUFFER_CAPACITY;
108            let current_len = self.buffer.len();
109            if overflow >= current_len {
110                self.buffer.clear();
111                let data_start = data.len().saturating_sub(MAX_BUFFER_CAPACITY);
112                self.buffer.extend_from_slice(&data[data_start..]);
113                return;
114            } else {
115                self.buffer.drain(..overflow);
116            }
117        }
118        self.buffer.extend_from_slice(data);
119    }
120
121    /// Parses and extracts the next complete, validated protocol frame from the stream buffer.
122    ///
123    /// Returns `None` if insufficient bytes are available or if awaiting further stream chunks.
124    pub fn next_frame(&mut self) -> Option<Frame> {
125        loop {
126            let unconsumed = self.unconsumed();
127            if unconsumed.len() < SYNC_WORD_LEN {
128                return None;
129            }
130
131            let Some(sync_pos) = unconsumed.windows(SYNC_WORD_LEN).position(|w| w == SYNC_WORD)
132            else {
133                let to_consume = if unconsumed.last() == Some(&SYNC_WORD[0]) {
134                    unconsumed.len() - 1
135                } else {
136                    unconsumed.len()
137                };
138                self.consume(to_consume);
139                return None;
140            };
141            if sync_pos > 0 {
142                self.consume(sync_pos);
143            }
144
145            let unconsumed = self.unconsumed();
146            if unconsumed.len() < HEADER_LEN {
147                return None;
148            }
149
150            // Verify Header Checksum (HdrChk at byte offset 12)
151            let header_fields = &unconsumed[SYNC_WORD_LEN..HEADER_CHECKSUM_OFFSET];
152            let received_hdr_chk = unconsumed[HEADER_CHECKSUM_OFFSET];
153            let calculated_hdr_chk = crc8(header_fields);
154
155            if received_hdr_chk != calculated_hdr_chk {
156                log::debug!(
157                    "FrameParser: header checksum mismatch! Calculated: {calculated_hdr_chk:#04x}, received: {received_hdr_chk:#04x}"
158                );
159                self.checksum_errors += 1;
160                self.consume(SYNC_WORD_LEN);
161                continue;
162            }
163
164            let (wire_header, _) = WireHeader::read_from_prefix(unconsumed)
165                .expect("unconsumed length verified >= HEADER_LEN");
166            let session_id = wire_header.session_id.get();
167            let channel_id = wire_header.channel_id.get();
168            let seq = wire_header.seq;
169            let frame_type = FrameType::from(wire_header.frame_type);
170            let length = wire_header.payload_len.get() as usize;
171
172            if length > MAX_PAYLOAD_SIZE {
173                log::debug!("FrameParser: invalid payload length in header: {length}");
174                self.consume(SYNC_WORD_LEN);
175                continue;
176            }
177
178            let total_needed = HEADER_LEN + length + CHECKSUM_LEN;
179            if unconsumed.len() < total_needed {
180                return None;
181            }
182
183            let (received_cksum_be, _) =
184                BeU32::read_from_prefix(&unconsumed[HEADER_LEN + length..])
185                    .expect("unconsumed length verified >= total_needed");
186            let received_cksum = received_cksum_be.get();
187
188            let calculated_cksum = crc32(&unconsumed[SYNC_WORD_LEN..HEADER_LEN + length]);
189
190            if received_cksum == calculated_cksum {
191                let is_negotiate =
192                    frame_type == FrameType::NegotiateReq || frame_type == FrameType::NegotiateResp;
193                if is_negotiate && channel_id != 0 {
194                    log::debug!(
195                        "FrameParser: dropping handshake frame {frame_type:?} on non-zero channel {channel_id}"
196                    );
197                    self.consume(total_needed);
198                    continue;
199                }
200
201                let payload = unconsumed[HEADER_LEN..HEADER_LEN + length].to_vec();
202                self.consume(total_needed);
203                return Some(Frame { session_id, channel_id, seq, frame_type, payload });
204            } else {
205                log::debug!(
206                    "FrameParser: CRC-32 mismatch for frame type {frame_type:?}, length={length}! Calculated: {calculated_cksum:#010x}, received: {received_cksum:#010x}"
207                );
208                self.checksum_errors += 1;
209                self.consume(SYNC_WORD_LEN);
210            }
211        }
212    }
213}
214
215impl Iterator for FrameParser {
216    type Item = Frame;
217
218    fn next(&mut self) -> Option<Self::Item> {
219        self.next_frame()
220    }
221}
222
223#[cfg(test)]
224mod tests {
225    use super::*;
226    use crate::frame::{FrameType, PAYLOAD_LEN_OFFSET, encode_frame};
227
228    fn assert_frame_eq(
229        frame: &Frame,
230        session_id: u32,
231        channel_id: u16,
232        seq: u8,
233        frame_type: FrameType,
234        payload: &[u8],
235    ) {
236        assert_eq!(frame.session_id, session_id);
237        assert_eq!(frame.channel_id, channel_id);
238        assert_eq!(frame.seq, seq);
239        assert_eq!(frame.frame_type, frame_type);
240        assert_eq!(frame.payload, payload);
241    }
242
243    #[fuchsia::test]
244    fn test_frame_parser_happy_path() {
245        let payload = b"ping_message";
246        let frame = encode_frame(100, 2, 5, FrameType::Data, payload).unwrap();
247        let mut parser = FrameParser::new();
248        parser.feed(&frame);
249
250        let parsed = parser.next_frame().expect("expected parsed frame");
251        assert_frame_eq(&parsed, 100, 2, 5, FrameType::Data, payload);
252        assert_eq!(parser.next_frame(), None);
253    }
254
255    #[fuchsia::test]
256    fn test_frame_parser_partial_reads() {
257        let payload = b"fragmented";
258        let frame = encode_frame(100, 2, 5, FrameType::Data, payload).unwrap();
259        let mut parser = FrameParser::new();
260
261        for chunk in frame.chunks(3) {
262            parser.feed(chunk);
263        }
264
265        let parsed = parser.next_frame().expect("expected frame after fragments");
266        assert_frame_eq(&parsed, 100, 2, 5, FrameType::Data, payload);
267        assert_eq!(parser.next_frame(), None);
268    }
269
270    #[fuchsia::test]
271    fn test_frame_parser_sync_recovery() {
272        let payload = b"recovered";
273        let valid_frame = encode_frame(1, 1, 0, FrameType::Data, payload).unwrap();
274
275        let mut corrupted_stream = vec![0xFF, 0xAA, 0x00, 0x55, 0x12];
276        corrupted_stream.extend_from_slice(&valid_frame);
277
278        let mut parser = FrameParser::new();
279        parser.feed(&corrupted_stream);
280
281        let parsed = parser.next_frame().expect("parser must parse valid frame");
282        assert_eq!(parsed.session_id, 1);
283        assert_eq!(parsed.payload, b"recovered");
284    }
285
286    #[fuchsia::test]
287    fn test_frame_parser_bad_checksum() {
288        let payload = b"corrupted";
289        let mut frame = encode_frame(1, 1, 0, FrameType::Data, payload).unwrap();
290        let len = frame.len();
291        frame[len - 1] ^= 0xFF; // Flip bit in trailing CRC-32
292
293        let mut parser = FrameParser::new();
294        parser.feed(&frame);
295        assert_eq!(parser.next_frame(), None);
296        assert_eq!(parser.checksum_errors(), 1);
297    }
298
299    #[fuchsia::test]
300    fn test_frame_parser_bad_header_checksum() {
301        let payload = b"header_corrupt";
302        let mut frame = encode_frame(1, 1, 0, FrameType::Data, payload).unwrap();
303        frame[HEADER_CHECKSUM_OFFSET] ^= 0xFF; // Flip bit in header checksum
304
305        let mut parser = FrameParser::new();
306        parser.feed(&frame);
307        assert_eq!(parser.next_frame(), None);
308        assert_eq!(parser.checksum_errors(), 1);
309    }
310
311    #[fuchsia::test]
312    fn test_frame_parser_checksum_errors_counter() {
313        let mut parser = FrameParser::new();
314        assert_eq!(parser.checksum_errors(), 0);
315
316        let mut bad_frame = encode_frame(1, 1, 0, FrameType::Data, b"test_crc").unwrap();
317        let len = bad_frame.len();
318        bad_frame[len - 1] ^= 0xFF;
319
320        parser.feed(&bad_frame);
321        assert_eq!(parser.next_frame(), None);
322        assert_eq!(parser.checksum_errors(), 1);
323    }
324
325    #[fuchsia::test]
326    fn test_frame_parser_invalid_length() {
327        let mut frame = vec![0xAA, 0x55];
328        let mut header_fields = [0u8; 10];
329        header_fields[0..4].copy_from_slice(&1u32.to_be_bytes());
330        header_fields[4..6].copy_from_slice(&1u16.to_be_bytes());
331        header_fields[6] = 0;
332        header_fields[7] = u8::from(FrameType::Data);
333        header_fields[8..10].copy_from_slice(&2000u16.to_be_bytes()); // Length > 1024
334
335        let hdr_chk = crc8(&header_fields);
336        frame.extend_from_slice(&header_fields);
337        frame.push(hdr_chk);
338        frame.extend_from_slice(&[0u8; 100]);
339
340        let mut parser = FrameParser::new();
341        parser.feed(&frame);
342        assert_eq!(parser.next_frame(), None);
343    }
344
345    #[fuchsia::test]
346    fn test_frame_parser_bounded_buffer_capacity() {
347        let mut parser = FrameParser::new();
348        let massive_chunk = vec![0x42u8; MAX_BUFFER_CAPACITY + 1024];
349        parser.feed(&massive_chunk);
350        assert!(parser.unconsumed_len() <= MAX_BUFFER_CAPACITY);
351
352        let payload = b"recovered";
353        let valid_frame = encode_frame(1, 1, 0, FrameType::Data, payload).unwrap();
354        parser.feed(&valid_frame);
355
356        let parsed = parser.next_frame().expect("parser must parse valid frame");
357        assert_eq!(parsed.payload, b"recovered");
358    }
359
360    #[fuchsia::test]
361    fn test_frame_parser_recovers_from_corrupted_length_via_header_checksum() {
362        let mut parser = FrameParser::new();
363
364        let mut raw_hdr = [0u8; 10];
365        raw_hdr[0..4].copy_from_slice(&1u32.to_be_bytes());
366        raw_hdr[4..6].copy_from_slice(&1u16.to_be_bytes());
367        raw_hdr[6] = 0;
368        raw_hdr[7] = u8::from(FrameType::Data);
369        raw_hdr[8..10].copy_from_slice(&10u16.to_be_bytes());
370        let hdr_chk = crc8(&raw_hdr);
371
372        let mut corrupted = vec![0xAA, 0x55];
373        corrupted.extend_from_slice(&raw_hdr);
374        corrupted.push(hdr_chk);
375        corrupted.extend_from_slice(b"0123456789");
376
377        // Corrupt length field in header
378        corrupted[PAYLOAD_LEN_OFFSET + 1] ^= 0x01;
379
380        parser.feed(&corrupted);
381        assert_eq!(parser.next_frame(), None);
382        assert_eq!(parser.checksum_errors(), 1);
383
384        // Feed subsequent valid TYPE_DATA frame
385        let valid_frame = encode_frame(1, 1, 1, FrameType::Data, b"hello world payload").unwrap();
386        parser.feed(&valid_frame);
387
388        let parsed = parser.next_frame().expect("Parser must recover after corrupted header");
389        assert_eq!(parsed.session_id, 1);
390        assert_eq!(parsed.payload, b"hello world payload");
391        assert_eq!(parsed.seq, 1);
392
393        // Subsequent valid ACK frame
394        let ack_frame = encode_frame(1, 1, 1, FrameType::Ack, &[]).unwrap();
395        parser.feed(&ack_frame);
396        let parsed_ack = parser.next_frame().expect("Subsequent ACK must parse");
397        assert_eq!(parsed_ack.frame_type, FrameType::Ack);
398
399        // Subsequent valid CLOSE frame
400        let close_frame = encode_frame(1, 1, 2, FrameType::Close, &[]).unwrap();
401        parser.feed(&close_frame);
402        let parsed_close = parser.next_frame().expect("Subsequent CLOSE must parse");
403        assert_eq!(parsed_close.frame_type, FrameType::Close);
404    }
405
406    #[fuchsia::test]
407    fn test_negotiate_req_dropped_on_non_zero_channel() {
408        let mut parser = FrameParser::new();
409        let invalid_req = encode_frame(1, 1, 0, FrameType::NegotiateReq, b"invalid").unwrap();
410        parser.feed(&invalid_req);
411        assert_eq!(parser.next_frame(), None);
412
413        let valid_req = encode_frame(1, 0, 0, FrameType::NegotiateReq, b"valid").unwrap();
414        parser.feed(&valid_req);
415        let parsed = parser.next_frame().expect("NegotiateReq on channel 0 accepted");
416        assert_eq!(parsed.channel_id, 0);
417        assert_eq!(parsed.frame_type, FrameType::NegotiateReq);
418        assert_eq!(parsed.payload, b"valid");
419    }
420
421    #[fuchsia::test]
422    fn test_frame_parser_state_management() {
423        let mut parser = FrameParser::new();
424        assert!(parser.is_empty());
425        assert_eq!(parser.unconsumed_len(), 0);
426
427        parser.feed(b"partial_stream_data");
428        assert!(!parser.is_empty());
429        assert_eq!(parser.unconsumed_len(), 19);
430
431        let unconsumed = parser.take_unconsumed();
432        assert_eq!(unconsumed, b"partial_stream_data");
433        assert!(parser.is_empty());
434        assert_eq!(parser.unconsumed_len(), 0);
435
436        parser.feed(b"new_data");
437        assert!(!parser.is_empty());
438        parser.reset();
439        assert!(parser.is_empty());
440        assert_eq!(parser.unconsumed_len(), 0);
441    }
442
443    #[fuchsia::test]
444    fn test_frame_parser_iterator() {
445        let frame1 = encode_frame(1, 1, 0, FrameType::Data, b"first").unwrap();
446        let frame2 = encode_frame(1, 1, 1, FrameType::Data, b"second").unwrap();
447
448        let mut parser = FrameParser::new();
449        parser.feed(&frame1);
450        parser.feed(&frame2);
451
452        let frames: Vec<Frame> = parser.collect();
453        assert_eq!(frames.len(), 2);
454        assert_eq!(frames[0].payload, b"first");
455        assert_eq!(frames[1].payload, b"second");
456    }
457
458    #[fuchsia::test]
459    fn test_differential_chunking() {
460        let mut stream = Vec::new();
461        for i in 0..50u8 {
462            let payload = format!("chunk-msg-{}", i).into_bytes();
463            let frame = encode_frame(42, 1, i, FrameType::Data, &payload).unwrap();
464            stream.extend_from_slice(&frame);
465            if i % 5 == 0 {
466                stream.extend_from_slice(&[0x11, 0x22, 0x33, 0x44]);
467            }
468        }
469
470        let mut parser_mono = FrameParser::new();
471        parser_mono.feed(&stream);
472        let mono_frames: Vec<Frame> = parser_mono.collect();
473
474        let mut parser_frag = FrameParser::new();
475        let mut offset = 0;
476        let chunk_sizes = [1, 3, 7, 2, 5, 4, 11, 1];
477        let mut size_idx = 0;
478        while offset < stream.len() {
479            let sz = chunk_sizes[size_idx % chunk_sizes.len()];
480            size_idx += 1;
481            let end = std::cmp::min(offset + sz, stream.len());
482            parser_frag.feed(&stream[offset..end]);
483            offset = end;
484        }
485        let frag_frames: Vec<Frame> = parser_frag.collect();
486
487        assert_eq!(mono_frames.len(), 50);
488        assert_eq!(mono_frames, frag_frames);
489    }
490}
uart_fpl/parser.rs

uart_fpl/
parser.rs
