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::ParserError;
4use crate::models::*;
5use crate::parser::{
6    parse_bgp4mp, parse_table_dump_message, parse_table_dump_v2_message, ParserErrorWithBytes,
7};
8use crate::utils::convert_timestamp;
9use bytes::{BufMut, Bytes, BytesMut};
10use log::warn;
11use std::convert::TryFrom;
12use std::fs::File;
13use std::io::{Read, Write};
14use std::net::IpAddr;
15use std::path::Path;
16use std::str::FromStr;
17
18/// Raw MRT record containing the common header and unparsed message bytes.
19/// This allows for lazy parsing of the MRT message body, and provides
20/// utilities for debugging and exporting problematic records.
21#[derive(Debug, Clone)]
22pub struct RawMrtRecord {
23    pub common_header: CommonHeader,
24    /// The raw bytes of the MRT common header (as read from the wire).
25    pub header_bytes: Bytes,
26    /// The raw bytes of the MRT message body (excluding the common header).
27    pub message_bytes: Bytes,
28}
29
30impl RawMrtRecord {
31    /// Parse the raw MRT record into a fully parsed MrtRecord.
32    /// This consumes the RawMrtRecord and returns a MrtRecord.
33    pub fn parse(self) -> Result<MrtRecord, ParserError> {
34        let message = parse_mrt_body(
35            self.common_header.entry_type as u16,
36            self.common_header.entry_subtype,
37            self.message_bytes,
38        )?;
39
40        Ok(MrtRecord {
41            common_header: self.common_header,
42            message,
43        })
44    }
45
46    /// Returns the complete MRT record as raw bytes (header + message body).
47    ///
48    /// This returns the exact bytes as they were read from the wire,
49    /// without any re-encoding. This is useful for debugging problematic
50    /// MRT records by exporting them as-is to a file for further analysis.
51    ///
52    /// # Example
53    /// ```ignore
54    /// let raw_record = parser.into_raw_record_iter().next().unwrap();
55    /// let bytes = raw_record.raw_bytes();
56    /// std::fs::write("record.mrt", &bytes).unwrap();
57    /// ```
58    pub fn raw_bytes(&self) -> Bytes {
59        let mut bytes = BytesMut::with_capacity(self.header_bytes.len() + self.message_bytes.len());
60        bytes.put_slice(&self.header_bytes);
61        bytes.put_slice(&self.message_bytes);
62        bytes.freeze()
63    }
64
65    /// Writes the raw MRT record (header + message body) to a file.
66    ///
67    /// This is useful for extracting problematic MRT records for debugging
68    /// or further analysis with other tools.
69    ///
70    /// # Arguments
71    /// * `path` - The path to write the raw bytes to.
72    ///
73    /// # Example
74    /// ```ignore
75    /// let raw_record = parser.into_raw_record_iter().next().unwrap();
76    /// raw_record.write_raw_bytes("problematic_record.mrt").unwrap();
77    /// ```
78    pub fn write_raw_bytes<P: AsRef<Path>>(&self, path: P) -> std::io::Result<()> {
79        let mut file = File::create(path)?;
80        file.write_all(&self.header_bytes)?;
81        file.write_all(&self.message_bytes)?;
82        Ok(())
83    }
84
85    /// Appends the raw MRT record (header + message body) to a file.
86    ///
87    /// This is useful for collecting multiple problematic records into a single file.
88    ///
89    /// # Arguments
90    /// * `path` - The path to append the raw bytes to.
91    ///
92    /// # Example
93    /// ```ignore
94    /// for raw_record in parser.into_raw_record_iter() {
95    ///     if is_problematic(&raw_record) {
96    ///         raw_record.append_raw_bytes("problematic_records.mrt").unwrap();
97    ///     }
98    /// }
99    /// ```
100    pub fn append_raw_bytes<P: AsRef<Path>>(&self, path: P) -> std::io::Result<()> {
101        let mut file = std::fs::OpenOptions::new()
102            .create(true)
103            .append(true)
104            .open(path)?;
105        file.write_all(&self.header_bytes)?;
106        file.write_all(&self.message_bytes)?;
107        Ok(())
108    }
109
110    /// Returns the total length of the complete MRT record in bytes (header + body).
111    pub fn total_bytes_len(&self) -> usize {
112        self.header_bytes.len() + self.message_bytes.len()
113    }
114}
115
116pub fn chunk_mrt_record(input: &mut impl Read) -> Result<RawMrtRecord, ParserErrorWithBytes> {
117    // parse common header and capture raw bytes
118    let parsed_header = match parse_common_header_with_bytes(input) {
119        Ok(v) => v,
120        Err(e) => {
121            if let ParserError::EofError(e) = &e {
122                if e.kind() == std::io::ErrorKind::UnexpectedEof {
123                    return Err(ParserErrorWithBytes::from(ParserError::EofExpected));
124                }
125            }
126            return Err(ParserErrorWithBytes {
127                error: e,
128                bytes: None,
129            });
130        }
131    };
132
133    let common_header = parsed_header.header;
134    let header_bytes = parsed_header.raw_bytes;
135
136    // Protect against unreasonable allocations from corrupt headers
137    const MAX_MRT_MESSAGE_LEN: u32 = 16 * 1024 * 1024; // 16 MiB upper bound
138    if common_header.length > MAX_MRT_MESSAGE_LEN {
139        return Err(ParserErrorWithBytes::from(ParserError::Unsupported(
140            format!("MRT message too large: {} bytes", common_header.length),
141        )));
142    }
143
144    // read the whole message bytes to buffer
145    let mut buffer = BytesMut::zeroed(common_header.length as usize);
146    match input
147        .take(common_header.length as u64)
148        .read_exact(&mut buffer)
149    {
150        Ok(_) => {}
151        Err(e) => {
152            return Err(ParserErrorWithBytes {
153                error: ParserError::IoError(e),
154                bytes: None,
155            })
156        }
157    }
158
159    Ok(RawMrtRecord {
160        common_header,
161        header_bytes,
162        message_bytes: buffer.freeze(),
163    })
164}
165
166pub fn parse_mrt_record(input: &mut impl Read) -> Result<MrtRecord, ParserErrorWithBytes> {
167    let raw_record = chunk_mrt_record(input)?;
168    // Parse from a clone so the original is available for raw_bytes() on error,
169    // avoiding manual reassembly that could diverge from RawMrtRecord::raw_bytes().
170    match raw_record.clone().parse() {
171        Ok(record) => Ok(record),
172        Err(e) => Err(ParserErrorWithBytes {
173            error: e,
174            bytes: Some(raw_record.raw_bytes().to_vec()),
175        }),
176    }
177}
178
179/// Parse MRT message body with given entry type and subtype.
180///
181/// The entry type and subtype are parsed from the common header. The message body is parsed
182/// according to the entry type and subtype. The message body is the remaining bytes after the
183/// common header. The length of the message body is also parsed from the common header.
184pub fn parse_mrt_body(
185    entry_type: u16,
186    entry_subtype: u16,
187    data: Bytes,
188) -> Result<MrtMessage, ParserError> {
189    let etype = EntryType::try_from(entry_type)?;
190
191    let message: MrtMessage = match &etype {
192        EntryType::TABLE_DUMP => {
193            let msg = parse_table_dump_message(entry_subtype, data);
194            match msg {
195                Ok(msg) => MrtMessage::TableDumpMessage(msg),
196                Err(e) => {
197                    return Err(e);
198                }
199            }
200        }
201        EntryType::TABLE_DUMP_V2 => {
202            let msg = parse_table_dump_v2_message(entry_subtype, data);
203            match msg {
204                Ok(msg) => MrtMessage::TableDumpV2Message(msg),
205                Err(e) => {
206                    return Err(e);
207                }
208            }
209        }
210        EntryType::BGP4MP | EntryType::BGP4MP_ET => {
211            let msg = parse_bgp4mp(entry_subtype, data);
212            match msg {
213                Ok(msg) => MrtMessage::Bgp4Mp(msg),
214                Err(e) => {
215                    return Err(e);
216                }
217            }
218        }
219        v => {
220            // deprecated
221            return Err(ParserError::Unsupported(format!(
222                "unsupported MRT type: {v:?}"
223            )));
224        }
225    };
226    Ok(message)
227}
228
229impl MrtRecord {
230    pub fn encode(&self) -> Bytes {
231        let message_bytes = self.message.encode(self.common_header.entry_subtype);
232        let mut new_header = self.common_header;
233        if message_bytes.len() != new_header.length as usize {
234            warn!(
235                "message length {} does not match the length in the header {} (encoding MRT record)",
236                message_bytes.len(),
237                new_header.length
238            );
239        }
240        new_header.length = message_bytes.len() as u32;
241        let header_bytes = new_header.encode();
242
243        // // debug begins
244        // let parsed_body = parse_mrt_body(
245        //     self.common_header.entry_type as u16,
246        //     self.common_header.entry_subtype,
247        //     message_bytes.clone(),
248        // )
249        // .unwrap();
250        // assert!(self.message == parsed_body);
251        // // debug ends
252
253        let mut bytes = BytesMut::with_capacity(header_bytes.len() + message_bytes.len());
254        bytes.put_slice(&header_bytes);
255        bytes.put_slice(&message_bytes);
256        bytes.freeze()
257    }
258}
259
260impl TryFrom<&BmpMessage> for MrtRecord {
261    type Error = String;
262
263    fn try_from(bmp_message: &BmpMessage) -> Result<Self, Self::Error> {
264        let bgp_message = match &bmp_message.message_body {
265            BmpMessageBody::RouteMonitoring(m) => &m.bgp_message,
266            _ => return Err("unsupported bmp message type".to_string()),
267        };
268        let bmp_header = match &bmp_message.per_peer_header {
269            Some(h) => h,
270            None => return Err("missing per peer header".to_string()),
271        };
272
273        let local_ip = match bmp_header.peer_ip {
274            IpAddr::V4(_) => IpAddr::from_str("0.0.0.0").unwrap(),
275            IpAddr::V6(_) => IpAddr::from_str("::").unwrap(),
276        };
277        let local_asn = match bmp_header.peer_asn.is_four_byte() {
278            true => Asn::new_32bit(0),
279            false => Asn::new_16bit(0),
280        };
281
282        let bgp4mp_message = Bgp4MpMessage {
283            msg_type: Bgp4MpType::MessageAs4, // TODO: check Message or MessageAs4
284            peer_asn: bmp_header.peer_asn,
285            local_asn,
286            interface_index: 0,
287            peer_ip: bmp_header.peer_ip,
288            local_ip,
289            bgp_message: bgp_message.clone(),
290        };
291
292        let mrt_message = MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(bgp4mp_message));
293
294        let (seconds, microseconds) = convert_timestamp(bmp_header.timestamp);
295
296        let subtype = Bgp4MpType::MessageAs4 as u16;
297        let mrt_header = CommonHeader {
298            timestamp: seconds,
299            microsecond_timestamp: Some(microseconds),
300            entry_type: EntryType::BGP4MP_ET,
301            entry_subtype: Bgp4MpType::MessageAs4 as u16,
302            length: mrt_message.encode(subtype).len() as u32,
303        };
304
305        Ok(MrtRecord {
306            common_header: mrt_header,
307            message: mrt_message,
308        })
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    use super::*;
315    use crate::bmp::messages::headers::{BmpPeerType, PeerFlags, PerPeerFlags};
316    use crate::bmp::messages::{BmpCommonHeader, BmpMsgType, BmpPerPeerHeader, RouteMonitoring};
317    use std::io::Cursor;
318    use std::net::Ipv4Addr;
319    use tempfile::tempdir;
320
321    #[test]
322    fn test_raw_mrt_record_raw_bytes() {
323        let header = CommonHeader {
324            timestamp: 1609459200,
325            microsecond_timestamp: None,
326            entry_type: EntryType::BGP4MP,
327            entry_subtype: 4,
328            length: 10,
329        };
330        let header_bytes = Bytes::from_static(&[
331            0x5f, 0xee, 0x6a, 0x80, // timestamp
332            0x00, 0x10, // entry type
333            0x00, 0x04, // entry subtype
334            0x00, 0x00, 0x00, 0x0a, // length
335        ]);
336        let message_bytes = Bytes::from_static(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
337
338        let raw_record = RawMrtRecord {
339            common_header: header,
340            header_bytes,
341            message_bytes,
342        };
343
344        let mrt_bytes = raw_record.raw_bytes();
345        // Header is 12 bytes + 10 bytes body = 22 bytes total
346        assert_eq!(mrt_bytes.len(), 22);
347        assert_eq!(raw_record.total_bytes_len(), 22);
348    }
349
350    #[test]
351    fn test_raw_mrt_record_raw_bytes_with_et() {
352        let header = CommonHeader {
353            timestamp: 1609459200,
354            microsecond_timestamp: Some(500000),
355            entry_type: EntryType::BGP4MP_ET,
356            entry_subtype: 4,
357            length: 10,
358        };
359        let header_bytes = Bytes::from_static(&[
360            0x5f, 0xee, 0x6a, 0x80, // timestamp
361            0x00, 0x11, // entry type (BGP4MP_ET = 17)
362            0x00, 0x04, // entry subtype
363            0x00, 0x00, 0x00, 0x0e, // length (10 + 4 for microseconds)
364            0x00, 0x07, 0xa1, 0x20, // microsecond timestamp (500000)
365        ]);
366        let message_bytes = Bytes::from_static(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
367
368        let raw_record = RawMrtRecord {
369            common_header: header,
370            header_bytes,
371            message_bytes,
372        };
373
374        let mrt_bytes = raw_record.raw_bytes();
375        // ET Header is 16 bytes + 10 bytes body = 26 bytes total
376        assert_eq!(mrt_bytes.len(), 26);
377        assert_eq!(raw_record.total_bytes_len(), 26);
378    }
379
380    #[test]
381    fn test_raw_mrt_record_write_to_file() {
382        let dir = tempdir().unwrap();
383        let file_path = dir.path().join("test_record.mrt");
384
385        let header = CommonHeader {
386            timestamp: 1609459200,
387            microsecond_timestamp: None,
388            entry_type: EntryType::BGP4MP,
389            entry_subtype: 4,
390            length: 5,
391        };
392        let header_bytes = Bytes::from_static(&[
393            0x5f, 0xee, 0x6a, 0x80, // timestamp
394            0x00, 0x10, // entry type
395            0x00, 0x04, // entry subtype
396            0x00, 0x00, 0x00, 0x05, // length
397        ]);
398        let message_bytes = Bytes::from_static(&[1, 2, 3, 4, 5]);
399
400        let raw_record = RawMrtRecord {
401            common_header: header,
402            header_bytes,
403            message_bytes,
404        };
405
406        raw_record.write_raw_bytes(&file_path).unwrap();
407
408        let written_bytes = std::fs::read(&file_path).unwrap();
409        assert_eq!(written_bytes.len(), 17); // 12 header + 5 body
410    }
411
412    #[test]
413    fn test_raw_mrt_record_append_to_file() {
414        let dir = tempdir().unwrap();
415        let file_path = dir.path().join("test_records.mrt");
416
417        let header = CommonHeader {
418            timestamp: 1609459200,
419            microsecond_timestamp: None,
420            entry_type: EntryType::BGP4MP,
421            entry_subtype: 4,
422            length: 3,
423        };
424        let header_bytes = Bytes::from_static(&[
425            0x5f, 0xee, 0x6a, 0x80, // timestamp
426            0x00, 0x10, // entry type
427            0x00, 0x04, // entry subtype
428            0x00, 0x00, 0x00, 0x03, // length
429        ]);
430        let message_bytes = Bytes::from_static(&[1, 2, 3]);
431
432        let raw_record = RawMrtRecord {
433            common_header: header,
434            header_bytes,
435            message_bytes,
436        };
437
438        raw_record.append_raw_bytes(&file_path).unwrap();
439        raw_record.append_raw_bytes(&file_path).unwrap();
440
441        let written_bytes = std::fs::read(&file_path).unwrap();
442        assert_eq!(written_bytes.len(), 30); // (12 header + 3 body) * 2
443    }
444
445    #[test]
446    fn test_try_from_bmp_message() {
447        let bmp_message = BmpMessage {
448            common_header: BmpCommonHeader {
449                version: 0,
450                msg_len: 0,
451                msg_type: BmpMsgType::RouteMonitoring,
452            },
453            per_peer_header: Some(BmpPerPeerHeader {
454                peer_asn: Asn::new_32bit(0),
455                peer_ip: IpAddr::from_str("10.0.0.1").unwrap(),
456                peer_bgp_id: Ipv4Addr::from_str("10.0.0.2").unwrap(),
457                timestamp: 0.0,
458                peer_type: BmpPeerType::Global,
459                peer_flags: PerPeerFlags::PeerFlags(PeerFlags::empty()),
460                peer_distinguisher: 0,
461            }),
462            message_body: BmpMessageBody::RouteMonitoring(RouteMonitoring {
463                bgp_message: BgpMessage::KeepAlive,
464            }),
465        };
466
467        let mrt_record = MrtRecord::try_from(&bmp_message).unwrap();
468        assert_eq!(mrt_record.common_header.entry_type, EntryType::BGP4MP_ET);
469    }
470
471    #[test]
472    fn test_parse_mrt_body() {
473        let mut data = BytesMut::new();
474        data.put_u16(0);
475        data.put_u16(0);
476        data.put_u32(0);
477        data.put_u16(0);
478
479        let result = parse_mrt_body(0, 0, data.freeze());
480        assert!(result.is_err());
481    }
482
483    #[test]
484    fn test_mrt_record_encode_updates_header_length() {
485        let record = MrtRecord {
486            common_header: CommonHeader {
487                timestamp: 1609459200,
488                microsecond_timestamp: None,
489                entry_type: EntryType::BGP4MP,
490                entry_subtype: Bgp4MpType::MessageAs4 as u16,
491                length: 0,
492            },
493            message: MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(Bgp4MpMessage {
494                msg_type: Bgp4MpType::MessageAs4,
495                peer_asn: Asn::new_32bit(65000),
496                local_asn: Asn::new_32bit(65001),
497                interface_index: 1,
498                peer_ip: IpAddr::from_str("10.0.0.1").unwrap(),
499                local_ip: IpAddr::from_str("10.0.0.2").unwrap(),
500                bgp_message: BgpMessage::KeepAlive,
501            })),
502        };
503
504        let encoded = record.encode();
505        let mut cursor = Cursor::new(encoded);
506        let parsed = parse_mrt_record(&mut cursor).unwrap();
507        let expected_len = parsed
508            .message
509            .encode(parsed.common_header.entry_subtype)
510            .len() as u32;
511
512        assert_eq!(parsed.common_header.length, expected_len);
513    }
514}
bgpkit_parser/parser/mrt/mrt_record.rs

bgpkit_parser/parser/mrt/
mrt_record.rs
