bgpkit_parser/parser/iters/
raw.rs

1/*!
2The RawMrtRecord Iterator module provides functionality for iterating over raw MRT records
3from a BGP data source. This iterator is responsible for:
4
5* Reading and parsing raw MRT records sequentially from an input stream
6* Handling parsing errors and warnings appropriately
7* Providing a clean interface for processing MRT records one at a time
8
9The iterator implements error recovery strategies, allowing it to skip malformed records
10when possible and continue processing the remaining data. It also supports configurable
11warning messages and core dump generation for debugging purposes.
12*/
13
14use crate::parser::iters::{handle_record_parse_error, record_matches_filters};
15use crate::parser::mrt::mrt_record::raw_record_uses_zebra_compat;
16use crate::{chunk_mrt_record, BgpkitParser, Elementor, Filter, MrtRecord, RawMrtRecord};
17use std::io::Read;
18
19pub struct RawRecordIterator<R> {
20    parser: BgpkitParser<R>,
21    count: u64,
22}
23
24impl<R> RawRecordIterator<R> {
25    pub(crate) fn new(parser: BgpkitParser<R>) -> Self {
26        RawRecordIterator { parser, count: 0 }
27    }
28}
29
30impl<R: Read> Iterator for RawRecordIterator<R> {
31    type Item = RawMrtRecord;
32
33    fn next(&mut self) -> Option<RawMrtRecord> {
34        // Text-dump parsers have no MRT-record representation; short-circuit.
35        if self.parser.text_dump_iter.is_some() {
36            return None;
37        }
38        self.count += 1;
39        loop {
40            match chunk_mrt_record(&mut self.parser.reader) {
41                Ok(raw_record) => return Some(raw_record),
42                Err(e) => {
43                    if !handle_record_parse_error(&mut self.parser, e.error, e.bytes) {
44                        return None;
45                    }
46                }
47            }
48        }
49    }
50}
51
52/// Iterator over raw MRT records with record-level filter semantics
53/// applied.
54///
55/// Behaves like [`RawRecordIterator`] for chunking, but only yields the
56/// raw bytes of records that pass the parser's filters under the same
57/// semantics as [`RecordIterator`](crate::RecordIterator): filters match
58/// on the elem projection, so records that produce no elems (KEEPALIVE,
59/// OPEN, NOTIFICATION, state changes) are dropped while filters are
60/// active, and the `PeerIndexTable` always passes through so RIB peer
61/// resolution keeps working.
62///
63/// The yielded [`RawMrtRecord`]s carry the *original* wire bytes — no
64/// re-encoding is involved — which is what byte-exact outputs (hex dumps,
65/// re-dissection) require. Every record body is parsed once inside the
66/// iterator (for filter matching and for uniform error diagnostics), and
67/// the parsed record is yielded alongside so consumers never parse the
68/// bytes twice. Parse failures go through the same variant-aware policy
69/// as [`RecordIterator`](crate::RecordIterator) — warnings honor
70/// `disable_warnings()`, core dumps are written where applicable, and
71/// fatal classes stop the iterator — so this pipeline cannot continue
72/// past errors the record pipeline would stop for.
73///
74/// Shortened Zebra BGP4MP records are detected and warned about once,
75/// matching the record iterator's data-quality diagnostic.
76pub struct FilteredRawRecordIterator<R> {
77    inner: RawRecordIterator<R>,
78    elementor: Elementor,
79    filters: Vec<Filter>,
80}
81
82impl<R> FilteredRawRecordIterator<R> {
83    pub(crate) fn new(parser: BgpkitParser<R>) -> Self {
84        let filters = parser.filters.clone();
85        FilteredRawRecordIterator {
86            inner: RawRecordIterator::new(parser),
87            elementor: Elementor::new(),
88            filters,
89        }
90    }
91
92    /// Zebra detection delegates to the parser's own warn-once state, so
93    /// `disable_warnings()` is honored and a parser that already warned
94    /// before conversion does not warn again.
95    fn warn_zebra_once(&mut self, raw: &RawMrtRecord) {
96        if raw_record_uses_zebra_compat(raw) {
97            self.inner.parser.warn_zebra_compat_once();
98        }
99    }
100}
101
102impl<R: Read> Iterator for FilteredRawRecordIterator<R> {
103    type Item = (RawMrtRecord, MrtRecord);
104
105    fn next(&mut self) -> Option<Self::Item> {
106        loop {
107            let raw = self.inner.next()?;
108            self.warn_zebra_once(&raw);
109            // Body-parse failures share the record iterator's
110            // variant-aware error policy (warnings vs. errors, core
111            // dumps, stop-after-dump) — on every path, filtered or not.
112            let record = match raw.clone().parse() {
113                Ok(record) => record,
114                Err(error) => {
115                    if !handle_record_parse_error(
116                        &mut self.inner.parser,
117                        error,
118                        Some(raw.raw_bytes().to_vec()),
119                    ) {
120                        return None;
121                    }
122                    continue;
123                }
124            };
125            if self.filters.is_empty()
126                || record_matches_filters(&record, &self.filters, &mut self.elementor)
127            {
128                return Some((raw, record));
129            }
130        }
131    }
132}
133
134#[cfg(test)]
135mod filtered_tests {
136    use super::*;
137    use crate::models::*;
138    use std::io::Cursor;
139
140    fn bgp4mp_record(timestamp: u32, bgp_message: BgpMessage) -> Vec<u8> {
141        let body_of = |update: &BgpUpdateMessage| update.encode(AsnLength::Bits32).unwrap();
142        let body = match &bgp_message {
143            BgpMessage::Update(update) => body_of(update),
144            BgpMessage::KeepAlive => Vec::new().into(),
145            _ => unreachable!("test builds updates and keepalives only"),
146        };
147        let mut bgp = vec![0xFF; 16];
148        bgp.extend_from_slice(&((19 + body.len()) as u16).to_be_bytes());
149        bgp.push(bgp_message.msg_type() as u8);
150        bgp.extend_from_slice(&body);
151
152        let mut mrt_body = Vec::new();
153        mrt_body.extend_from_slice(&64496u32.to_be_bytes());
154        mrt_body.extend_from_slice(&64497u32.to_be_bytes());
155        mrt_body.extend_from_slice(&0u16.to_be_bytes());
156        mrt_body.extend_from_slice(&1u16.to_be_bytes());
157        mrt_body.extend_from_slice(&[192, 0, 2, 1]);
158        mrt_body.extend_from_slice(&[192, 0, 2, 2]);
159        mrt_body.extend_from_slice(&bgp);
160
161        let mut wire = Vec::new();
162        wire.extend_from_slice(&timestamp.to_be_bytes());
163        wire.extend_from_slice(&(EntryType::BGP4MP as u16).to_be_bytes());
164        wire.extend_from_slice(&(Bgp4MpType::MessageAs4 as u16).to_be_bytes());
165        wire.extend_from_slice(&(mrt_body.len() as u32).to_be_bytes());
166        wire.extend_from_slice(&mrt_body);
167        wire
168    }
169
170    fn keepalive(timestamp: u32) -> Vec<u8> {
171        bgp4mp_record(timestamp, BgpMessage::KeepAlive)
172    }
173
174    fn update(timestamp: u32) -> Vec<u8> {
175        let mut attributes = Attributes::default();
176        attributes.add_attr(AttributeValue::Origin(Origin::IGP).into());
177        attributes.add_attr(AttributeValue::AsPath(AsPath::from_sequence([65000])).into());
178        attributes.add_attr(AttributeValue::NextHop("192.0.2.254".parse().unwrap()).into());
179        bgp4mp_record(
180            timestamp,
181            BgpMessage::Update(BgpUpdateMessage {
182                withdrawn_prefixes: vec![],
183                attributes,
184                announced_prefixes: vec!["198.51.100.0/24".parse().unwrap()],
185            }),
186        )
187    }
188
189    #[test]
190    fn no_filters_yield_original_bytes_exactly() {
191        let mut input = keepalive(1);
192        let upd = update(2);
193        input.extend_from_slice(&upd);
194
195        let parser = BgpkitParser::from_reader(Cursor::new(input.clone()));
196        let yielded: Vec<Vec<u8>> = parser
197            .into_filtered_raw_record_iter()
198            .map(|(raw, _)| raw.raw_bytes().to_vec())
199            .collect();
200        assert_eq!(yielded.len(), 2);
201        assert_eq!(yielded[0], input[..input.len() - upd.len()].to_vec());
202        assert_eq!(yielded[1], upd);
203    }
204
205    #[test]
206    fn filters_drop_no_elem_records_and_preserve_bytes() {
207        let keep = keepalive(1);
208        let upd = update(2);
209        let mut input = keep.clone();
210        input.extend_from_slice(&upd);
211
212        let parser = BgpkitParser::from_reader(Cursor::new(input))
213            .add_filter("prefix", "198.51.100.0/24")
214            .unwrap();
215        let yielded: Vec<Vec<u8>> = parser
216            .into_filtered_raw_record_iter()
217            .map(|(raw, _)| raw.raw_bytes().to_vec())
218            .collect();
219
220        // the keepalive is dropped; the update passes with byte-exact bytes
221        assert_eq!(yielded.len(), 1);
222        assert_eq!(yielded[0], upd);
223    }
224}
bgpkit_parser/parser/iters/raw.rs

bgpkit_parser/parser/iters/
raw.rs
