bgpkit_parser/parser/mrt/
mrt_record.rs

1use super::mrt_header::parse_common_header_with_bytes;
2use crate::bmp::messages::{BmpMessage, BmpMessageBody};
3use crate::error::{check_max, EncodingError, ParserError};
4use crate::models::*;
5use crate::parser::{
6    parse_bgp4mp, parse_legacy_bgp, parse_table_dump_messages, parse_table_dump_v2_message,
7    ParserErrorWithBytes,
8};
9use crate::utils::convert_timestamp;
10use bytes::{BufMut, Bytes, BytesMut};
11use log::{debug, warn};
12use std::convert::TryFrom;
13use std::fs::File;
14use std::io::{Read, Write};
15use std::net::IpAddr;
16use std::path::Path;
17use std::str::FromStr;
18
19/// Raw MRT record containing the common header and unparsed message bytes.
20/// This allows for lazy parsing of the MRT message body, and provides
21/// utilities for debugging and exporting problematic records.
22#[derive(Debug, Clone)]
23pub struct RawMrtRecord {
24    pub common_header: CommonHeader,
25    /// The raw bytes of the MRT common header (as read from the wire).
26    pub header_bytes: Bytes,
27    /// The raw bytes of the MRT message body (excluding the common header).
28    pub message_bytes: Bytes,
29}
30
31/// Internal framing error that retains MRT header context when it was parsed.
32pub(crate) struct RawMrtRecordError {
33    pub(crate) error: ParserError,
34    pub(crate) common_header: Option<CommonHeader>,
35    pub(crate) bytes: Option<Vec<u8>>,
36}
37
38impl RawMrtRecord {
39    /// Parse the raw MRT record into a fully parsed MrtRecord.
40    /// This consumes the RawMrtRecord and returns a MrtRecord.
41    pub fn parse(self) -> Result<MrtRecord, ParserError> {
42        let message = parse_mrt_body(
43            self.common_header.entry_type as u16,
44            self.common_header.entry_subtype,
45            self.message_bytes,
46        )?;
47
48        Ok(MrtRecord {
49            common_header: self.common_header,
50            message,
51        })
52    }
53
54    /// Returns the complete MRT record as raw bytes (header + message body).
55    ///
56    /// This returns the exact bytes as they were read from the wire,
57    /// without any re-encoding. This is useful for debugging problematic
58    /// MRT records by exporting them as-is to a file for further analysis.
59    ///
60    /// # Example
61    /// ```ignore
62    /// let raw_record = parser.into_raw_record_iter().next().unwrap();
63    /// let bytes = raw_record.raw_bytes();
64    /// std::fs::write("record.mrt", &bytes).unwrap();
65    /// ```
66    pub fn raw_bytes(&self) -> Bytes {
67        let mut bytes = BytesMut::with_capacity(self.header_bytes.len() + self.message_bytes.len());
68        bytes.put_slice(&self.header_bytes);
69        bytes.put_slice(&self.message_bytes);
70        bytes.freeze()
71    }
72
73    /// Writes the raw MRT record (header + message body) to a file.
74    ///
75    /// This is useful for extracting problematic MRT records for debugging
76    /// or further analysis with other tools.
77    ///
78    /// # Arguments
79    /// * `path` - The path to write the raw bytes to.
80    ///
81    /// # Example
82    /// ```ignore
83    /// let raw_record = parser.into_raw_record_iter().next().unwrap();
84    /// raw_record.write_raw_bytes("problematic_record.mrt").unwrap();
85    /// ```
86    pub fn write_raw_bytes<P: AsRef<Path>>(&self, path: P) -> std::io::Result<()> {
87        let mut file = File::create(path)?;
88        file.write_all(&self.header_bytes)?;
89        file.write_all(&self.message_bytes)?;
90        Ok(())
91    }
92
93    /// Appends the raw MRT record (header + message body) to a file.
94    ///
95    /// This is useful for collecting multiple problematic records into a single file.
96    ///
97    /// # Arguments
98    /// * `path` - The path to append the raw bytes to.
99    ///
100    /// # Example
101    /// ```ignore
102    /// for raw_record in parser.into_raw_record_iter() {
103    ///     if is_problematic(&raw_record) {
104    ///         raw_record.append_raw_bytes("problematic_records.mrt").unwrap();
105    ///     }
106    /// }
107    /// ```
108    pub fn append_raw_bytes<P: AsRef<Path>>(&self, path: P) -> std::io::Result<()> {
109        let mut file = std::fs::OpenOptions::new()
110            .create(true)
111            .append(true)
112            .open(path)?;
113        file.write_all(&self.header_bytes)?;
114        file.write_all(&self.message_bytes)?;
115        Ok(())
116    }
117
118    /// Returns the total length of the complete MRT record in bytes (header + body).
119    pub fn total_bytes_len(&self) -> usize {
120        self.header_bytes.len() + self.message_bytes.len()
121    }
122}
123
124pub fn chunk_mrt_record(input: &mut impl Read) -> Result<RawMrtRecord, ParserErrorWithBytes> {
125    chunk_mrt_record_with_context(input).map_err(|error| ParserErrorWithBytes {
126        error: error.error,
127        bytes: error.bytes,
128    })
129}
130
131/// Read one raw MRT record while retaining parsed header context on framing errors.
132pub(crate) fn chunk_mrt_record_with_context(
133    input: &mut impl Read,
134) -> Result<RawMrtRecord, RawMrtRecordError> {
135    // parse common header and capture raw bytes
136    let mut consumed_header = Vec::with_capacity(16);
137    let parsed_header = match parse_common_header_with_bytes(&mut CapturingReader {
138        inner: input,
139        captured: &mut consumed_header,
140    }) {
141        Ok(v) => v,
142        Err(e) => {
143            if let ParserError::EofError(e) = &e {
144                if e.kind() == std::io::ErrorKind::UnexpectedEof && consumed_header.is_empty() {
145                    return Err(RawMrtRecordError {
146                        error: ParserError::EofExpected,
147                        common_header: None,
148                        bytes: None,
149                    });
150                }
151            }
152            return Err(RawMrtRecordError {
153                error: e,
154                common_header: None,
155                bytes: Some(consumed_header),
156            });
157        }
158    };
159
160    let common_header = parsed_header.header;
161    let header_bytes = parsed_header.raw_bytes;
162
163    // Protect against unreasonable allocations from corrupt headers
164    const MAX_MRT_MESSAGE_LEN: u32 = 16 * 1024 * 1024; // 16 MiB upper bound
165    if common_header.length > MAX_MRT_MESSAGE_LEN {
166        return Err(RawMrtRecordError {
167            error: ParserError::Unsupported(format!(
168                "MRT message too large: {} bytes",
169                common_header.length
170            )),
171            common_header: Some(common_header),
172            bytes: Some(header_bytes.to_vec()),
173        });
174    }
175
176    // Read the declared body while retaining partial bytes on failure.
177    let mut buffer = Vec::with_capacity(common_header.length as usize + 4);
178    if let Err(error) = input
179        .take(common_header.length as u64)
180        .read_to_end(&mut buffer)
181    {
182        return Err(record_io_error(
183            error,
184            common_header,
185            &header_bytes,
186            &buffer,
187        ));
188    }
189    if buffer.len() != common_header.length as usize {
190        return Err(record_io_error(
191            std::io::Error::new(
192                std::io::ErrorKind::UnexpectedEof,
193                format!(
194                    "truncated MRT body: expected {} bytes, read {}",
195                    common_header.length,
196                    buffer.len()
197                ),
198            ),
199            common_header,
200            &header_bytes,
201            &buffer,
202        ));
203    }
204
205    if common_header.entry_type == EntryType::TABLE_DUMP
206        && super::messages::table_dump::needs_legacy_length_correction(
207            common_header.entry_subtype,
208            &buffer,
209        )
210    {
211        let mut correction = Vec::with_capacity(4);
212        if let Err(error) = input.take(4).read_to_end(&mut correction) {
213            buffer.extend_from_slice(&correction);
214            return Err(record_io_error(
215                error,
216                common_header,
217                &header_bytes,
218                &buffer,
219            ));
220        }
221        buffer.extend_from_slice(&correction);
222        if correction.len() != 4 {
223            return Err(record_io_error(
224                std::io::Error::new(
225                    std::io::ErrorKind::UnexpectedEof,
226                    "truncated historical TABLE_DUMP length correction",
227                ),
228                common_header,
229                &header_bytes,
230                &buffer,
231            ));
232        }
233        debug!(
234            "recovered historical TABLE_DUMP record whose declared length was four bytes short (timestamp={}, subtype={}, declared_length={})",
235            common_header.timestamp,
236            common_header.entry_subtype,
237            common_header.length
238        );
239    }
240
241    Ok(RawMrtRecord {
242        common_header,
243        header_bytes,
244        message_bytes: Bytes::from(buffer),
245    })
246}
247
248struct CapturingReader<'a, R: ?Sized> {
249    inner: &'a mut R,
250    captured: &'a mut Vec<u8>,
251}
252
253impl<R: Read + ?Sized> Read for CapturingReader<'_, R> {
254    fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
255        let read = self.inner.read(buffer)?;
256        self.captured.extend_from_slice(&buffer[..read]);
257        Ok(read)
258    }
259}
260
261fn record_io_error(
262    error: std::io::Error,
263    common_header: CommonHeader,
264    header: &[u8],
265    body: &[u8],
266) -> RawMrtRecordError {
267    let mut bytes = Vec::with_capacity(header.len() + body.len());
268    bytes.extend_from_slice(header);
269    bytes.extend_from_slice(body);
270    RawMrtRecordError {
271        error: ParserError::IoError(error),
272        common_header: Some(common_header),
273        bytes: Some(bytes),
274    }
275}
276
277pub fn parse_mrt_record(input: &mut impl Read) -> Result<MrtRecord, ParserErrorWithBytes> {
278    parse_mrt_record_with_zebra_compat(input).map(|(record, _)| record)
279}
280
281pub(crate) fn raw_record_uses_zebra_compat(raw_record: &RawMrtRecord) -> bool {
282    matches!(
283        raw_record.common_header.entry_type,
284        EntryType::BGP4MP | EntryType::BGP4MP_ET
285    ) && crate::parser::mrt::messages::bgp4mp::uses_zebra_compat(
286        raw_record.common_header.entry_subtype,
287        &raw_record.message_bytes,
288    )
289}
290
291pub(crate) fn parse_mrt_record_with_zebra_compat(
292    input: &mut impl Read,
293) -> Result<(MrtRecord, bool), ParserErrorWithBytes> {
294    let raw_record = chunk_mrt_record(input)?;
295    let used_zebra_compat = raw_record_uses_zebra_compat(&raw_record);
296    // Parse from a clone so the original is available for raw_bytes() on error,
297    // avoiding manual reassembly that could diverge from RawMrtRecord::raw_bytes().
298    match raw_record.clone().parse() {
299        Ok(record) => Ok((record, used_zebra_compat)),
300        Err(e) => Err(ParserErrorWithBytes {
301            error: e,
302            bytes: Some(raw_record.raw_bytes().to_vec()),
303        }),
304    }
305}
306
307/// Parse MRT message body with given entry type and subtype.
308///
309/// The entry type and subtype are parsed from the common header. The message body is parsed
310/// according to the entry type and subtype. The message body is the remaining bytes after the
311/// common header. The length of the message body is also parsed from the common header.
312pub fn parse_mrt_body(
313    entry_type: u16,
314    entry_subtype: u16,
315    data: Bytes,
316) -> Result<MrtMessage, ParserError> {
317    let etype = EntryType::try_from(entry_type)?;
318
319    let message: MrtMessage = match &etype {
320        EntryType::TABLE_DUMP => {
321            let mut messages = parse_table_dump_messages(entry_subtype, data)?;
322            if messages.len() == 1 {
323                MrtMessage::TableDumpMessage(messages.remove(0))
324            } else {
325                MrtMessage::TableDumpMessageBatch(messages)
326            }
327        }
328        EntryType::TABLE_DUMP_V2 => {
329            let msg = parse_table_dump_v2_message(entry_subtype, data);
330            match msg {
331                Ok(msg) => MrtMessage::TableDumpV2Message(msg),
332                Err(e) => {
333                    return Err(e);
334                }
335            }
336        }
337        EntryType::BGP4MP | EntryType::BGP4MP_ET => {
338            let msg = parse_bgp4mp(entry_subtype, data);
339            match msg {
340                Ok(msg) => MrtMessage::Bgp4Mp(msg),
341                Err(e) => {
342                    return Err(e);
343                }
344            }
345        }
346        EntryType::BGP => MrtMessage::LegacyBgp(parse_legacy_bgp(entry_subtype, data)?),
347        v => {
348            // deprecated
349            return Err(ParserError::Unsupported(format!(
350                "unsupported MRT type: {v:?}"
351            )));
352        }
353    };
354    Ok(message)
355}
356
357impl MrtRecord {
358    pub fn encode(&self) -> Result<Bytes, EncodingError> {
359        let message_bytes = self.message.encode(self.common_header.entry_subtype)?;
360        let mut new_header = self.common_header;
361        if message_bytes.len() != new_header.length as usize {
362            warn!(
363                "message length {} does not match the length in the header {} (encoding MRT record)",
364                message_bytes.len(),
365                new_header.length
366            );
367        }
368        check_max(
369            "MRT record message length",
370            message_bytes.len(),
371            u32::MAX as usize,
372        )?;
373        new_header.length = message_bytes.len() as u32;
374        let header_bytes = new_header.encode();
375
376        // // debug begins
377        // let parsed_body = parse_mrt_body(
378        //     self.common_header.entry_type as u16,
379        //     self.common_header.entry_subtype,
380        //     message_bytes.clone(),
381        // )
382        // .unwrap();
383        // assert!(self.message == parsed_body);
384        // // debug ends
385
386        let mut bytes = BytesMut::with_capacity(header_bytes.len() + message_bytes.len());
387        bytes.put_slice(&header_bytes);
388        bytes.put_slice(&message_bytes);
389        Ok(bytes.freeze())
390    }
391}
392
393impl TryFrom<&BmpMessage> for MrtRecord {
394    type Error = String;
395
396    fn try_from(bmp_message: &BmpMessage) -> Result<Self, Self::Error> {
397        let bgp_message = match &bmp_message.message_body {
398            BmpMessageBody::RouteMonitoring(m) => &m.bgp_message,
399            _ => return Err("unsupported bmp message type".to_string()),
400        };
401        let bmp_header = match &bmp_message.per_peer_header {
402            Some(h) => h,
403            None => return Err("missing per peer header".to_string()),
404        };
405
406        let local_ip = match bmp_header.peer_ip {
407            IpAddr::V4(_) => IpAddr::from_str("0.0.0.0").unwrap(),
408            IpAddr::V6(_) => IpAddr::from_str("::").unwrap(),
409        };
410        let local_asn = match bmp_header.peer_asn.is_four_byte() {
411            true => Asn::new_32bit(0),
412            false => Asn::new_16bit(0),
413        };
414
415        let bgp4mp_message = Bgp4MpMessage {
416            msg_type: Bgp4MpType::MessageAs4, // TODO: check Message or MessageAs4
417            peer_asn: bmp_header.peer_asn,
418            local_asn,
419            interface_index: 0,
420            peer_ip: bmp_header.peer_ip,
421            local_ip,
422            bgp_message: bgp_message.clone(),
423        };
424
425        let mrt_message = MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(bgp4mp_message));
426
427        let (seconds, microseconds) = convert_timestamp(bmp_header.timestamp);
428
429        let subtype = Bgp4MpType::MessageAs4 as u16;
430        let encoded_message = mrt_message
431            .encode(subtype)
432            .map_err(|e| format!("cannot encode MRT message: {e}"))?;
433        let mrt_header = CommonHeader {
434            timestamp: seconds,
435            microsecond_timestamp: Some(microseconds),
436            entry_type: EntryType::BGP4MP_ET,
437            entry_subtype: Bgp4MpType::MessageAs4 as u16,
438            length: encoded_message.len() as u32,
439        };
440
441        Ok(MrtRecord {
442            common_header: mrt_header,
443            message: mrt_message,
444        })
445    }
446}
447
448#[cfg(test)]
449mod tests {
450    use super::*;
451    use crate::bmp::messages::headers::{BmpPeerType, PeerFlags, PerPeerFlags};
452    use crate::bmp::messages::{BmpCommonHeader, BmpMsgType, BmpPerPeerHeader, RouteMonitoring};
453    use crate::models::{AttributeValue, Origin};
454    use crate::parser::mrt::messages::table_dump::encode_table_dump_batch;
455    use std::io::Cursor;
456    use std::net::Ipv4Addr;
457    use tempfile::tempdir;
458
459    fn table_dump_message(prefix: &str) -> TableDumpMessage {
460        let mut attributes = Attributes::default();
461        attributes.add_attr(AttributeValue::Origin(Origin::IGP).into());
462        TableDumpMessage {
463            view_number: 0,
464            sequence_number: 1,
465            prefix: prefix.parse().unwrap(),
466            status: 1,
467            originated_time: 946_684_800,
468            peer_ip: IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)),
469            peer_asn: Asn::new_16bit(64512),
470            attributes,
471        }
472    }
473
474    #[test]
475    fn test_raw_mrt_record_raw_bytes() {
476        let header = CommonHeader {
477            timestamp: 1609459200,
478            microsecond_timestamp: None,
479            entry_type: EntryType::BGP4MP,
480            entry_subtype: 4,
481            length: 10,
482        };
483        let header_bytes = Bytes::from_static(&[
484            0x5f, 0xee, 0x6a, 0x80, // timestamp
485            0x00, 0x10, // entry type
486            0x00, 0x04, // entry subtype
487            0x00, 0x00, 0x00, 0x0a, // length
488        ]);
489        let message_bytes = Bytes::from_static(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
490
491        let raw_record = RawMrtRecord {
492            common_header: header,
493            header_bytes,
494            message_bytes,
495        };
496
497        let mrt_bytes = raw_record.raw_bytes();
498        // Header is 12 bytes + 10 bytes body = 22 bytes total
499        assert_eq!(mrt_bytes.len(), 22);
500        assert_eq!(raw_record.total_bytes_len(), 22);
501    }
502
503    #[test]
504    fn test_raw_mrt_record_raw_bytes_with_et() {
505        let header = CommonHeader {
506            timestamp: 1609459200,
507            microsecond_timestamp: Some(500000),
508            entry_type: EntryType::BGP4MP_ET,
509            entry_subtype: 4,
510            length: 10,
511        };
512        let header_bytes = Bytes::from_static(&[
513            0x5f, 0xee, 0x6a, 0x80, // timestamp
514            0x00, 0x11, // entry type (BGP4MP_ET = 17)
515            0x00, 0x04, // entry subtype
516            0x00, 0x00, 0x00, 0x0e, // length (10 + 4 for microseconds)
517            0x00, 0x07, 0xa1, 0x20, // microsecond timestamp (500000)
518        ]);
519        let message_bytes = Bytes::from_static(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
520
521        let raw_record = RawMrtRecord {
522            common_header: header,
523            header_bytes,
524            message_bytes,
525        };
526
527        let mrt_bytes = raw_record.raw_bytes();
528        // ET Header is 16 bytes + 10 bytes body = 26 bytes total
529        assert_eq!(mrt_bytes.len(), 26);
530        assert_eq!(raw_record.total_bytes_len(), 26);
531    }
532
533    #[test]
534    fn test_raw_mrt_record_write_to_file() {
535        let dir = tempdir().unwrap();
536        let file_path = dir.path().join("test_record.mrt");
537
538        let header = CommonHeader {
539            timestamp: 1609459200,
540            microsecond_timestamp: None,
541            entry_type: EntryType::BGP4MP,
542            entry_subtype: 4,
543            length: 5,
544        };
545        let header_bytes = Bytes::from_static(&[
546            0x5f, 0xee, 0x6a, 0x80, // timestamp
547            0x00, 0x10, // entry type
548            0x00, 0x04, // entry subtype
549            0x00, 0x00, 0x00, 0x05, // length
550        ]);
551        let message_bytes = Bytes::from_static(&[1, 2, 3, 4, 5]);
552
553        let raw_record = RawMrtRecord {
554            common_header: header,
555            header_bytes,
556            message_bytes,
557        };
558
559        raw_record.write_raw_bytes(&file_path).unwrap();
560
561        let written_bytes = std::fs::read(&file_path).unwrap();
562        assert_eq!(written_bytes.len(), 17); // 12 header + 5 body
563    }
564
565    #[test]
566    fn test_raw_mrt_record_append_to_file() {
567        let dir = tempdir().unwrap();
568        let file_path = dir.path().join("test_records.mrt");
569
570        let header = CommonHeader {
571            timestamp: 1609459200,
572            microsecond_timestamp: None,
573            entry_type: EntryType::BGP4MP,
574            entry_subtype: 4,
575            length: 3,
576        };
577        let header_bytes = Bytes::from_static(&[
578            0x5f, 0xee, 0x6a, 0x80, // timestamp
579            0x00, 0x10, // entry type
580            0x00, 0x04, // entry subtype
581            0x00, 0x00, 0x00, 0x03, // length
582        ]);
583        let message_bytes = Bytes::from_static(&[1, 2, 3]);
584
585        let raw_record = RawMrtRecord {
586            common_header: header,
587            header_bytes,
588            message_bytes,
589        };
590
591        raw_record.append_raw_bytes(&file_path).unwrap();
592        raw_record.append_raw_bytes(&file_path).unwrap();
593
594        let written_bytes = std::fs::read(&file_path).unwrap();
595        assert_eq!(written_bytes.len(), 30); // (12 header + 3 body) * 2
596    }
597
598    #[test]
599    fn test_try_from_bmp_message() {
600        let bmp_message = BmpMessage {
601            common_header: BmpCommonHeader {
602                version: 0,
603                msg_len: 0,
604                msg_type: BmpMsgType::RouteMonitoring,
605            },
606            per_peer_header: Some(BmpPerPeerHeader {
607                peer_asn: Asn::new_32bit(0),
608                peer_ip: IpAddr::from_str("10.0.0.1").unwrap(),
609                peer_bgp_id: Ipv4Addr::from_str("10.0.0.2").unwrap(),
610                timestamp: 0.0,
611                peer_type: BmpPeerType::Global,
612                peer_flags: PerPeerFlags::PeerFlags(PeerFlags::empty()),
613                peer_distinguisher: 0,
614            }),
615            message_body: BmpMessageBody::RouteMonitoring(RouteMonitoring {
616                bgp_message: BgpMessage::KeepAlive,
617            }),
618        };
619
620        let mrt_record = MrtRecord::try_from(&bmp_message).unwrap();
621        assert_eq!(mrt_record.common_header.entry_type, EntryType::BGP4MP_ET);
622    }
623
624    #[test]
625    fn test_parse_mrt_body() {
626        let mut data = BytesMut::new();
627        data.put_u16(0);
628        data.put_u16(0);
629        data.put_u32(0);
630        data.put_u16(0);
631
632        let result = parse_mrt_body(0, 0, data.freeze());
633        assert!(result.is_err());
634    }
635
636    #[test]
637    fn test_mrt_record_encode_updates_header_length() {
638        let record = MrtRecord {
639            common_header: CommonHeader {
640                timestamp: 1609459200,
641                microsecond_timestamp: None,
642                entry_type: EntryType::BGP4MP,
643                entry_subtype: Bgp4MpType::MessageAs4 as u16,
644                length: 0,
645            },
646            message: MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(Bgp4MpMessage {
647                msg_type: Bgp4MpType::MessageAs4,
648                peer_asn: Asn::new_32bit(65000),
649                local_asn: Asn::new_32bit(65001),
650                interface_index: 1,
651                peer_ip: IpAddr::from_str("10.0.0.1").unwrap(),
652                local_ip: IpAddr::from_str("10.0.0.2").unwrap(),
653                bgp_message: BgpMessage::KeepAlive,
654            })),
655        };
656
657        let encoded = record.encode().unwrap();
658        let mut cursor = Cursor::new(encoded);
659        let parsed = parse_mrt_record(&mut cursor).unwrap();
660        let expected_len = parsed
661            .message
662            .encode(parsed.common_header.entry_subtype)
663            .unwrap()
664            .len() as u32;
665
666        assert_eq!(parsed.common_header.length, expected_len);
667    }
668
669    #[test]
670    fn chunk_recovers_four_byte_short_table_dump_and_preserves_alignment() {
671        let body = encode_table_dump_batch(
672            &[
673                table_dump_message("192.0.2.0/24"),
674                table_dump_message("198.51.100.0/24"),
675            ],
676            1,
677        )
678        .unwrap();
679        let header = CommonHeader {
680            timestamp: 946_684_800,
681            microsecond_timestamp: None,
682            entry_type: EntryType::TABLE_DUMP,
683            entry_subtype: 1,
684            length: (body.len() - 4) as u32,
685        };
686        let next_body = Bytes::from_static(&[0, 1, 192, 0, 2, 1, 0, 2, 192, 0, 2, 2]);
687        let next_header = CommonHeader {
688            timestamp: 946_684_801,
689            microsecond_timestamp: None,
690            entry_type: EntryType::BGP,
691            entry_subtype: 7,
692            length: next_body.len() as u32,
693        };
694
695        let mut wire = BytesMut::new();
696        wire.put_slice(&header.encode());
697        wire.put_slice(&body);
698        wire.put_slice(&next_header.encode());
699        wire.put_slice(&next_body);
700        let mut cursor = Cursor::new(wire.freeze());
701
702        let first = chunk_mrt_record(&mut cursor).unwrap();
703        assert_eq!(first.message_bytes, body);
704        assert!(matches!(
705            first.parse().unwrap().message,
706            MrtMessage::TableDumpMessageBatch(messages) if messages.len() == 2
707        ));
708        let second = chunk_mrt_record(&mut cursor).unwrap();
709        assert_eq!(second.common_header, next_header);
710        assert!(matches!(
711            second.parse().unwrap().message,
712            MrtMessage::LegacyBgp(LegacyBgp::Message(LegacyBgpMessage {
713                bgp_message: BgpMessage::KeepAlive,
714                ..
715            }))
716        ));
717    }
718
719    #[test]
720    fn chunk_does_not_overread_near_match_table_dump() {
721        let body = encode_table_dump_batch(&[table_dump_message("192.0.2.0/24")], 1).unwrap();
722        let declared_length = body.len() - 3;
723        let header = CommonHeader {
724            timestamp: 946_684_800,
725            microsecond_timestamp: None,
726            entry_type: EntryType::TABLE_DUMP,
727            entry_subtype: 1,
728            length: declared_length as u32,
729        };
730        let mut wire = BytesMut::new();
731        wire.put_slice(&header.encode());
732        wire.put_slice(&body);
733        wire.put_slice(&[0xaa; 12]);
734        let mut cursor = Cursor::new(wire.freeze());
735
736        let raw = chunk_mrt_record(&mut cursor).unwrap();
737        assert_eq!(raw.message_bytes.len(), declared_length);
738        assert_eq!(cursor.position(), (12 + declared_length) as u64);
739        let error = raw.parse().unwrap_err();
740        assert!(matches!(error, ParserError::TruncatedMsg(_)));
741    }
742
743    #[test]
744    fn chunk_errors_include_invalid_header_and_partial_body_bytes() {
745        let invalid_header = Bytes::from_static(&[0, 0, 0, 1, 0xff, 0xff, 0, 0, 0, 0, 0, 0]);
746        let error = chunk_mrt_record(&mut Cursor::new(invalid_header.clone())).unwrap_err();
747        assert_eq!(error.bytes.as_deref(), Some(invalid_header.as_ref()));
748
749        let header = CommonHeader {
750            timestamp: 1,
751            microsecond_timestamp: None,
752            entry_type: EntryType::BGP,
753            entry_subtype: 7,
754            length: 5,
755        };
756        let mut wire = BytesMut::new();
757        wire.put_slice(&header.encode());
758        wire.put_slice(&[1, 2]);
759        let error = chunk_mrt_record(&mut Cursor::new(wire.clone().freeze())).unwrap_err();
760        assert_eq!(error.bytes.as_deref(), Some(wire.as_ref()));
761    }
762}
bgpkit_parser/parser/mrt/mrt_record.rs

bgpkit_parser/parser/mrt/
mrt_record.rs
