bgpkit_parser/parser/mrt/
mrt_elem.rs

1#![allow(unused)]
2//! This module handles converting MRT records into individual per-prefix BGP elements.
3//!
4//! Each MRT record may contain reachability information for multiple prefixes. This module breaks
5//! down MRT records into corresponding BGP elements, and thus allowing users to more conveniently
6//! process BGP information on a per-prefix basis.
7use crate::models::*;
8use crate::parser::bgp::messages::parse_bgp_update_message;
9use crate::ParserError;
10use crate::ParserError::ParseError;
11use bytes::Bytes;
12use itertools::Itertools;
13use log::{error, warn};
14use std::collections::HashMap;
15use std::fmt::{Display, Formatter};
16use std::net::{IpAddr, Ipv4Addr};
17
18#[derive(Default, Debug, Clone)]
19pub struct Elementor {
20    pub peer_table: Option<PeerIndexTable>,
21}
22
23/// Error returned by [`Elementor::record_to_elems_iter`].
24#[derive(Debug)]
25pub enum ElemError {
26    /// The record contains a [`PeerIndexTable`]. The contained table can be
27    /// passed to [`Elementor::with_peer_table`] to create an initialized elementor.
28    UnexpectedPeerIndexTable(Box<PeerIndexTable>),
29    /// A peer table is required for processing TableDumpV2 RIB entries,
30    /// but none has been set on this elementor.
31    MissingPeerTable,
32    /// The record contains a [`RibGenericEntries`] which is not yet supported.
33    UnsupportedRibGeneric,
34}
35
36impl Display for ElemError {
37    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
38        match self {
39            ElemError::UnexpectedPeerIndexTable(_) => {
40                write!(f, "unexpected PeerIndexTable record")
41            }
42            ElemError::MissingPeerTable => {
43                write!(
44                    f,
45                    "peer table not set; call set_peer_table or use with_peer_table first"
46                )
47            }
48            ElemError::UnsupportedRibGeneric => {
49                write!(f, "RibGenericEntries not yet supported")
50            }
51        }
52    }
53}
54
55impl std::error::Error for ElemError {}
56
57// use macro_rules! <name of macro>{<Body>}
58macro_rules! get_attr_value {
59    ($a:tt, $b:expr) => {
60        if let Attribute::$a(x) = $b {
61            Some(x)
62        } else {
63            None
64        }
65    };
66}
67
68#[allow(clippy::type_complexity)]
69fn get_relevant_attributes(
70    attributes: Attributes,
71) -> (
72    Option<AsPath>,
73    Option<AsPath>,
74    Option<Origin>,
75    Option<IpAddr>,
76    Option<u32>,
77    Option<u32>,
78    Option<Vec<MetaCommunity>>,
79    bool,
80    Option<(Asn, BgpIdentifier)>,
81    Option<Nlri>,
82    Option<Nlri>,
83    Option<Asn>,
84    Option<Vec<AttrRaw>>,
85    Option<Vec<AttrRaw>>,
86) {
87    let mut as_path = None;
88    let mut as4_path = None;
89    let mut origin = None;
90    let mut next_hop = None;
91    let mut local_pref = Some(0);
92    let mut med = Some(0);
93    let mut atomic = false;
94    let mut aggregator = None;
95    let mut announced = None;
96    let mut withdrawn = None;
97    let mut otc = None;
98    let mut unknown = vec![];
99    let mut deprecated = vec![];
100
101    let mut communities_vec: Vec<MetaCommunity> = vec![];
102
103    for attr in attributes {
104        match attr {
105            AttributeValue::Origin(v) => origin = Some(v),
106            AttributeValue::AsPath {
107                path,
108                is_as4: false,
109            } => as_path = Some(path),
110            AttributeValue::AsPath { path, is_as4: true } => as4_path = Some(path),
111            AttributeValue::NextHop(v) => next_hop = Some(v),
112            AttributeValue::MultiExitDiscriminator(v) => med = Some(v),
113            AttributeValue::LocalPreference(v) => local_pref = Some(v),
114            AttributeValue::AtomicAggregate => atomic = true,
115            AttributeValue::Communities(v) => communities_vec.extend(
116                v.into_iter()
117                    .map(MetaCommunity::Plain)
118                    .collect::<Vec<MetaCommunity>>(),
119            ),
120            AttributeValue::ExtendedCommunities(v) => communities_vec.extend(
121                v.into_iter()
122                    .map(MetaCommunity::Extended)
123                    .collect::<Vec<MetaCommunity>>(),
124            ),
125            AttributeValue::Ipv6AddressSpecificExtendedCommunities(v) => communities_vec.extend(
126                v.into_iter()
127                    .map(MetaCommunity::Ipv6Extended)
128                    .collect::<Vec<MetaCommunity>>(),
129            ),
130            AttributeValue::LargeCommunities(v) => communities_vec.extend(
131                v.into_iter()
132                    .map(MetaCommunity::Large)
133                    .collect::<Vec<MetaCommunity>>(),
134            ),
135            AttributeValue::Aggregator { asn, id, .. } => aggregator = Some((asn, id)),
136            AttributeValue::MpReachNlri(nlri) => announced = Some(nlri),
137            AttributeValue::MpUnreachNlri(nlri) => withdrawn = Some(nlri),
138            AttributeValue::OnlyToCustomer(o) => otc = Some(o),
139
140            AttributeValue::Unknown(t) | AttributeValue::Raw(t) => {
141                unknown.push(t);
142            }
143            AttributeValue::Deprecated(t) => {
144                deprecated.push(t);
145            }
146
147            AttributeValue::OriginatorId(_)
148            | AttributeValue::Clusters(_)
149            | AttributeValue::Development(_)
150            | AttributeValue::LinkState(_)
151            | AttributeValue::TunnelEncapsulation(_)
152            | AttributeValue::Aigp(_)
153            | AttributeValue::BfdDiscriminator(_)
154            | AttributeValue::BgpPrefixSid(_)
155            | AttributeValue::Bier(_)
156            | AttributeValue::Sfp(_)
157            | AttributeValue::AttrSet(_) => {}
158        };
159    }
160
161    let communities = match !communities_vec.is_empty() {
162        true => Some(communities_vec),
163        false => None,
164    };
165
166    // If the next_hop is not set, we try to get it from the announced NLRI.
167    let next_hop = next_hop.or_else(|| {
168        announced.as_ref().and_then(|v| {
169            v.next_hop.as_ref().map(|h| match h {
170                NextHopAddress::Ipv4(v) => IpAddr::from(*v),
171                NextHopAddress::Ipv6(v) => IpAddr::from(*v),
172                NextHopAddress::Ipv6LinkLocal(v, _) => IpAddr::from(*v),
173                // RFC 8950: VPN next hops - return the IPv6 address part
174                NextHopAddress::VpnIpv6(_, v) => IpAddr::from(*v),
175                NextHopAddress::VpnIpv6LinkLocal(_, v, _, _) => IpAddr::from(*v),
176            })
177        })
178    });
179
180    (
181        as_path,
182        as4_path,
183        origin,
184        next_hop,
185        local_pref,
186        med,
187        communities,
188        atomic,
189        aggregator,
190        announced,
191        withdrawn,
192        otc,
193        if unknown.is_empty() {
194            None
195        } else {
196            Some(unknown)
197        },
198        if deprecated.is_empty() {
199            None
200        } else {
201            Some(deprecated)
202        },
203    )
204}
205
206fn rib_entry_to_elem(prefix: NetworkPrefix, peer: &Peer, entry: RibEntry) -> BgpElem {
207    let (
208        as_path,
209        as4_path,
210        origin,
211        next_hop,
212        local_pref,
213        med,
214        communities,
215        atomic,
216        aggregator,
217        announced,
218        _withdrawn,
219        only_to_customer,
220        unknown,
221        deprecated,
222    ) = get_relevant_attributes(entry.attributes);
223
224    let path = match (as_path, as4_path) {
225        (None, None) => None,
226        (Some(v), None) => Some(v),
227        (None, Some(v)) => Some(v),
228        (Some(v1), Some(v2)) => Some(AsPath::merge_aspath_as4path(&v1, &v2)),
229    };
230
231    let next_hop = match next_hop {
232        Some(v) => Some(v),
233        None => announced.and_then(|v| {
234            v.next_hop.map(|h| match h {
235                NextHopAddress::Ipv4(v) => IpAddr::from(v),
236                NextHopAddress::Ipv6(v) => IpAddr::from(v),
237                NextHopAddress::Ipv6LinkLocal(v, _) => IpAddr::from(v),
238                NextHopAddress::VpnIpv6(_, v) => IpAddr::from(v),
239                NextHopAddress::VpnIpv6LinkLocal(_, v, _, _) => IpAddr::from(v),
240            })
241        }),
242    };
243
244    let origin_asns = path
245        .as_ref()
246        .map(|as_path| as_path.iter_origins().collect());
247
248    BgpElem {
249        timestamp: entry.originated_time as f64,
250        elem_type: ElemType::ANNOUNCE,
251        peer_ip: peer.peer_ip,
252        peer_asn: peer.peer_asn,
253        peer_bgp_id: Some(peer.peer_bgp_id),
254        prefix,
255        next_hop,
256        as_path: path,
257        origin,
258        origin_asns,
259        local_pref,
260        med,
261        communities,
262        atomic,
263        aggr_asn: aggregator.map(|v| v.0),
264        aggr_ip: aggregator.map(|v| v.1),
265        only_to_customer,
266        unknown,
267        deprecated,
268    }
269}
270
271/// Iterator over [`BgpElem`]s produced from a single [`MrtRecord`],
272/// without requiring a mutable reference to the [`Elementor`].
273///
274/// This avoids allocating a `Vec` for the common RIB table dump case
275/// by lazily converting each [`RibEntry`] into a [`BgpElem`] on demand.
276pub enum RecordElemIter<'a> {
277    #[doc(hidden)]
278    Empty,
279    #[doc(hidden)]
280    TableDump(Option<BgpElem>),
281    #[doc(hidden)]
282    RibAfi {
283        peer_table: &'a PeerIndexTable,
284        prefix: NetworkPrefix,
285        entries: std::vec::IntoIter<RibEntry>,
286    },
287    #[doc(hidden)]
288    Bgp4Mp(BgpUpdateElemIter),
289}
290
291impl Iterator for RecordElemIter<'_> {
292    type Item = BgpElem;
293
294    fn next(&mut self) -> Option<BgpElem> {
295        match self {
296            RecordElemIter::Empty => None,
297            RecordElemIter::TableDump(elem) => elem.take(),
298            RecordElemIter::Bgp4Mp(iter) => iter.next(),
299            RecordElemIter::RibAfi {
300                peer_table,
301                prefix,
302                entries,
303            } => {
304                let entry = entries.next()?;
305                let pid = entry.peer_index;
306                match peer_table.get_peer_by_id(&pid) {
307                    Some(peer) => Some(rib_entry_to_elem(*prefix, peer, entry)),
308                    None => {
309                        error!("peer ID {} not found in peer_index table", pid);
310                        *self = RecordElemIter::Empty;
311                        None
312                    }
313                }
314            }
315        }
316    }
317
318    fn size_hint(&self) -> (usize, Option<usize>) {
319        match self {
320            RecordElemIter::Empty => (0, Some(0)),
321            RecordElemIter::TableDump(elem) => {
322                let n = elem.is_some() as usize;
323                (n, Some(n))
324            }
325            RecordElemIter::Bgp4Mp(iter) => iter.size_hint(),
326            RecordElemIter::RibAfi { entries, .. } => {
327                let len = entries.len();
328                (len, Some(len))
329            }
330        }
331    }
332}
333
334/// Iterator over [`BgpElem`]s produced from a [`BgpUpdateMessage`],
335/// avoiding allocation by lazily yielding elements from announced and
336/// withdrawn prefixes in two phases.
337pub struct BgpUpdateElemIter {
338    timestamp: f64,
339    peer_ip: IpAddr,
340    peer_asn: Asn,
341    peer_bgp_id: Option<BgpIdentifier>,
342    only_to_customer: Option<Asn>,
343    // Announce-specific shared attributes
344    path: Option<AsPath>,
345    origin_asns: Option<Vec<Asn>>,
346    origin: Option<Origin>,
347    next_hop: Option<IpAddr>,
348    local_pref: Option<u32>,
349    med: Option<u32>,
350    communities: Option<Vec<MetaCommunity>>,
351    atomic: bool,
352    aggr_asn: Option<Asn>,
353    aggr_ip: Option<BgpIdentifier>,
354    unknown: Option<Vec<AttrRaw>>,
355    deprecated: Option<Vec<AttrRaw>>,
356    // Prefix iterators (two chained sources each)
357    announced:
358        std::iter::Chain<std::vec::IntoIter<NetworkPrefix>, std::vec::IntoIter<NetworkPrefix>>,
359    withdrawn:
360        std::iter::Chain<std::vec::IntoIter<NetworkPrefix>, std::vec::IntoIter<NetworkPrefix>>,
361    in_withdrawn_phase: bool,
362}
363
364impl Iterator for BgpUpdateElemIter {
365    type Item = BgpElem;
366
367    fn next(&mut self) -> Option<BgpElem> {
368        if !self.in_withdrawn_phase {
369            if let Some(prefix) = self.announced.next() {
370                return Some(BgpElem {
371                    timestamp: self.timestamp,
372                    elem_type: ElemType::ANNOUNCE,
373                    peer_ip: self.peer_ip,
374                    peer_asn: self.peer_asn,
375                    peer_bgp_id: self.peer_bgp_id,
376                    prefix,
377                    next_hop: self.next_hop,
378                    as_path: self.path.clone(),
379                    origin: self.origin,
380                    origin_asns: self.origin_asns.clone(),
381                    local_pref: self.local_pref,
382                    med: self.med,
383                    communities: self.communities.clone(),
384                    atomic: self.atomic,
385                    aggr_asn: self.aggr_asn,
386                    aggr_ip: self.aggr_ip,
387                    only_to_customer: self.only_to_customer,
388                    unknown: self.unknown.clone(),
389                    deprecated: self.deprecated.clone(),
390                });
391            }
392            self.in_withdrawn_phase = true;
393        }
394
395        self.withdrawn.next().map(|prefix| BgpElem {
396            timestamp: self.timestamp,
397            elem_type: ElemType::WITHDRAW,
398            peer_ip: self.peer_ip,
399            peer_asn: self.peer_asn,
400            peer_bgp_id: self.peer_bgp_id,
401            prefix,
402            next_hop: None,
403            as_path: None,
404            origin: None,
405            origin_asns: None,
406            local_pref: None,
407            med: None,
408            communities: None,
409            atomic: false,
410            aggr_asn: None,
411            aggr_ip: None,
412            only_to_customer: None,
413            unknown: None,
414            deprecated: None,
415        })
416    }
417
418    fn size_hint(&self) -> (usize, Option<usize>) {
419        let (ann_lo, ann_hi) = if self.in_withdrawn_phase {
420            (0, Some(0))
421        } else {
422            self.announced.size_hint()
423        };
424        let (wd_lo, wd_hi) = self.withdrawn.size_hint();
425        (ann_lo + wd_lo, ann_hi.and_then(|a| wd_hi.map(|w| a + w)))
426    }
427}
428
429impl Elementor {
430    pub fn new() -> Elementor {
431        Self::default()
432    }
433
434    /// Sets the peer index table for the elementor.
435    ///
436    /// This method takes an MRT record and extracts the peer index table from it if the record contains one.
437    /// The peer index table is required for processing TableDumpV2 records, as it contains the mapping between
438    /// peer indices and their corresponding IP addresses and ASNs.
439    ///
440    /// # Arguments
441    ///
442    /// * `record` - An MRT record that should contain a peer index table
443    ///
444    /// # Returns
445    ///
446    /// * `Ok(())` - If the peer table was successfully extracted and set
447    /// * `Err(ParserError)` - If the record does not contain a peer index table
448    ///
449    /// # Example
450    ///
451    /// ```no_run
452    /// use bgpkit_parser::{BgpkitParser, Elementor};
453    ///
454    /// let mut parser = BgpkitParser::new("rib.dump.bz2").unwrap();
455    /// let mut elementor = Elementor::new();
456    ///
457    /// // Get the first record which should be the peer index table
458    /// if let Ok(record) = parser.next_record() {
459    ///     elementor.set_peer_table(record).unwrap();
460    /// }
461    /// ```
462    pub fn set_peer_table(&mut self, record: MrtRecord) -> Result<(), ParserError> {
463        if let MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(p)) =
464            record.message
465        {
466            self.peer_table = Some(p);
467            Ok(())
468        } else {
469            Err(ParseError("peer_table is not a PeerIndexTable".to_string()))
470        }
471    }
472
473    /// Creates an [`Elementor`] with the given [`PeerIndexTable`] already set.
474    pub fn with_peer_table(peer_table: PeerIndexTable) -> Elementor {
475        Elementor {
476            peer_table: Some(peer_table),
477        }
478    }
479
480    /// Convert a [`MrtRecord`] into an iterator of [`BgpElem`]s without
481    /// requiring `&mut self`.
482    ///
483    /// Unlike [`record_to_elems`](Elementor::record_to_elems), this method:
484    /// - Takes `&self` instead of `&mut self`, since the peer table must
485    ///   already be set via [`set_peer_table`](Elementor::set_peer_table) or
486    ///   [`with_peer_table`](Elementor::with_peer_table).
487    /// - Returns an error if the record contains a [`PeerIndexTable`] (which
488    ///   would require mutation).
489    /// - Returns a lazy [`RecordElemIter`] instead of collecting into a `Vec`,
490    ///   avoiding allocation for the common RIB table dump case.
491    ///
492    /// # Errors
493    ///
494    /// - [`ElemError::UnexpectedPeerIndexTable`] if the record is a PeerIndexTable message.
495    /// - [`ElemError::MissingPeerTable`] if the record requires a peer table but none is set.
496    pub fn record_to_elems_iter(&self, record: MrtRecord) -> Result<RecordElemIter<'_>, ElemError> {
497        let timestamp = {
498            let t = record.common_header.timestamp;
499            if let Some(micro) = &record.common_header.microsecond_timestamp {
500                let m = (*micro as f64) / 1000000.0;
501                t as f64 + m
502            } else {
503                f64::from(t)
504            }
505        };
506
507        match record.message {
508            MrtMessage::TableDumpMessage(msg) => {
509                let (
510                    as_path,
511                    _as4_path,
512                    origin,
513                    next_hop,
514                    local_pref,
515                    med,
516                    communities,
517                    atomic,
518                    aggregator,
519                    _announced,
520                    _withdrawn,
521                    only_to_customer,
522                    unknown,
523                    deprecated,
524                ) = get_relevant_attributes(msg.attributes);
525
526                let origin_asns = as_path
527                    .as_ref()
528                    .map(|as_path| as_path.iter_origins().collect());
529
530                Ok(RecordElemIter::TableDump(Some(BgpElem {
531                    timestamp: msg.originated_time as f64,
532                    elem_type: ElemType::ANNOUNCE,
533                    peer_ip: msg.peer_ip,
534                    peer_asn: msg.peer_asn,
535                    peer_bgp_id: None,
536                    prefix: msg.prefix,
537                    next_hop,
538                    as_path,
539                    origin,
540                    origin_asns,
541                    local_pref,
542                    med,
543                    communities,
544                    atomic,
545                    aggr_asn: aggregator.map(|v| v.0),
546                    aggr_ip: aggregator.map(|v| v.1),
547                    only_to_customer,
548                    unknown,
549                    deprecated,
550                })))
551            }
552
553            MrtMessage::TableDumpV2Message(msg) => match msg {
554                TableDumpV2Message::PeerIndexTable(p) => {
555                    Err(ElemError::UnexpectedPeerIndexTable(Box::new(p)))
556                }
557                TableDumpV2Message::RibAfi(t) => {
558                    let peer_table = self
559                        .peer_table
560                        .as_ref()
561                        .ok_or(ElemError::MissingPeerTable)?;
562                    Ok(RecordElemIter::RibAfi {
563                        peer_table,
564                        prefix: t.prefix,
565                        entries: t.rib_entries.into_iter(),
566                    })
567                }
568                TableDumpV2Message::RibGeneric(_) => Err(ElemError::UnsupportedRibGeneric),
569                TableDumpV2Message::GeoPeerTable(_) => Ok(RecordElemIter::Empty),
570            },
571
572            MrtMessage::Bgp4Mp(msg) => match msg {
573                Bgp4MpEnum::StateChange(_) => Ok(RecordElemIter::Empty),
574                Bgp4MpEnum::Message(v) => {
575                    match Elementor::bgp_to_elems_iter(
576                        v.bgp_message,
577                        timestamp,
578                        &v.peer_ip,
579                        &v.peer_asn,
580                    ) {
581                        Some(iter) => Ok(RecordElemIter::Bgp4Mp(iter)),
582                        None => Ok(RecordElemIter::Empty),
583                    }
584                }
585            },
586        }
587    }
588
589    /// Convert a [BgpMessage] to a vector of [BgpElem]s.
590    ///
591    /// A [BgpMessage] may include `Update`, `Open`, `Notification` or `KeepAlive` messages,
592    /// and only `Update` message contains [BgpElem]s.
593    pub fn bgp_to_elems(
594        msg: BgpMessage,
595        timestamp: f64,
596        peer_ip: &IpAddr,
597        peer_asn: &Asn,
598    ) -> Vec<BgpElem> {
599        Elementor::bgp_to_elems_iter(msg, timestamp, peer_ip, peer_asn)
600            .map(|iter| iter.collect())
601            .unwrap_or_default()
602    }
603
604    /// Convert a [BgpMessage] into an iterator of [BgpElem]s.
605    ///
606    /// Returns `None` for non-Update messages (Open, Notification, KeepAlive).
607    pub fn bgp_to_elems_iter(
608        msg: BgpMessage,
609        timestamp: f64,
610        peer_ip: &IpAddr,
611        peer_asn: &Asn,
612    ) -> Option<BgpUpdateElemIter> {
613        match msg {
614            BgpMessage::Update(msg) => Some(Elementor::bgp_update_to_elems_iter(
615                msg, timestamp, peer_ip, peer_asn,
616            )),
617            BgpMessage::Open(_) | BgpMessage::Notification(_) | BgpMessage::KeepAlive => None,
618        }
619    }
620
621    /// Convert a [BgpUpdateMessage] to a vector of [BgpElem]s.
622    pub fn bgp_update_to_elems(
623        msg: BgpUpdateMessage,
624        timestamp: f64,
625        peer_ip: &IpAddr,
626        peer_asn: &Asn,
627    ) -> Vec<BgpElem> {
628        Elementor::bgp_update_to_elems_iter(msg, timestamp, peer_ip, peer_asn).collect()
629    }
630
631    /// Convert a [BgpUpdateMessage] into a [`BgpUpdateElemIter`] that lazily
632    /// yields [BgpElem]s without allocating a `Vec`.
633    pub fn bgp_update_to_elems_iter(
634        msg: BgpUpdateMessage,
635        timestamp: f64,
636        peer_ip: &IpAddr,
637        peer_asn: &Asn,
638    ) -> BgpUpdateElemIter {
639        let (
640            as_path,
641            as4_path,
642            origin,
643            next_hop,
644            local_pref,
645            med,
646            communities,
647            atomic,
648            aggregator,
649            announced,
650            withdrawn,
651            only_to_customer,
652            unknown,
653            deprecated,
654        ) = get_relevant_attributes(msg.attributes);
655
656        let path = match (as_path, as4_path) {
657            (None, None) => None,
658            (Some(v), None) => Some(v),
659            (None, Some(v)) => Some(v),
660            (Some(v1), Some(v2)) => Some(AsPath::merge_aspath_as4path(&v1, &v2)),
661        };
662
663        let origin_asns = path
664            .as_ref()
665            .map(|as_path| as_path.iter_origins().collect());
666
667        let nlri_announced = announced.map(|n| n.prefixes).unwrap_or_default();
668        let nlri_withdrawn = withdrawn.map(|n| n.prefixes).unwrap_or_default();
669
670        BgpUpdateElemIter {
671            timestamp,
672            peer_ip: *peer_ip,
673            peer_asn: *peer_asn,
674            peer_bgp_id: None,
675            only_to_customer,
676            path,
677            origin_asns,
678            origin,
679            next_hop,
680            local_pref,
681            med,
682            communities,
683            atomic,
684            aggr_asn: aggregator.as_ref().map(|v| v.0),
685            aggr_ip: aggregator.as_ref().map(|v| v.1),
686            unknown,
687            deprecated,
688            announced: msg.announced_prefixes.into_iter().chain(nlri_announced),
689            withdrawn: msg.withdrawn_prefixes.into_iter().chain(nlri_withdrawn),
690            in_withdrawn_phase: false,
691        }
692    }
693
694    /// Convert a [MrtRecord] to a vector of [BgpElem]s.
695    ///
696    /// If the record is a [`PeerIndexTable`], it is consumed to set the internal
697    /// peer table. Errors are logged.
698    ///
699    /// For a non-mutating, lazy alternative, see
700    /// [`record_to_elems_iter`](Elementor::record_to_elems_iter).
701    pub fn record_to_elems(&mut self, record: MrtRecord) -> Vec<BgpElem> {
702        match record.message {
703            MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(_)) => {
704                self.set_peer_table(record);
705                vec![]
706            }
707            _ => match self.record_to_elems_iter(record) {
708                Ok(iter) => iter.collect(),
709                Err(e) => {
710                    error!("{}", e);
711                    vec![]
712                }
713            },
714        }
715    }
716}
717
718#[inline(always)]
719pub fn option_to_string<T>(o: &Option<T>) -> String
720where
721    T: Display,
722{
723    if let Some(v) = o {
724        v.to_string()
725    } else {
726        String::new()
727    }
728}
729
730impl From<&BgpElem> for Attributes {
731    fn from(value: &BgpElem) -> Self {
732        let mut values = Vec::<AttributeValue>::new();
733        let mut attributes = Attributes::default();
734        let prefix = value.prefix;
735
736        if value.elem_type == ElemType::WITHDRAW {
737            values.push(AttributeValue::MpUnreachNlri(Nlri::new_unreachable(prefix)));
738            attributes.extend(values);
739            return attributes;
740        }
741
742        values.push(AttributeValue::MpReachNlri(Nlri::new_reachable(
743            prefix,
744            value.next_hop,
745        )));
746
747        if let Some(v) = value.next_hop {
748            values.push(AttributeValue::NextHop(v));
749        }
750
751        if let Some(v) = value.as_path.as_ref() {
752            let is_as4 = match v.get_origin_opt() {
753                None => true,
754                Some(asn) => asn.is_four_byte(),
755            };
756            values.push(AttributeValue::AsPath {
757                path: v.clone(),
758                is_as4,
759            });
760        }
761
762        if let Some(v) = value.origin {
763            values.push(AttributeValue::Origin(v));
764        }
765
766        if let Some(v) = value.local_pref {
767            values.push(AttributeValue::LocalPreference(v));
768        }
769
770        if let Some(v) = value.med {
771            values.push(AttributeValue::MultiExitDiscriminator(v));
772        }
773
774        if let Some(v) = value.communities.as_ref() {
775            let mut communites = vec![];
776            let mut extended_communities = vec![];
777            let mut ipv6_extended_communities = vec![];
778            let mut large_communities = vec![];
779            for c in v {
780                match c {
781                    MetaCommunity::Plain(v) => communites.push(*v),
782                    MetaCommunity::Extended(v) => extended_communities.push(*v),
783                    MetaCommunity::Large(v) => large_communities.push(*v),
784                    MetaCommunity::Ipv6Extended(v) => ipv6_extended_communities.push(*v),
785                }
786            }
787            if !communites.is_empty() {
788                values.push(AttributeValue::Communities(communites));
789            }
790            if !extended_communities.is_empty() {
791                values.push(AttributeValue::ExtendedCommunities(extended_communities));
792            }
793            if !large_communities.is_empty() {
794                values.push(AttributeValue::LargeCommunities(large_communities));
795            }
796            if !ipv6_extended_communities.is_empty() {
797                values.push(AttributeValue::Ipv6AddressSpecificExtendedCommunities(
798                    ipv6_extended_communities,
799                ));
800            }
801        }
802
803        if let Some(v) = value.aggr_asn {
804            let aggregator_id = match value.aggr_ip {
805                Some(v) => v,
806                None => Ipv4Addr::UNSPECIFIED,
807            };
808            values.push(AttributeValue::Aggregator {
809                asn: v,
810                id: aggregator_id,
811                is_as4: v.is_four_byte(),
812            });
813        }
814
815        if let Some(v) = value.only_to_customer {
816            values.push(AttributeValue::OnlyToCustomer(v));
817        }
818
819        if let Some(v) = value.unknown.as_ref() {
820            for t in v {
821                values.push(AttributeValue::Unknown(t.clone()));
822            }
823        }
824
825        if let Some(v) = value.deprecated.as_ref() {
826            for t in v {
827                values.push(AttributeValue::Deprecated(t.clone()));
828            }
829        }
830
831        attributes.extend(values);
832        attributes
833    }
834}
835
836#[cfg(test)]
837mod tests {
838    use super::*;
839    use crate::BgpkitParser;
840    use std::net::{Ipv4Addr, Ipv6Addr};
841    use std::str::FromStr;
842
843    #[test]
844    fn test_option_to_string() {
845        let o1 = Some(1);
846        let o2: Option<u32> = None;
847        assert_eq!(option_to_string(&o1), "1");
848        assert_eq!(option_to_string(&o2), "");
849    }
850
851    #[test]
852    fn test_record_to_elems() {
853        let url_table_dump_v1 = "https://data.ris.ripe.net/rrc00/2003.01/bview.20030101.0000.gz";
854        let url_table_dump_v2 = "https://data.ris.ripe.net/rrc00/2023.01/bview.20230101.0000.gz";
855        let url_bgp4mp = "https://data.ris.ripe.net/rrc00/2021.10/updates.20211001.0000.gz";
856
857        let mut elementor = Elementor::new();
858        let parser = BgpkitParser::new(url_table_dump_v1).unwrap();
859        let mut record_iter = parser.into_record_iter();
860        let record = record_iter.next().unwrap();
861        let elems = elementor.record_to_elems(record);
862        assert_eq!(elems.len(), 1);
863
864        let parser = BgpkitParser::new(url_table_dump_v2).unwrap();
865        let mut record_iter = parser.into_record_iter();
866        let peer_index_table = record_iter.next().unwrap();
867        let _elems = elementor.record_to_elems(peer_index_table);
868        let record = record_iter.next().unwrap();
869        let elems = elementor.record_to_elems(record);
870        assert!(!elems.is_empty());
871
872        let parser = BgpkitParser::new(url_bgp4mp).unwrap();
873        let mut record_iter = parser.into_record_iter();
874        let record = record_iter.next().unwrap();
875        let elems = elementor.record_to_elems(record);
876        assert!(!elems.is_empty());
877    }
878
879    #[test]
880    fn test_attributes_from_bgp_elem() {
881        let mut elem = BgpElem {
882            timestamp: 0.0,
883            elem_type: ElemType::ANNOUNCE,
884            peer_ip: IpAddr::from_str("10.0.0.1").unwrap(),
885            peer_asn: Asn::new_32bit(65000),
886            peer_bgp_id: None,
887            prefix: NetworkPrefix::from_str("10.0.1.0/24").unwrap(),
888            next_hop: Some(IpAddr::from_str("10.0.0.2").unwrap()),
889            as_path: Some(AsPath::from_sequence([65000, 65001, 65002])),
890            origin: Some(Origin::EGP),
891            origin_asns: Some(vec![Asn::new_32bit(65000)]),
892            local_pref: Some(100),
893            med: Some(200),
894            communities: Some(vec![
895                MetaCommunity::Plain(Community::NoAdvertise),
896                MetaCommunity::Extended(ExtendedCommunity::Raw([0, 0, 0, 0, 0, 0, 0, 0])),
897                MetaCommunity::Large(LargeCommunity {
898                    global_admin: 0,
899                    local_data: [0, 0],
900                }),
901                MetaCommunity::Ipv6Extended(Ipv6AddrExtCommunity {
902                    community_type: ExtendedCommunityType::TransitiveTwoOctetAs,
903                    subtype: 0,
904                    global_admin: Ipv6Addr::from_str("2001:db8::").unwrap(),
905                    local_admin: [0, 0],
906                }),
907            ]),
908            atomic: false,
909            aggr_asn: Some(Asn::new_32bit(65000)),
910            aggr_ip: Some(Ipv4Addr::from_str("10.2.0.0").unwrap()),
911            only_to_customer: Some(Asn::new_32bit(65000)),
912            unknown: Some(vec![AttrRaw {
913                code: AttrType::RESERVED.into(),
914                bytes: Bytes::new(),
915            }]),
916            deprecated: Some(vec![AttrRaw {
917                code: AttrType::RESERVED.into(),
918                bytes: Bytes::new(),
919            }]),
920        };
921
922        let _attributes = Attributes::from(&elem);
923        elem.elem_type = ElemType::WITHDRAW;
924        let _attributes = Attributes::from(&elem);
925    }
926
927    #[test]
928    fn test_get_relevant_attributes() {
929        let attributes = vec![
930            AttributeValue::Origin(Origin::IGP),
931            AttributeValue::AsPath {
932                path: AsPath::from_sequence([65000, 65001, 65002]),
933                is_as4: true,
934            },
935            AttributeValue::NextHop(IpAddr::from_str("10.0.0.1").unwrap()),
936            AttributeValue::MultiExitDiscriminator(100),
937            AttributeValue::LocalPreference(200),
938            AttributeValue::AtomicAggregate,
939            AttributeValue::Aggregator {
940                asn: Asn::new_32bit(65000),
941                id: Ipv4Addr::from_str("10.0.0.1").unwrap(),
942                is_as4: false,
943            },
944            AttributeValue::Communities(vec![Community::NoExport]),
945            AttributeValue::ExtendedCommunities(vec![ExtendedCommunity::Raw([
946                0, 0, 0, 0, 0, 0, 0, 0,
947            ])]),
948            AttributeValue::LargeCommunities(vec![LargeCommunity {
949                global_admin: 0,
950                local_data: [0, 0],
951            }]),
952            AttributeValue::Ipv6AddressSpecificExtendedCommunities(vec![Ipv6AddrExtCommunity {
953                community_type: ExtendedCommunityType::TransitiveTwoOctetAs,
954                subtype: 0,
955                global_admin: Ipv6Addr::from_str("2001:db8::").unwrap(),
956                local_admin: [0, 0],
957            }]),
958            AttributeValue::MpReachNlri(Nlri::new_reachable(
959                NetworkPrefix::from_str("10.0.0.0/24").unwrap(),
960                Some(IpAddr::from_str("10.0.0.1").unwrap()),
961            )),
962            AttributeValue::MpUnreachNlri(Nlri::new_unreachable(
963                NetworkPrefix::from_str("10.0.0.0/24").unwrap(),
964            )),
965            AttributeValue::OnlyToCustomer(Asn::new_32bit(65000)),
966            AttributeValue::Unknown(AttrRaw {
967                code: AttrType::RESERVED.into(),
968                bytes: Bytes::new(),
969            }),
970            AttributeValue::Deprecated(AttrRaw {
971                code: AttrType::RESERVED.into(),
972                bytes: Bytes::new(),
973            }),
974        ]
975        .into_iter()
976        .map(Attribute::from)
977        .collect::<Vec<Attribute>>();
978
979        let attributes = Attributes::from(attributes);
980
981        let (
982            _as_path,
983            _as4_path, // Table dump v1 does not have 4-byte AS number
984            _origin,
985            _next_hop,
986            _local_pref,
987            _med,
988            _communities,
989            _atomic,
990            _aggregator,
991            _announced,
992            _withdrawn,
993            _only_to_customer,
994            _unknown,
995            _deprecated,
996        ) = get_relevant_attributes(attributes);
997    }
998
999    #[test]
1000    fn test_next_hop_from_nlri() {
1001        let attributes = vec![AttributeValue::NextHop(
1002            IpAddr::from_str("10.0.0.1").unwrap(),
1003        )]
1004        .into_iter()
1005        .map(Attribute::from)
1006        .collect::<Vec<Attribute>>();
1007
1008        let attributes = Attributes::from(attributes);
1009
1010        let (
1011            _as_path,
1012            _as4_path, // Table dump v1 does not have 4-byte AS number
1013            _origin,
1014            next_hop,
1015            _local_pref,
1016            _med,
1017            _communities,
1018            _atomic,
1019            _aggregator,
1020            _announced,
1021            _withdrawn,
1022            _only_to_customer,
1023            _unknown,
1024            _deprecated,
1025        ) = get_relevant_attributes(attributes);
1026
1027        assert_eq!(next_hop, Some(IpAddr::from_str("10.0.0.1").unwrap()));
1028
1029        let attributes = vec![AttributeValue::MpReachNlri(Nlri::new_reachable(
1030            NetworkPrefix::from_str("10.0.0.0/24").unwrap(),
1031            Some(IpAddr::from_str("10.0.0.2").unwrap()),
1032        ))]
1033        .into_iter()
1034        .map(Attribute::from)
1035        .collect::<Vec<Attribute>>();
1036
1037        let attributes = Attributes::from(attributes);
1038
1039        let (
1040            _as_path,
1041            _as4_path, // Table dump v1 does not have 4-byte AS number
1042            _origin,
1043            next_hop,
1044            _local_pref,
1045            _med,
1046            _communities,
1047            _atomic,
1048            _aggregator,
1049            _announced,
1050            _withdrawn,
1051            _only_to_customer,
1052            _unknown,
1053            _deprecated,
1054        ) = get_relevant_attributes(attributes);
1055
1056        assert_eq!(next_hop, Some(IpAddr::from_str("10.0.0.2").unwrap()));
1057    }
1058
1059    #[test]
1060    fn test_record_to_elems_iter_equivalence_tabledumpv2_small() {
1061        // rib-example-small.bz2 is a TableDumpV2 file (starts with PeerIndexTable)
1062        let url = "https://spaces.bgpkit.org/parser/rib-example-small.bz2";
1063
1064        let mut elementor = Elementor::new();
1065        let parser = BgpkitParser::new(url).unwrap();
1066        let mut record_iter = parser.into_record_iter();
1067
1068        // Skip the PeerIndexTable
1069        let peer_index_table = record_iter.next().unwrap();
1070        let _ = elementor.record_to_elems(peer_index_table);
1071
1072        // Process the first RIB entry
1073        let record = record_iter.next().unwrap();
1074        let elems_vec = elementor.record_to_elems(record.clone());
1075        let elems_iter: Vec<BgpElem> = elementor.record_to_elems_iter(record).unwrap().collect();
1076        assert_eq!(elems_vec, elems_iter);
1077        assert!(!elems_vec.is_empty());
1078    }
1079
1080    #[test]
1081    fn test_record_to_elems_iter_equivalence_bgp4mp() {
1082        let url = "https://spaces.bgpkit.org/parser/update-example.gz";
1083
1084        let mut elementor = Elementor::new();
1085        let parser = BgpkitParser::new(url).unwrap();
1086        let mut record_iter = parser.into_record_iter();
1087        let record = record_iter.next().unwrap();
1088
1089        let elems_vec = elementor.record_to_elems(record.clone());
1090        let elems_iter: Vec<BgpElem> = elementor.record_to_elems_iter(record).unwrap().collect();
1091        assert_eq!(elems_vec, elems_iter);
1092        assert!(!elems_vec.is_empty());
1093    }
1094
1095    #[test]
1096    #[ignore = "requires large RIB file download"]
1097    fn test_record_to_elems_iter_equivalence_tabledumpv2() {
1098        let url = "https://data.ris.ripe.net/rrc00/2023.01/bview.20230101.0000.gz";
1099
1100        let mut elementor = Elementor::new();
1101        let parser = BgpkitParser::new(url).unwrap();
1102        let mut record_iter = parser.into_record_iter();
1103
1104        let peer_index_table = record_iter.next().unwrap();
1105        let _ = elementor.record_to_elems(peer_index_table);
1106
1107        let record = record_iter.next().unwrap();
1108        let elems_vec = elementor.record_to_elems(record.clone());
1109        let elems_iter: Vec<BgpElem> = elementor.record_to_elems_iter(record).unwrap().collect();
1110        assert_eq!(elems_vec, elems_iter);
1111        assert!(!elems_vec.is_empty());
1112    }
1113
1114    #[test]
1115    fn test_record_to_elems_iter_tabledumpv2_with_peer_table() {
1116        let url = "https://spaces.bgpkit.org/parser/rib-example-small.bz2";
1117
1118        let parser = BgpkitParser::new(url).unwrap();
1119        let mut record_iter = parser.into_record_iter();
1120
1121        let peer_index_table = record_iter.next().unwrap();
1122        let mut elementor = Elementor::with_peer_table(
1123            if let MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(pit)) =
1124                peer_index_table.message
1125            {
1126                pit
1127            } else {
1128                panic!("Expected PeerIndexTable");
1129            },
1130        );
1131
1132        let record = record_iter.next().unwrap();
1133        let elems_vec = elementor.record_to_elems(record.clone());
1134        let elems_iter: Vec<BgpElem> = elementor.record_to_elems_iter(record).unwrap().collect();
1135        assert_eq!(elems_vec, elems_iter);
1136        assert!(!elems_vec.is_empty());
1137    }
1138
1139    #[test]
1140    fn test_record_to_elems_iter_error_unexpected_peer_index_table() {
1141        let url = "https://spaces.bgpkit.org/parser/rib-example-small.bz2";
1142
1143        let elementor = Elementor::new();
1144        let parser = BgpkitParser::new(url).unwrap();
1145        let mut record_iter = parser.into_record_iter();
1146        let record = record_iter.next().unwrap();
1147
1148        let result = elementor.record_to_elems_iter(record);
1149        assert!(matches!(
1150            result,
1151            Err(ElemError::UnexpectedPeerIndexTable(_))
1152        ));
1153    }
1154
1155    #[test]
1156    fn test_record_to_elems_iter_error_missing_peer_table() {
1157        // rib-example-small.bz2 is a TableDumpV2 file (starts with PeerIndexTable)
1158        let url = "https://spaces.bgpkit.org/parser/rib-example-small.bz2";
1159
1160        let elementor = Elementor::new();
1161        let parser = BgpkitParser::new(url).unwrap();
1162        let mut record_iter = parser.into_record_iter();
1163
1164        // Skip the PeerIndexTable without consuming it via record_to_elems
1165        // which would set the peer table in the elementor
1166        let _peer_index_table = record_iter.next().unwrap();
1167
1168        // Now try to process a RIB entry without having set the peer table
1169        let record = record_iter.next().unwrap();
1170        let result = elementor.record_to_elems_iter(record);
1171        assert!(matches!(result, Err(ElemError::MissingPeerTable)));
1172    }
1173
1174    #[test]
1175    fn test_bgp_to_elems_iter_equivalence() {
1176        let timestamp = 0.0;
1177        let peer_ip = IpAddr::from_str("10.0.0.1").unwrap();
1178        let peer_asn = Asn::new_32bit(65000);
1179
1180        let attributes = vec![
1181            AttributeValue::Origin(Origin::IGP),
1182            AttributeValue::AsPath {
1183                path: AsPath::from_sequence([65000, 65001, 65002]),
1184                is_as4: false,
1185            },
1186            AttributeValue::NextHop(peer_ip),
1187        ]
1188        .into_iter()
1189        .map(Attribute::from)
1190        .collect::<Vec<Attribute>>();
1191        let attributes = Attributes::from(attributes);
1192
1193        let announced_prefixes = vec![NetworkPrefix::from_str("10.0.0.0/24").unwrap()];
1194
1195        let bgp_message = BgpMessage::Update(BgpUpdateMessage {
1196            attributes,
1197            announced_prefixes,
1198            withdrawn_prefixes: vec![],
1199        });
1200
1201        let elems_vec =
1202            Elementor::bgp_to_elems(bgp_message.clone(), timestamp, &peer_ip, &peer_asn);
1203        let elems_iter: Vec<BgpElem> =
1204            Elementor::bgp_to_elems_iter(bgp_message, timestamp, &peer_ip, &peer_asn)
1205                .unwrap()
1206                .collect();
1207        assert_eq!(elems_vec, elems_iter);
1208        assert_eq!(elems_vec.len(), 1);
1209    }
1210
1211    #[test]
1212    fn test_bgp_to_elems_iter_non_update_messages() {
1213        use std::net::Ipv4Addr;
1214
1215        let timestamp = 0.0;
1216        let peer_ip = IpAddr::from_str("10.0.0.1").unwrap();
1217        let peer_asn = Asn::new_32bit(65000);
1218
1219        let open_msg = BgpOpenMessage {
1220            version: 4,
1221            asn: Asn::new_32bit(1),
1222            hold_time: 180,
1223            bgp_identifier: Ipv4Addr::new(192, 0, 2, 1),
1224            extended_length: false,
1225            opt_params: vec![],
1226        };
1227        assert!(Elementor::bgp_to_elems_iter(
1228            BgpMessage::Open(open_msg),
1229            timestamp,
1230            &peer_ip,
1231            &peer_asn
1232        )
1233        .is_none());
1234
1235        let notification_msg = BgpNotificationMessage {
1236            error: BgpError::Unknown(0, 0),
1237            data: vec![],
1238        };
1239        assert!(Elementor::bgp_to_elems_iter(
1240            BgpMessage::Notification(notification_msg),
1241            timestamp,
1242            &peer_ip,
1243            &peer_asn
1244        )
1245        .is_none());
1246
1247        assert!(Elementor::bgp_to_elems_iter(
1248            BgpMessage::KeepAlive,
1249            timestamp,
1250            &peer_ip,
1251            &peer_asn
1252        )
1253        .is_none());
1254    }
1255
1256    #[test]
1257    fn test_bgp_update_to_elems_iter_equivalence() {
1258        let timestamp = 0.0;
1259        let peer_ip = IpAddr::from_str("10.0.0.1").unwrap();
1260        let peer_asn = Asn::new_32bit(65000);
1261
1262        let attributes = vec![
1263            AttributeValue::Origin(Origin::IGP),
1264            AttributeValue::AsPath {
1265                path: AsPath::from_sequence([65000, 65001, 65002]),
1266                is_as4: false,
1267            },
1268            AttributeValue::NextHop(peer_ip),
1269        ]
1270        .into_iter()
1271        .map(Attribute::from)
1272        .collect::<Vec<Attribute>>();
1273        let attributes = Attributes::from(attributes);
1274
1275        let announced_prefixes = vec![NetworkPrefix::from_str("10.0.0.0/24").unwrap()];
1276        let withdrawn_prefixes = vec![NetworkPrefix::from_str("10.0.1.0/24").unwrap()];
1277
1278        let update = BgpUpdateMessage {
1279            attributes,
1280            announced_prefixes,
1281            withdrawn_prefixes,
1282        };
1283
1284        let elems_vec =
1285            Elementor::bgp_update_to_elems(update.clone(), timestamp, &peer_ip, &peer_asn);
1286        let elems_iter: Vec<BgpElem> =
1287            Elementor::bgp_update_to_elems_iter(update, timestamp, &peer_ip, &peer_asn).collect();
1288        assert_eq!(elems_vec, elems_iter);
1289        assert_eq!(elems_vec.len(), 2);
1290    }
1291
1292    #[test]
1293    fn test_record_elem_iter_size_hint() {
1294        use std::collections::HashMap;
1295
1296        let peer_table = PeerIndexTable {
1297            collector_bgp_id: BgpIdentifier::from_str("10.0.0.1").unwrap(),
1298            view_name: "".to_string(),
1299            id_peer_map: HashMap::new(),
1300            peer_ip_id_map: HashMap::new(),
1301        };
1302
1303        let entries: Vec<RibEntry> = vec![];
1304        let iter = RecordElemIter::RibAfi {
1305            peer_table: &peer_table,
1306            prefix: NetworkPrefix::from_str("10.0.0.0/24").unwrap(),
1307            entries: entries.into_iter(),
1308        };
1309        assert_eq!(iter.size_hint(), (0, Some(0)));
1310
1311        let entries: Vec<RibEntry> = (0..5)
1312            .map(|i| RibEntry {
1313                peer_index: i as u16,
1314                originated_time: 0,
1315                path_id: None,
1316                attributes: Attributes::default(),
1317            })
1318            .collect();
1319        let iter = RecordElemIter::RibAfi {
1320            peer_table: &peer_table,
1321            prefix: NetworkPrefix::from_str("10.0.0.0/24").unwrap(),
1322            entries: entries.into_iter(),
1323        };
1324        assert_eq!(iter.size_hint(), (5, Some(5)));
1325    }
1326}
bgpkit_parser/parser/mrt/mrt_elem.rs

bgpkit_parser/parser/mrt/
mrt_elem.rs
