bgpkit_parser/parser/mrt/
mrt_elem.rs

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

bgpkit_parser/parser/mrt/
mrt_elem.rs
