1use crate::error::{ParserError, ParserErrorWithBytes};
2use crate::models::*;
3use crate::parser::bgp::attributes::{parse_as_path, parse_nlri, AttributeValidationState};
4use crate::parser::bgp::messages::read_and_validate_bgp_marker;
5use crate::parser::iters::{handle_record_parse_error, write_mrt_core_dump};
6use crate::parser::mrt::messages::bgp4mp::{bgp4mp_message_payload_len, is_short_zebra_open};
7use crate::parser::mrt::messages::legacy_bgp::{
8 BGP_KEEPALIVE, BGP_NOTIFY, BGP_OPEN, BGP_STATE_CHANGE, BGP_UPDATE,
9};
10use crate::parser::mrt::messages::table_dump_v2::rib_entry_min_len;
11use crate::parser::mrt::mrt_record::raw_record_uses_zebra_compat;
12use crate::parser::{
13 chunk_mrt_record, parse_legacy_bgp, parse_nlri_list, BgpkitParser, Filterable, ReadUtils,
14};
15use bytes::{Buf, Bytes};
16use ipnet::IpNet;
17use log::{error, warn};
18use std::io::Read;
19use std::net::{IpAddr, Ipv4Addr};
20use std::sync::Arc;
21
22#[derive(Default)]
23struct RouteAttributes {
24 as_path: Option<Arc<AsPath>>,
25 announced: Vec<NetworkPrefix>,
26 withdrawn: Vec<NetworkPrefix>,
27}
28
29struct RouteAttributeContext<'a> {
30 afi: Option<Afi>,
31 safi: Option<Safi>,
32 prefixes: Option<&'a [NetworkPrefix]>,
33 is_announcement: Option<bool>,
34 has_standard_nlri: bool,
35}
36
37fn merge_as_path(as_path: Option<AsPath>, as4_path: Option<AsPath>) -> Option<Arc<AsPath>> {
38 let path = match (as_path, as4_path) {
39 (None, None) => None,
40 (Some(path), None) | (None, Some(path)) => Some(path),
41 (Some(path), Some(as4_path)) => Some(AsPath::merge_aspath_as4path(&path, &as4_path)),
42 };
43 path.map(Arc::new)
44}
45
46fn parse_route_attributes(
47 mut data: Bytes,
48 asn_len: &AsnLength,
49 add_path: bool,
50 ctx: RouteAttributeContext<'_>,
51) -> Result<RouteAttributes, ParserError> {
52 let mut validation = AttributeValidationState::new();
53 let mut as_path = None;
54 let mut as4_path = None;
55 let mut announced = Vec::new();
56 let mut withdrawn = Vec::new();
57
58 while data.remaining() >= 3 {
59 let flags = AttrFlags::from_bits_retain(data.read_u8()?);
60 let raw_attr_type = data.read_u8()?;
61 let attr_length = if flags.contains(AttrFlags::EXTENDED) {
62 data.read_u16()? as usize
63 } else {
64 data.read_u8()? as usize
65 };
66 let attr_type = AttrType::from(raw_attr_type);
67 let partial = validation.observe_header(raw_attr_type, attr_type, flags, attr_length);
68
69 if data.remaining() < attr_length {
70 warn!(
71 "{:?} attribute encodes a length ({}) that is longer than the remaining attribute data ({}). Skipping remaining attribute data for BGP message",
72 attr_type,
73 attr_length,
74 data.remaining()
75 );
76 break;
77 }
78
79 let attr_data = data.split_to(attr_length);
80 let result = match attr_type {
81 AttrType::AS_PATH => parse_as_path(attr_data, asn_len).map(|path| {
82 as_path = Some(path);
83 }),
84 AttrType::AS4_PATH => parse_as_path(attr_data, &AsnLength::Bits32).map(|path| {
85 as4_path = Some(path);
86 }),
87 AttrType::MP_REACHABLE_NLRI => parse_nlri(
88 attr_data,
89 &ctx.afi,
90 &ctx.safi,
91 &ctx.prefixes,
92 true,
93 add_path,
94 )
95 .map(|attr| {
96 if let AttributeValue::MpReachNlri(nlri) = attr {
97 announced = nlri.prefixes;
98 }
99 }),
100 AttrType::MP_UNREACHABLE_NLRI => parse_nlri(
101 attr_data,
102 &ctx.afi,
103 &ctx.safi,
104 &ctx.prefixes,
105 false,
106 add_path,
107 )
108 .map(|attr| {
109 if let AttributeValue::MpUnreachNlri(nlri) = attr {
110 withdrawn = nlri.prefixes;
111 }
112 }),
113 _ => Ok(()),
114 };
115
116 if let Err(err) = result {
117 validation.observe_parse_error(attr_type, partial, &err);
118 }
119 }
120
121 let is_announcement = ctx
122 .is_announcement
123 .unwrap_or(ctx.has_standard_nlri || validation.has_attr(AttrType::MP_REACHABLE_NLRI));
124 validation.check_mandatory_attributes(is_announcement, ctx.has_standard_nlri);
125 let _warnings = validation.finish();
126 Ok(RouteAttributes {
127 as_path: merge_as_path(as_path, as4_path),
128 announced,
129 withdrawn,
130 })
131}
132
133fn record_timestamp(common_header: &CommonHeader) -> f64 {
134 match common_header.microsecond_timestamp {
135 Some(microseconds) => common_header.timestamp as f64 + microseconds as f64 / 1_000_000.0,
136 None => common_header.timestamp as f64,
137 }
138}
139
140struct RouteUpdateIter {
141 timestamp: f64,
142 peer_ip: IpAddr,
143 peer_asn: Asn,
144 as_path: Option<Arc<AsPath>>,
145 announced:
146 std::iter::Chain<std::vec::IntoIter<NetworkPrefix>, std::vec::IntoIter<NetworkPrefix>>,
147 withdrawn:
148 std::iter::Chain<std::vec::IntoIter<NetworkPrefix>, std::vec::IntoIter<NetworkPrefix>>,
149 in_withdrawn_phase: bool,
150}
151
152impl RouteUpdateIter {
153 fn next_route(&mut self) -> Option<BgpRouteElem> {
154 if !self.in_withdrawn_phase {
155 if let Some(prefix) = self.announced.next() {
156 return Some(BgpRouteElem {
157 timestamp: self.timestamp,
158 elem_type: ElemType::ANNOUNCE,
159 peer_ip: self.peer_ip,
160 peer_asn: self.peer_asn,
161 prefix,
162 as_path: self.as_path.clone(),
163 });
164 }
165 self.in_withdrawn_phase = true;
166 }
167
168 self.withdrawn.next().map(|prefix| BgpRouteElem {
169 timestamp: self.timestamp,
170 elem_type: ElemType::WITHDRAW,
171 peer_ip: self.peer_ip,
172 peer_asn: self.peer_asn,
173 prefix,
174 as_path: None,
175 })
176 }
177}
178
179#[derive(Clone, Default)]
180struct RoutePeerTable {
181 peers: Arc<[Peer]>,
182}
183
184impl RoutePeerTable {
185 fn get_peer_by_id(&self, peer_index: u16) -> Option<Peer> {
186 self.peers.get(peer_index as usize).copied()
187 }
188}
189
190fn parse_route_peer_table(mut data: Bytes) -> Result<RoutePeerTable, ParserError> {
191 let _collector_bgp_id = data.read_u32()?;
192 let view_name_length = data.read_u16()? as usize;
193 data.has_n_remaining(view_name_length)?;
194 data.advance(view_name_length);
195
196 let peer_count = data.read_u16()? as usize;
197 let mut peers = Vec::with_capacity(peer_count);
198 for _ in 0..peer_count {
199 let peer_type = PeerType::from_bits_retain(data.read_u8()?);
200 let afi = if peer_type.contains(PeerType::ADDRESS_FAMILY_IPV6) {
201 Afi::Ipv6
202 } else {
203 Afi::Ipv4
204 };
205 let asn_len = if peer_type.contains(PeerType::AS_SIZE_32BIT) {
206 AsnLength::Bits32
207 } else {
208 AsnLength::Bits16
209 };
210
211 let peer_bgp_id = data.read_ipv4_address()?;
212 let peer_ip = data.read_address(&afi)?;
213 let peer_asn = data.read_asn(asn_len)?;
214 peers.push(Peer {
215 peer_type,
216 peer_bgp_id,
217 peer_ip,
218 peer_asn,
219 });
220 }
221
222 Ok(RoutePeerTable {
223 peers: Arc::from(peers),
224 })
225}
226
227#[derive(Default)]
228enum RouteRecordIter {
229 #[default]
230 Empty,
231 Update(RouteUpdateIter),
232 TableDump(RouteTableDumpIter),
233 RibAfi(RouteRibAfiIter),
234}
235
236impl RouteRecordIter {
237 fn next_route(&mut self) -> Result<Option<BgpRouteElem>, ParserError> {
238 match self {
239 RouteRecordIter::Empty => Ok(None),
240 RouteRecordIter::Update(iter) => Ok(iter.next_route()),
241 RouteRecordIter::TableDump(iter) => iter.next_route(),
242 RouteRecordIter::RibAfi(iter) => iter.next_route(),
243 }
244 }
245}
246
247struct RouteTableDumpIter {
248 data: Bytes,
249 afi: Afi,
250}
251
252impl RouteTableDumpIter {
253 fn next_route(&mut self) -> Result<Option<BgpRouteElem>, ParserError> {
254 if self.data.is_empty() {
255 return Ok(None);
256 }
257
258 let prefix = match self.afi {
259 Afi::Ipv4 => self.data.read_ipv4_prefix().map(IpNet::V4),
260 Afi::Ipv6 => self.data.read_ipv6_prefix().map(IpNet::V6),
261 Afi::LinkState => unreachable!("table dumps carry IPv4 or IPv6 prefixes only"),
263 }?;
264 let _status = self.data.read_u8()?;
265 let originated_time = self.data.read_u32()? as f64;
266 let peer_ip = self.data.read_address(&self.afi)?;
267 let peer_asn = Asn::new_16bit(self.data.read_u16()?);
268 let attribute_length = self.data.read_u16()? as usize;
269 self.data.has_n_remaining(attribute_length)?;
270 let attrs = parse_route_attributes(
271 self.data.split_to(attribute_length),
272 &AsnLength::Bits16,
273 false,
274 RouteAttributeContext {
275 afi: None,
276 safi: None,
277 prefixes: None,
278 is_announcement: Some(true),
279 has_standard_nlri: self.afi == Afi::Ipv4,
280 },
281 )?;
282
283 Ok(Some(BgpRouteElem {
284 timestamp: originated_time,
285 elem_type: ElemType::ANNOUNCE,
286 peer_ip,
287 peer_asn,
288 prefix: NetworkPrefix::new(prefix, None),
289 as_path: attrs.as_path,
290 }))
291 }
292}
293
294struct RouteRibAfiIter {
295 data: Bytes,
296 peer_table: RoutePeerTable,
297 afi: Afi,
298 safi: Safi,
299 is_add_path: bool,
300 prefix: NetworkPrefix,
301 remaining_entries: u16,
302}
303
304impl RouteRibAfiIter {
305 fn next_route(&mut self) -> Result<Option<BgpRouteElem>, ParserError> {
306 while self.remaining_entries > 0 {
307 if self.data.remaining() < rib_entry_min_len(self.is_add_path) {
308 warn!("early break due to truncated msg while parsing RIB AFI entries");
309 self.remaining_entries = 0;
310 return Ok(None);
311 }
312
313 self.remaining_entries -= 1;
314 let peer_index = self.data.read_u16()?;
315 let originated_time = self.data.read_u32()? as f64;
316 let _path_id = if self.is_add_path {
317 Some(self.data.read_u32()?)
318 } else {
319 None
320 };
321 let attribute_length = self.data.read_u16()? as usize;
322 if self.data.remaining() < attribute_length {
323 warn!(
324 "early break due to truncated attribute payload while parsing RIB AFI entries: expected {} bytes, have {} bytes available",
325 attribute_length,
326 self.data.remaining()
327 );
328 self.remaining_entries = 0;
329 return Ok(None);
330 }
331
332 let prefixes = [self.prefix];
333 let attrs = parse_route_attributes(
334 self.data.split_to(attribute_length),
335 &AsnLength::Bits32,
336 self.is_add_path,
337 RouteAttributeContext {
338 afi: Some(self.afi),
339 safi: Some(self.safi),
340 prefixes: Some(&prefixes),
341 is_announcement: Some(true),
342 has_standard_nlri: self.afi == Afi::Ipv4,
343 },
344 )?;
345 let Some(peer) = self.peer_table.get_peer_by_id(peer_index) else {
346 error!("peer ID {} not found in peer_index table", peer_index);
347 continue;
348 };
349
350 return Ok(Some(BgpRouteElem {
351 timestamp: originated_time,
352 elem_type: ElemType::ANNOUNCE,
353 peer_ip: peer.peer_ip,
354 peer_asn: peer.peer_asn,
355 prefix: self.prefix,
356 as_path: attrs.as_path,
357 }));
358 }
359
360 Ok(None)
361 }
362}
363
364fn parse_bgp_update_routes(
365 mut input: Bytes,
366 add_path: bool,
367 asn_len: &AsnLength,
368 timestamp: f64,
369 peer_ip: IpAddr,
370 peer_asn: Asn,
371) -> Result<RouteUpdateIter, ParserError> {
372 let withdrawn_len = input.read_u16()? as usize;
373 input.has_n_remaining(withdrawn_len)?;
374 let withdrawn_prefixes = parse_nlri_list(input.split_to(withdrawn_len), add_path, &Afi::Ipv4)?;
375
376 let attribute_length = input.read_u16()? as usize;
377 input.has_n_remaining(attribute_length)?;
378 let attribute_bytes = input.split_to(attribute_length);
379 let announced_prefixes = parse_nlri_list(input, add_path, &Afi::Ipv4)?;
380 let attributes = parse_route_attributes(
381 attribute_bytes,
382 asn_len,
383 add_path,
384 RouteAttributeContext {
385 afi: None,
386 safi: None,
387 prefixes: None,
388 is_announcement: None,
389 has_standard_nlri: !announced_prefixes.is_empty(),
390 },
391 )?;
392
393 Ok(RouteUpdateIter {
394 timestamp,
395 peer_ip,
396 peer_asn,
397 as_path: attributes.as_path,
398 announced: announced_prefixes.into_iter().chain(attributes.announced),
399 withdrawn: withdrawn_prefixes.into_iter().chain(attributes.withdrawn),
400 in_withdrawn_phase: false,
401 })
402}
403
404fn parse_bgp_message_routes(
405 mut data: Bytes,
406 add_path: bool,
407 asn_len: &AsnLength,
408 timestamp: f64,
409 peer_ip: IpAddr,
410 peer_asn: Asn,
411) -> Result<RouteRecordIter, ParserError> {
412 let total_size = data.len();
413 data.has_n_remaining(19)?;
414 read_and_validate_bgp_marker(&mut data)?;
415 let length = data.read_u16()?;
416 if !(19..=65_535).contains(&length) {
417 return Err(ParserError::ParseError(format!(
418 "invalid BGP message length {length}"
419 )));
420 }
421
422 let bgp_msg_length = if length as usize > total_size {
423 total_size - 19
424 } else {
425 length as usize - 19
426 };
427 let msg_type = BgpMessageType::try_from(data.read_u8()?)
428 .map_err(|_| ParserError::ParseError("Unknown BGP Message Type".to_string()))?;
429
430 if matches!(msg_type, BgpMessageType::OPEN | BgpMessageType::KEEPALIVE) && length > 4096 {
431 return Err(ParserError::ParseError(format!(
432 "BGP {msg_type:?} message length {length} exceeds maximum allowed 4096 bytes (RFC 8654)"
433 )));
434 }
435
436 if data.remaining() != bgp_msg_length {
437 warn!(
438 "BGP message length {} does not match the actual length {} (parsing BGP message)",
439 bgp_msg_length,
440 data.remaining()
441 );
442 }
443 data.has_n_remaining(bgp_msg_length)?;
444 let msg_data = data.split_to(bgp_msg_length);
445
446 match msg_type {
447 BgpMessageType::UPDATE => Ok(RouteRecordIter::Update(parse_bgp_update_routes(
448 msg_data, add_path, asn_len, timestamp, peer_ip, peer_asn,
449 )?)),
450 BgpMessageType::OPEN
451 | BgpMessageType::NOTIFICATION
452 | BgpMessageType::KEEPALIVE
453 | BgpMessageType::ROUTE_REFRESH => Ok(RouteRecordIter::Empty),
454 }
455}
456
457fn bgp4mp_asn_len_and_add_path(msg_type: Bgp4MpType) -> Option<(AsnLength, bool)> {
458 match msg_type {
459 Bgp4MpType::Message | Bgp4MpType::MessageLocal => Some((AsnLength::Bits16, false)),
460 Bgp4MpType::MessageAs4 | Bgp4MpType::MessageAs4Local => Some((AsnLength::Bits32, false)),
461 Bgp4MpType::MessageAddpath | Bgp4MpType::MessageLocalAddpath => {
462 Some((AsnLength::Bits16, true))
463 }
464 Bgp4MpType::MessageAs4Addpath | Bgp4MpType::MessageLocalAs4Addpath => {
465 Some((AsnLength::Bits32, true))
466 }
467 Bgp4MpType::StateChange | Bgp4MpType::StateChangeAs4 => None,
468 }
469}
470
471fn parse_bgp4mp_routes(
472 sub_type: u16,
473 mut data: Bytes,
474 timestamp: f64,
475) -> Result<RouteRecordIter, ParserError> {
476 let msg_type = Bgp4MpType::try_from(sub_type)?;
477 let Some((asn_len, add_path)) = bgp4mp_asn_len_and_add_path(msg_type) else {
478 return Ok(RouteRecordIter::Empty);
479 };
480
481 let total_size = data.len();
482 let is_short_zebra_open = is_short_zebra_open(&data, &asn_len);
483 let peer_asn = data.read_asn(asn_len)?;
484 let _local_asn = data.read_asn(asn_len)?;
485 if is_short_zebra_open {
486 return parse_bgp_message_routes(
487 data,
488 add_path,
489 &asn_len,
490 timestamp,
491 IpAddr::V4(Ipv4Addr::UNSPECIFIED),
492 peer_asn,
493 );
494 }
495 let _interface_index = data.read_u16()?;
496 let afi = data.read_afi()?;
497 let should_read = bgp4mp_message_payload_len(&afi, &asn_len, total_size)?;
498 let peer_ip = data.read_address(&afi)?;
499 let _local_ip = data.read_address(&afi)?;
500
501 if should_read != data.remaining() {
502 return Err(ParserError::TruncatedMsg(format!(
503 "truncated bgp4mp message: should read {} bytes, have {} bytes available",
504 should_read,
505 data.remaining()
506 )));
507 }
508
509 parse_bgp_message_routes(data, add_path, &asn_len, timestamp, peer_ip, peer_asn)
510}
511
512fn table_dump_v2_afi_safi(rib_type: TableDumpV2Type) -> Result<(Afi, Safi), ParserError> {
513 match rib_type {
514 TableDumpV2Type::RibIpv4Unicast | TableDumpV2Type::RibIpv4UnicastAddPath => {
515 Ok((Afi::Ipv4, Safi::Unicast))
516 }
517 TableDumpV2Type::RibIpv4Multicast | TableDumpV2Type::RibIpv4MulticastAddPath => {
518 Ok((Afi::Ipv4, Safi::Multicast))
519 }
520 TableDumpV2Type::RibIpv6Unicast | TableDumpV2Type::RibIpv6UnicastAddPath => {
521 Ok((Afi::Ipv6, Safi::Unicast))
522 }
523 TableDumpV2Type::RibIpv6Multicast | TableDumpV2Type::RibIpv6MulticastAddPath => {
524 Ok((Afi::Ipv6, Safi::Multicast))
525 }
526 _ => Err(ParserError::ParseError(format!(
527 "wrong RIB type for parsing: {rib_type:?}"
528 ))),
529 }
530}
531
532fn is_add_path_rib_type(rib_type: TableDumpV2Type) -> bool {
533 matches!(
534 rib_type,
535 TableDumpV2Type::RibIpv4UnicastAddPath
536 | TableDumpV2Type::RibIpv4MulticastAddPath
537 | TableDumpV2Type::RibIpv6UnicastAddPath
538 | TableDumpV2Type::RibIpv6MulticastAddPath
539 )
540}
541
542fn parse_table_dump_routes(sub_type: u16, mut data: Bytes) -> Result<RouteRecordIter, ParserError> {
543 let afi = match sub_type {
544 1 => Afi::Ipv4,
545 2 => Afi::Ipv6,
546 _ => {
547 return Err(ParserError::ParseError(format!(
548 "Invalid subtype found for TABLE_DUMP (V1) message: {sub_type}"
549 )))
550 }
551 };
552
553 let _view_number = data.read_u16()?;
554 let _sequence_number = data.read_u16()?;
555 if data.is_empty() {
556 return Err(ParserError::TruncatedMsg(
557 "TABLE_DUMP record contains no entries".to_string(),
558 ));
559 }
560 Ok(RouteRecordIter::TableDump(RouteTableDumpIter { data, afi }))
561}
562
563fn parse_legacy_bgp_routes(
564 sub_type: u16,
565 mut data: Bytes,
566 timestamp: f64,
567) -> Result<RouteRecordIter, ParserError> {
568 match sub_type {
569 BGP_UPDATE => {
570 let peer_asn = Asn::new_16bit(data.read_u16()?);
571 let peer_ip = IpAddr::V4(data.read_ipv4_address()?);
572 let _local_asn = data.read_u16()?;
573 let _local_ip = data.read_ipv4_address()?;
574 Ok(RouteRecordIter::Update(parse_bgp_update_routes(
575 data,
576 false,
577 &AsnLength::Bits16,
578 timestamp,
579 peer_ip,
580 peer_asn,
581 )?))
582 }
583 BGP_STATE_CHANGE | BGP_OPEN | BGP_NOTIFY | BGP_KEEPALIVE => {
584 parse_legacy_bgp(sub_type, data)?;
585 Ok(RouteRecordIter::Empty)
586 }
587 _ => Err(ParserError::Unsupported(format!(
588 "unsupported legacy BGP subtype: {sub_type}"
589 ))),
590 }
591}
592
593fn parse_table_dump_v2_routes(
594 sub_type: u16,
595 mut data: Bytes,
596 peer_table: &mut Option<RoutePeerTable>,
597) -> Result<RouteRecordIter, ParserError> {
598 let v2_type = TableDumpV2Type::try_from(sub_type)?;
599 match v2_type {
600 TableDumpV2Type::PeerIndexTable => {
601 *peer_table = Some(parse_route_peer_table(data)?);
602 Ok(RouteRecordIter::Empty)
603 }
604 TableDumpV2Type::GeoPeerTable => Ok(RouteRecordIter::Empty),
605 TableDumpV2Type::RibGeneric | TableDumpV2Type::RibGenericAddPath => Err(
606 ParserError::Unsupported("TableDumpV2 RibGeneric is not currently supported".into()),
607 ),
608 rib_type => {
609 let (afi, safi) = table_dump_v2_afi_safi(rib_type)?;
610 let is_add_path = is_add_path_rib_type(rib_type);
611 let _sequence_number = data.read_u32()?;
612 let prefix = data.read_nlri_prefix(&afi, false)?;
613 let entry_count = data.read_u16()?;
614 let Some(peer_table) = peer_table.clone() else {
615 return Err(ParserError::ParseError(
616 "peer table not set for TableDumpV2 RIB entries".to_string(),
617 ));
618 };
619
620 Ok(RouteRecordIter::RibAfi(RouteRibAfiIter {
621 data,
622 peer_table,
623 afi,
624 safi,
625 is_add_path,
626 prefix,
627 remaining_entries: entry_count,
628 }))
629 }
630 }
631}
632
633fn parse_raw_record_route_iter(
634 raw_record: crate::RawMrtRecord,
635 peer_table: &mut Option<RoutePeerTable>,
636) -> Result<RouteRecordIter, ParserError> {
637 let timestamp = record_timestamp(&raw_record.common_header);
638 match raw_record.common_header.entry_type {
639 EntryType::TABLE_DUMP => parse_table_dump_routes(
640 raw_record.common_header.entry_subtype,
641 raw_record.message_bytes,
642 ),
643 EntryType::TABLE_DUMP_V2 => parse_table_dump_v2_routes(
644 raw_record.common_header.entry_subtype,
645 raw_record.message_bytes,
646 peer_table,
647 ),
648 EntryType::BGP4MP | EntryType::BGP4MP_ET => parse_bgp4mp_routes(
649 raw_record.common_header.entry_subtype,
650 raw_record.message_bytes,
651 timestamp,
652 ),
653 EntryType::BGP => parse_legacy_bgp_routes(
654 raw_record.common_header.entry_subtype,
655 raw_record.message_bytes,
656 timestamp,
657 ),
658 v => Err(ParserError::Unsupported(format!(
659 "unsupported MRT type: {v:?}"
660 ))),
661 }
662}
663
664pub struct RouteIterator<R> {
665 parser: BgpkitParser<R>,
666 pending_routes: RouteRecordIter,
667 peer_table: Option<RoutePeerTable>,
668 pending_raw_bytes: Option<Vec<u8>>,
669}
670
671impl<R> RouteIterator<R> {
672 pub(crate) fn new(parser: BgpkitParser<R>) -> Self {
673 Self {
674 parser,
675 pending_routes: RouteRecordIter::Empty,
676 peer_table: None,
677 pending_raw_bytes: None,
678 }
679 }
680}
681
682fn log_route_body_error<R>(parser: &mut BgpkitParser<R>, err: &ParserError) {
686 match err {
687 ParserError::ParseError(err_str) => parser.log_parse_error_once(err_str),
688 other => error!("parser error: {}", other),
689 }
690}
691
692impl<R: Read> Iterator for RouteIterator<R> {
693 type Item = BgpRouteElem;
694
695 fn next(&mut self) -> Option<Self::Item> {
696 if self.parser.text_dump_iter.is_some() {
698 return None;
699 }
700 loop {
701 match self.pending_routes.next_route() {
702 Ok(Some(route)) => {
703 if route.match_filters(&self.parser.filters) {
704 return Some(route);
705 }
706 continue;
707 }
708 Ok(None) => {
709 self.pending_raw_bytes = None;
710 }
711 Err(err) => {
712 log_route_body_error(&mut self.parser, &err);
713 self.pending_routes = RouteRecordIter::Empty;
714 write_mrt_core_dump(self.parser.core_dump, self.pending_raw_bytes.take());
715 if self.parser.core_dump {
716 return None;
717 }
718 continue;
719 }
720 }
721
722 let raw_record = match chunk_mrt_record(&mut self.parser.reader) {
723 Ok(raw_record) => raw_record,
724 Err(e) => {
725 if !handle_record_parse_error(&mut self.parser, e.error, e.bytes) {
726 return None;
727 }
728 continue;
729 }
730 };
731
732 let used_zebra_compat = raw_record_uses_zebra_compat(&raw_record);
733 let raw_bytes = raw_record.raw_bytes().to_vec();
734 match parse_raw_record_route_iter(raw_record, &mut self.peer_table) {
735 Ok(routes) => {
736 if used_zebra_compat {
737 self.parser.warn_zebra_compat_once();
738 }
739 self.pending_routes = routes;
740 self.pending_raw_bytes = Some(raw_bytes);
741 }
742 Err(err) => {
743 log_route_body_error(&mut self.parser, &err);
744 write_mrt_core_dump(self.parser.core_dump, Some(raw_bytes));
745 if self.parser.core_dump {
746 return None;
747 }
748 continue;
749 }
750 }
751 }
752 }
753}
754
755pub struct FallibleRouteIterator<R> {
756 parser: BgpkitParser<R>,
757 pending_routes: RouteRecordIter,
758 peer_table: Option<RoutePeerTable>,
759 pending_raw_bytes: Option<Vec<u8>>,
760 finished: bool,
761}
762
763impl<R> FallibleRouteIterator<R> {
764 pub(crate) fn new(parser: BgpkitParser<R>) -> Self {
765 Self {
766 parser,
767 pending_routes: RouteRecordIter::Empty,
768 peer_table: None,
769 pending_raw_bytes: None,
770 finished: false,
771 }
772 }
773}
774
775impl<R: Read> Iterator for FallibleRouteIterator<R> {
776 type Item = Result<BgpRouteElem, ParserErrorWithBytes>;
777
778 fn next(&mut self) -> Option<Self::Item> {
779 if self.parser.text_dump_iter.is_some() {
781 return None;
782 }
783 loop {
784 match self.pending_routes.next_route() {
785 Ok(Some(route)) => {
786 if route.match_filters(&self.parser.filters) {
787 return Some(Ok(route));
788 }
789 continue;
790 }
791 Ok(None) => {
792 self.pending_raw_bytes = None;
793 }
794 Err(error) => {
795 self.pending_routes = RouteRecordIter::Empty;
796 return Some(Err(ParserErrorWithBytes {
797 error,
798 bytes: self.pending_raw_bytes.take(),
799 }));
800 }
801 }
802
803 let raw_record =
804 match super::next_fallible_raw_record(&mut self.parser.reader, &mut self.finished)?
805 {
806 Ok(raw_record) => raw_record,
807 Err(e) => return Some(Err(e)),
808 };
809
810 let used_zebra_compat = raw_record_uses_zebra_compat(&raw_record);
811 let raw_bytes = raw_record.raw_bytes().to_vec();
812 match parse_raw_record_route_iter(raw_record, &mut self.peer_table) {
813 Ok(routes) => {
814 if used_zebra_compat {
815 self.parser.warn_zebra_compat_once();
816 }
817 self.pending_routes = routes;
818 self.pending_raw_bytes = Some(raw_bytes);
819 }
820 Err(error) => {
821 return Some(Err(ParserErrorWithBytes {
822 error,
823 bytes: Some(raw_bytes),
824 }))
825 }
826 }
827 }
828 }
829}
830
831#[cfg(test)]
832mod tests {
833 use super::*;
834 use crate::parser::iters::write_mrt_core_dump_to_path;
835 use bytes::{BufMut, BytesMut};
836 use std::io::Cursor;
837 use std::net::{Ipv4Addr, Ipv6Addr};
838 use std::str::FromStr;
839
840 fn route_projection(elem: BgpElem) -> BgpRouteElem {
841 BgpRouteElem {
842 timestamp: elem.timestamp,
843 elem_type: elem.elem_type,
844 peer_ip: elem.peer_ip,
845 peer_asn: elem.peer_asn,
846 prefix: elem.prefix,
847 as_path: elem.as_path.map(Arc::new),
848 }
849 }
850
851 fn collect_route_record_iter(
852 mut iter: RouteRecordIter,
853 ) -> Result<Vec<BgpRouteElem>, ParserError> {
854 let mut routes = Vec::new();
855 while let Some(route) = iter.next_route()? {
856 routes.push(route);
857 }
858 Ok(routes)
859 }
860
861 fn route_peer_table_from_peer_index(peer_table: PeerIndexTable) -> RoutePeerTable {
862 let mut peer_ids = peer_table.id_peer_map.keys().copied().collect::<Vec<_>>();
863 peer_ids.sort_unstable();
864 let peers = peer_ids
865 .into_iter()
866 .map(|peer_id| peer_table.id_peer_map[&peer_id])
867 .collect::<Vec<_>>();
868
869 RoutePeerTable {
870 peers: Arc::from(peers),
871 }
872 }
873
874 fn update_record() -> MrtRecord {
875 let mut attributes = Attributes::default();
876 attributes.add_attr(AttributeValue::Origin(Origin::IGP).into());
877 attributes.add_attr(AttributeValue::AsPath(AsPath::from_sequence([64500, 64501])).into());
878 attributes
879 .add_attr(AttributeValue::NextHop(IpAddr::from_str("192.0.2.254").unwrap()).into());
880
881 MrtRecord {
882 common_header: CommonHeader {
883 timestamp: 1_700_000_000,
884 microsecond_timestamp: None,
885 entry_type: EntryType::BGP4MP,
886 entry_subtype: Bgp4MpType::MessageAs4 as u16,
887 length: 0,
888 },
889 message: MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(Bgp4MpMessage {
890 msg_type: Bgp4MpType::MessageAs4,
891 peer_asn: Asn::new_32bit(64496),
892 local_asn: Asn::new_32bit(64497),
893 interface_index: 0,
894 peer_ip: IpAddr::from_str("192.0.2.1").unwrap(),
895 local_ip: IpAddr::from_str("192.0.2.2").unwrap(),
896 bgp_message: BgpMessage::Update(BgpUpdateMessage {
897 withdrawn_prefixes: vec![NetworkPrefix::from_str("198.51.100.0/24").unwrap()],
898 attributes,
899 announced_prefixes: vec![NetworkPrefix::from_str("203.0.113.0/24").unwrap()],
900 }),
901 })),
902 }
903 }
904
905 fn route_attributes(as_path: impl AsRef<[u32]>) -> Attributes {
906 let mut attributes = Attributes::default();
907 attributes.add_attr(AttributeValue::Origin(Origin::IGP).into());
908 attributes.add_attr(AttributeValue::AsPath(AsPath::from_sequence(as_path)).into());
909 attributes
910 .add_attr(AttributeValue::NextHop(IpAddr::from_str("192.0.2.254").unwrap()).into());
911 attributes
912 }
913
914 fn bgp4mp_record(msg_type: Bgp4MpType, bgp_message: BgpMessage) -> MrtRecord {
915 let asn = if matches!(
916 msg_type,
917 Bgp4MpType::Message
918 | Bgp4MpType::MessageLocal
919 | Bgp4MpType::MessageAddpath
920 | Bgp4MpType::MessageLocalAddpath
921 ) {
922 Asn::new_16bit(64496)
923 } else {
924 Asn::new_32bit(64496)
925 };
926
927 MrtRecord {
928 common_header: CommonHeader {
929 timestamp: 1_700_000_000,
930 microsecond_timestamp: None,
931 entry_type: EntryType::BGP4MP,
932 entry_subtype: msg_type as u16,
933 length: 0,
934 },
935 message: MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(Bgp4MpMessage {
936 msg_type,
937 peer_asn: asn,
938 local_asn: Asn::new_32bit(64497),
939 interface_index: 0,
940 peer_ip: IpAddr::from_str("192.0.2.1").unwrap(),
941 local_ip: IpAddr::from_str("192.0.2.2").unwrap(),
942 bgp_message,
943 })),
944 }
945 }
946
947 fn open_message() -> BgpMessage {
948 BgpMessage::Open(BgpOpenMessage {
949 version: 4,
950 asn: Asn::new_16bit(64496),
951 hold_time: 180,
952 bgp_identifier: Ipv4Addr::new(192, 0, 2, 1),
953 extended_length: false,
954 opt_params: vec![],
955 })
956 }
957
958 fn raw_bgp_message(length: u16, msg_type: BgpMessageType, payload: &[u8]) -> Bytes {
959 raw_bgp_message_with_marker([0xff; 16], length, msg_type, payload)
960 }
961
962 fn raw_bgp_message_with_marker(
963 marker: [u8; 16],
964 length: u16,
965 msg_type: BgpMessageType,
966 payload: &[u8],
967 ) -> Bytes {
968 let mut bytes = BytesMut::new();
969 bytes.put_slice(&marker);
970 bytes.put_u16(length);
971 bytes.put_u8(msg_type as u8);
972 bytes.put_slice(payload);
973 bytes.freeze()
974 }
975
976 fn table_dump_record() -> MrtRecord {
977 let mut attributes = Attributes::default();
978 attributes.add_attr(AttributeValue::Origin(Origin::IGP).into());
979 attributes.add_attr(AttributeValue::AsPath(AsPath::from_sequence([64500, 64501])).into());
980 attributes
981 .add_attr(AttributeValue::NextHop(IpAddr::from_str("192.0.2.254").unwrap()).into());
982
983 MrtRecord {
984 common_header: CommonHeader {
985 timestamp: 1_700_000_000,
986 microsecond_timestamp: None,
987 entry_type: EntryType::TABLE_DUMP,
988 entry_subtype: 1,
989 length: 0,
990 },
991 message: MrtMessage::TableDumpMessage(TableDumpMessage {
992 view_number: 0,
993 sequence_number: 1,
994 prefix: NetworkPrefix::from_str("203.0.113.0/24").unwrap(),
995 status: 1,
996 originated_time: 1_699_999_998,
997 peer_ip: IpAddr::from_str("192.0.2.20").unwrap(),
998 peer_asn: Asn::new_16bit(64496),
999 attributes,
1000 }),
1001 }
1002 }
1003
1004 fn table_dump_ipv6_record() -> MrtRecord {
1005 let mut attributes = Attributes::default();
1006 attributes.add_attr(AttributeValue::Origin(Origin::IGP).into());
1007 attributes.add_attr(AttributeValue::AsPath(AsPath::from_sequence([64500, 64501])).into());
1008
1009 MrtRecord {
1010 common_header: CommonHeader {
1011 timestamp: 1_700_000_000,
1012 microsecond_timestamp: None,
1013 entry_type: EntryType::TABLE_DUMP,
1014 entry_subtype: 2,
1015 length: 0,
1016 },
1017 message: MrtMessage::TableDumpMessage(TableDumpMessage {
1018 view_number: 0,
1019 sequence_number: 1,
1020 prefix: NetworkPrefix::from_str("2001:db8::/32").unwrap(),
1021 status: 1,
1022 originated_time: 1_699_999_998,
1023 peer_ip: IpAddr::from_str("2001:db8::20").unwrap(),
1024 peer_asn: Asn::new_16bit(64496),
1025 attributes,
1026 }),
1027 }
1028 }
1029
1030 fn table_dump_v2_records_bytes() -> Vec<u8> {
1031 let peer = Peer::new(
1032 "192.0.2.10".parse().unwrap(),
1033 "192.0.2.11".parse().unwrap(),
1034 Asn::new_32bit(64496),
1035 );
1036 let mut peer_table = PeerIndexTable::default();
1037 let peer_index = peer_table.add_peer(peer).unwrap();
1038
1039 let mut attributes = Attributes::default();
1040 attributes.add_attr(AttributeValue::Origin(Origin::IGP).into());
1041 attributes.add_attr(AttributeValue::AsPath(AsPath::from_sequence([64500, 64501])).into());
1042 attributes
1043 .add_attr(AttributeValue::NextHop(IpAddr::from_str("192.0.2.254").unwrap()).into());
1044
1045 let pit_record = MrtRecord {
1046 common_header: CommonHeader {
1047 timestamp: 1_700_000_000,
1048 microsecond_timestamp: None,
1049 entry_type: EntryType::TABLE_DUMP_V2,
1050 entry_subtype: TableDumpV2Type::PeerIndexTable as u16,
1051 length: 0,
1052 },
1053 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(peer_table)),
1054 };
1055 let rib_record = MrtRecord {
1056 common_header: CommonHeader {
1057 timestamp: 1_700_000_001,
1058 microsecond_timestamp: None,
1059 entry_type: EntryType::TABLE_DUMP_V2,
1060 entry_subtype: TableDumpV2Type::RibIpv4Unicast as u16,
1061 length: 0,
1062 },
1063 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::RibAfi(RibAfiEntries {
1064 rib_type: TableDumpV2Type::RibIpv4Unicast,
1065 sequence_number: 1,
1066 prefix: NetworkPrefix::from_str("203.0.113.0/24").unwrap(),
1067 rib_entries: vec![RibEntry {
1068 peer_index,
1069 originated_time: 1_699_999_999,
1070 path_id: None,
1071 attributes,
1072 }],
1073 })),
1074 };
1075
1076 let mut bytes = pit_record.encode().unwrap().to_vec();
1077 bytes.extend_from_slice(&rib_record.encode().unwrap());
1078 bytes
1079 }
1080
1081 fn table_dump_v2_truncated_attribute_payload() -> (Vec<u8>, Bytes, PeerIndexTable) {
1082 let peer = Peer::new(
1083 "192.0.2.10".parse().unwrap(),
1084 "192.0.2.11".parse().unwrap(),
1085 Asn::new_32bit(64496),
1086 );
1087 let mut peer_table = PeerIndexTable::default();
1088 let peer_index = peer_table.add_peer(peer).unwrap();
1089
1090 let pit_record = MrtRecord {
1091 common_header: CommonHeader {
1092 timestamp: 1_700_000_000,
1093 microsecond_timestamp: None,
1094 entry_type: EntryType::TABLE_DUMP_V2,
1095 entry_subtype: TableDumpV2Type::PeerIndexTable as u16,
1096 length: 0,
1097 },
1098 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(
1099 peer_table.clone(),
1100 )),
1101 };
1102
1103 let first_entry = RibEntry {
1104 peer_index,
1105 originated_time: 1_699_999_999,
1106 path_id: None,
1107 attributes: route_attributes([64500, 64501]),
1108 };
1109
1110 let mut rib_body = BytesMut::new();
1111 rib_body.put_u32(1);
1112 rib_body.extend(NetworkPrefix::from_str("203.0.113.0/24").unwrap().encode());
1113 rib_body.put_u16(2);
1114 rib_body.extend(first_entry.encode().unwrap());
1115 rib_body.put_u16(peer_index);
1116 rib_body.put_u32(1_699_999_998);
1117 rib_body.put_u16(32);
1118 rib_body.put_u8(0);
1119
1120 let rib_body = rib_body.freeze();
1121 let rib_header = CommonHeader {
1122 timestamp: 1_700_000_001,
1123 microsecond_timestamp: None,
1124 entry_type: EntryType::TABLE_DUMP_V2,
1125 entry_subtype: TableDumpV2Type::RibIpv4Unicast as u16,
1126 length: rib_body.len() as u32,
1127 };
1128
1129 let mut bytes = pit_record.encode().unwrap().to_vec();
1130 bytes.extend_from_slice(&rib_header.encode());
1131 bytes.extend_from_slice(&rib_body);
1132
1133 (bytes, rib_body, peer_table)
1134 }
1135
1136 fn assert_filtered_route_projection(bytes: Vec<u8>, filters: &[(&str, &str)]) {
1137 let elem_parser = filters.iter().fold(
1138 BgpkitParser::from_reader(Cursor::new(bytes.clone())),
1139 |parser, (filter_type, filter_value)| {
1140 parser.add_filter(filter_type, filter_value).unwrap()
1141 },
1142 );
1143 let route_parser = filters.iter().fold(
1144 BgpkitParser::from_reader(Cursor::new(bytes)),
1145 |parser, (filter_type, filter_value)| {
1146 parser.add_filter(filter_type, filter_value).unwrap()
1147 },
1148 );
1149
1150 let elem_projection = elem_parser
1151 .into_elem_iter()
1152 .map(route_projection)
1153 .collect::<Vec<_>>();
1154 let routes = route_parser.into_route_iter().collect::<Vec<_>>();
1155
1156 assert_eq!(routes, elem_projection, "filters: {filters:?}");
1157 }
1158
1159 fn assert_route_projection(bytes: Vec<u8>) -> Vec<BgpRouteElem> {
1160 let elem_projection = BgpkitParser::from_reader(Cursor::new(bytes.clone()))
1161 .into_elem_iter()
1162 .map(route_projection)
1163 .collect::<Vec<_>>();
1164 let routes = BgpkitParser::from_reader(Cursor::new(bytes))
1165 .into_route_iter()
1166 .collect::<Vec<_>>();
1167
1168 assert_eq!(routes, elem_projection);
1169 routes
1170 }
1171
1172 #[test]
1173 fn bgp4mp_routes_rejects_link_state_envelope_afi() {
1174 let mut data = BytesMut::new();
1175 data.put_u16(65000);
1176 data.put_u16(65001);
1177 data.put_u16(0);
1178 data.put_u16(Afi::LinkState as u16);
1179 data.put_slice(&BgpMessage::KeepAlive.encode(AsnLength::Bits16).unwrap());
1180
1181 let error =
1182 match parse_bgp4mp_routes(Bgp4MpType::Message as u16, data.freeze(), 1_700_000_000.0) {
1183 Err(error) => error,
1184 Ok(_) => panic!("unexpectedly parsed BGP4MP routes"),
1185 };
1186 assert!(matches!(
1187 error,
1188 ParserError::ParseError(message)
1189 if message == "Link-State AFI is invalid in a BGP4MP envelope"
1190 ));
1191 }
1192
1193 #[test]
1194 fn route_iterator_matches_elem_projection_for_update() {
1195 let bytes = update_record().encode().unwrap().to_vec();
1196 let routes = assert_route_projection(bytes);
1197 assert_eq!(routes.len(), 2);
1198 assert_eq!(routes[0].elem_type, ElemType::ANNOUNCE);
1199 assert_eq!(routes[1].elem_type, ElemType::WITHDRAW);
1200 assert!(routes[1].as_path.is_none());
1201 }
1202
1203 #[test]
1204 fn route_iterator_shares_as_path_for_update_announcements() {
1205 let bytes = bgp4mp_record(
1206 Bgp4MpType::MessageAs4,
1207 BgpMessage::Update(BgpUpdateMessage {
1208 withdrawn_prefixes: vec![],
1209 attributes: route_attributes([64500, 64501]),
1210 announced_prefixes: vec![
1211 NetworkPrefix::from_str("203.0.113.0/24").unwrap(),
1212 NetworkPrefix::from_str("198.51.100.0/24").unwrap(),
1213 ],
1214 }),
1215 )
1216 .encode()
1217 .unwrap()
1218 .to_vec();
1219
1220 let routes = BgpkitParser::from_reader(Cursor::new(bytes))
1221 .into_route_iter()
1222 .collect::<Vec<_>>();
1223
1224 assert_eq!(routes.len(), 2);
1225 assert!(Arc::ptr_eq(
1226 routes[0].as_path.as_ref().unwrap(),
1227 routes[1].as_path.as_ref().unwrap()
1228 ));
1229 }
1230
1231 #[test]
1232 fn route_iterator_uses_microsecond_timestamps() {
1233 let timestamp = record_timestamp(&CommonHeader {
1234 timestamp: 1_700_000_000,
1235 microsecond_timestamp: Some(123_456),
1236 entry_type: EntryType::BGP4MP_ET,
1237 entry_subtype: Bgp4MpType::MessageAs4 as u16,
1238 length: 0,
1239 });
1240
1241 assert_eq!(timestamp, 1_700_000_000.123_456);
1242 }
1243
1244 #[test]
1245 fn route_iterator_matches_elem_projection_for_mp_update() {
1246 let mut attributes = route_attributes([64500, 64501]);
1247 attributes.add_attr(
1248 AttributeValue::MpReachNlri(Nlri::new_reachable(
1249 NetworkPrefix::from_str("2001:db8::/32").unwrap(),
1250 Some(IpAddr::from_str("2001:db8::1").unwrap()),
1251 ))
1252 .into(),
1253 );
1254 attributes.add_attr(
1255 AttributeValue::MpUnreachNlri(Nlri::new_unreachable(
1256 NetworkPrefix::from_str("2001:db8:1::/48").unwrap(),
1257 ))
1258 .into(),
1259 );
1260
1261 let bytes = bgp4mp_record(
1262 Bgp4MpType::MessageAs4,
1263 BgpMessage::Update(BgpUpdateMessage {
1264 withdrawn_prefixes: vec![],
1265 attributes,
1266 announced_prefixes: vec![],
1267 }),
1268 )
1269 .encode()
1270 .unwrap()
1271 .to_vec();
1272
1273 let routes = assert_route_projection(bytes);
1274 assert_eq!(routes.len(), 2);
1275 assert_eq!(routes[0].elem_type, ElemType::ANNOUNCE);
1276 assert_eq!(
1277 routes[0].prefix,
1278 NetworkPrefix::from_str("2001:db8::/32").unwrap()
1279 );
1280 assert_eq!(routes[1].elem_type, ElemType::WITHDRAW);
1281 assert_eq!(
1282 routes[1].prefix,
1283 NetworkPrefix::from_str("2001:db8:1::/48").unwrap()
1284 );
1285 }
1286
1287 #[test]
1288 fn route_iterator_matches_elem_projection_for_non_update_bgp4mp_messages() {
1289 let records = [
1290 bgp4mp_record(Bgp4MpType::Message, open_message()),
1291 bgp4mp_record(
1292 Bgp4MpType::MessageAs4,
1293 BgpMessage::Notification(BgpNotificationMessage {
1294 error: BgpError::Unknown(1, 0),
1295 data: vec![],
1296 }),
1297 ),
1298 bgp4mp_record(Bgp4MpType::MessageAddpath, BgpMessage::KeepAlive),
1299 bgp4mp_record(Bgp4MpType::MessageAs4Addpath, BgpMessage::KeepAlive),
1300 ];
1301 let mut bytes = Vec::new();
1302 for record in records {
1303 bytes.extend_from_slice(&record.encode().unwrap());
1304 }
1305
1306 assert!(assert_route_projection(bytes).is_empty());
1307 }
1308
1309 #[test]
1310 fn route_iterator_matches_elem_projection_for_bgp4mp_16bit_update() {
1311 let bytes = bgp4mp_record(
1312 Bgp4MpType::Message,
1313 BgpMessage::Update(BgpUpdateMessage {
1314 withdrawn_prefixes: vec![],
1315 attributes: route_attributes([64500, 64501]),
1316 announced_prefixes: vec![NetworkPrefix::from_str("203.0.113.0/24").unwrap()],
1317 }),
1318 )
1319 .encode()
1320 .unwrap()
1321 .to_vec();
1322
1323 let routes = assert_route_projection(bytes);
1324 assert_eq!(routes.len(), 1);
1325 assert_eq!(routes[0].peer_asn, Asn::new_16bit(64496));
1326 }
1327
1328 #[test]
1329 fn route_iterator_filters_match_elem_projection_for_update() {
1330 let bytes = update_record().encode().unwrap().to_vec();
1331 let cases: &[&[(&str, &str)]] = &[
1332 &[("peer_ip", "192.0.2.1")],
1333 &[("peer_ip", "192.0.2.99")],
1334 &[("peer_asn", "64496")],
1335 &[("type", "a")],
1336 &[("type", "w")],
1337 &[("type", "!w")],
1338 &[("prefix", "203.0.113.0/24")],
1339 &[("prefix", "198.51.100.0/24")],
1340 &[("prefix_super", "203.0.113.128/25")],
1341 &[("origin_asn", "64501")],
1342 &[("origin_asns", "64496,64501")],
1343 &[("as_path", "64500 64501$")],
1344 &[("ip_version", "4")],
1345 &[("ts_start", "1700000000"), ("ts_end", "1700000000")],
1346 &[("peer_ip", "192.0.2.1"), ("type", "a")],
1347 ];
1348
1349 for filters in cases {
1350 assert_filtered_route_projection(bytes.clone(), filters);
1351 }
1352 }
1353
1354 #[test]
1355 fn selective_attribute_parser_merges_as4_path() {
1356 let mut attributes = Attributes::default();
1357 attributes.add_attr(AttributeValue::AsPath(AsPath::from_sequence([23456, 64497])).into());
1358 attributes.add_attr(AttributeValue::As4Path(AsPath::from_sequence([65536, 64497])).into());
1359
1360 let attrs = parse_route_attributes(
1361 attributes.encode(AsnLength::Bits16).unwrap(),
1362 &AsnLength::Bits16,
1363 false,
1364 RouteAttributeContext {
1365 afi: None,
1366 safi: None,
1367 prefixes: None,
1368 is_announcement: Some(true),
1369 has_standard_nlri: true,
1370 },
1371 )
1372 .unwrap();
1373
1374 assert_eq!(
1375 attrs.as_path.unwrap().to_u32_vec_opt(false).unwrap(),
1376 vec![65536, 64497]
1377 );
1378 }
1379
1380 #[test]
1381 fn selective_attribute_parser_handles_as_path_without_as4_path() {
1382 let attrs = parse_route_attributes(
1383 route_attributes([64500, 64501])
1384 .encode(AsnLength::Bits16)
1385 .unwrap(),
1386 &AsnLength::Bits16,
1387 false,
1388 RouteAttributeContext {
1389 afi: None,
1390 safi: None,
1391 prefixes: None,
1392 is_announcement: Some(true),
1393 has_standard_nlri: true,
1394 },
1395 )
1396 .unwrap();
1397
1398 assert_eq!(
1399 attrs.as_path.unwrap().to_u32_vec_opt(false).unwrap(),
1400 vec![64500, 64501]
1401 );
1402 }
1403
1404 #[test]
1405 fn selective_attribute_parser_handles_as4_path_without_as_path() {
1406 let mut attributes = Attributes::default();
1407 attributes.add_attr(AttributeValue::As4Path(AsPath::from_sequence([65536, 64497])).into());
1408
1409 let attrs = parse_route_attributes(
1410 attributes.encode(AsnLength::Bits16).unwrap(),
1411 &AsnLength::Bits16,
1412 false,
1413 RouteAttributeContext {
1414 afi: None,
1415 safi: None,
1416 prefixes: None,
1417 is_announcement: Some(false),
1418 has_standard_nlri: false,
1419 },
1420 )
1421 .unwrap();
1422
1423 assert_eq!(
1424 attrs.as_path.unwrap().to_u32_vec_opt(false).unwrap(),
1425 vec![65536, 64497]
1426 );
1427 }
1428
1429 #[test]
1430 fn selective_attribute_parser_handles_no_as_path() {
1431 let attrs = parse_route_attributes(
1432 Bytes::new(),
1433 &AsnLength::Bits16,
1434 false,
1435 RouteAttributeContext {
1436 afi: None,
1437 safi: None,
1438 prefixes: None,
1439 is_announcement: Some(false),
1440 has_standard_nlri: false,
1441 },
1442 )
1443 .unwrap();
1444
1445 assert!(attrs.as_path.is_none());
1446 }
1447
1448 #[test]
1449 fn selective_attribute_parser_handles_extended_and_truncated_attributes() {
1450 let mut extended_as_path = BytesMut::new();
1451 extended_as_path.put_u8((AttrFlags::TRANSITIVE | AttrFlags::EXTENDED).bits());
1452 extended_as_path.put_u8(u8::from(AttrType::AS_PATH));
1453 extended_as_path.put_u16(4);
1454 extended_as_path.put_u8(2);
1455 extended_as_path.put_u8(1);
1456 extended_as_path.put_u16(64500);
1457
1458 let attrs = parse_route_attributes(
1459 extended_as_path.freeze(),
1460 &AsnLength::Bits16,
1461 false,
1462 RouteAttributeContext {
1463 afi: None,
1464 safi: None,
1465 prefixes: None,
1466 is_announcement: Some(false),
1467 has_standard_nlri: false,
1468 },
1469 )
1470 .unwrap();
1471 assert_eq!(
1472 attrs.as_path.unwrap().to_u32_vec_opt(false).unwrap(),
1473 vec![64500]
1474 );
1475
1476 let attrs = parse_route_attributes(
1477 Bytes::from_static(&[0x40, 2, 5, 0]),
1478 &AsnLength::Bits16,
1479 false,
1480 RouteAttributeContext {
1481 afi: None,
1482 safi: None,
1483 prefixes: None,
1484 is_announcement: Some(false),
1485 has_standard_nlri: false,
1486 },
1487 )
1488 .unwrap();
1489 assert!(attrs.as_path.is_none());
1490 }
1491
1492 #[test]
1493 fn selective_attribute_parser_discards_malformed_as_path() {
1494 let attrs = parse_route_attributes(
1495 Bytes::from_static(&[0x40, 2, 1, 0]),
1496 &AsnLength::Bits16,
1497 false,
1498 RouteAttributeContext {
1499 afi: None,
1500 safi: None,
1501 prefixes: None,
1502 is_announcement: Some(false),
1503 has_standard_nlri: false,
1504 },
1505 )
1506 .unwrap();
1507
1508 assert!(attrs.as_path.is_none());
1509 }
1510
1511 #[test]
1512 fn route_iterator_matches_elem_projection_for_table_dump() {
1513 let bytes = table_dump_record().encode().unwrap().to_vec();
1514 let routes = assert_route_projection(bytes);
1515 assert_eq!(routes.len(), 1);
1516 assert_eq!(routes[0].timestamp, 1_699_999_998.0);
1517 assert_eq!(routes[0].peer_asn, Asn::new_16bit(64496));
1518 }
1519
1520 #[test]
1521 fn route_iterator_matches_elem_projection_for_table_dump_ipv6() {
1522 let bytes = table_dump_ipv6_record().encode().unwrap().to_vec();
1523 let routes = assert_route_projection(bytes);
1524 assert_eq!(routes.len(), 1);
1525 assert_eq!(
1526 routes[0].prefix,
1527 NetworkPrefix::from_str("2001:db8::/32").unwrap()
1528 );
1529 assert_eq!(
1530 routes[0].peer_ip,
1531 IpAddr::from(Ipv6Addr::from_str("2001:db8::20").unwrap())
1532 );
1533 }
1534
1535 #[test]
1536 fn route_iterator_matches_elem_projection_for_table_dump_v2() {
1537 let bytes = table_dump_v2_records_bytes();
1538 let routes = assert_route_projection(bytes);
1539 assert_eq!(routes.len(), 1);
1540 assert_eq!(routes[0].elem_type, ElemType::ANNOUNCE);
1541 assert_eq!(
1542 routes[0].as_path.as_ref().unwrap().to_u32_vec_opt(false),
1543 Some(vec![64500, 64501])
1544 );
1545 }
1546
1547 #[test]
1548 fn route_iterator_matches_elem_projection_for_table_dump_v2_ipv6_addpath() {
1549 let peer = Peer::new(
1550 "192.0.2.11".parse().unwrap(),
1551 "2001:db8::10".parse().unwrap(),
1552 Asn::new_32bit(64496),
1553 );
1554 let mut peer_table = PeerIndexTable::default();
1555 let peer_index = peer_table.add_peer(peer).unwrap();
1556
1557 let pit_record = MrtRecord {
1558 common_header: CommonHeader {
1559 timestamp: 1_700_000_000,
1560 microsecond_timestamp: None,
1561 entry_type: EntryType::TABLE_DUMP_V2,
1562 entry_subtype: TableDumpV2Type::PeerIndexTable as u16,
1563 length: 0,
1564 },
1565 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(peer_table)),
1566 };
1567 let rib_record = MrtRecord {
1568 common_header: CommonHeader {
1569 timestamp: 1_700_000_001,
1570 microsecond_timestamp: None,
1571 entry_type: EntryType::TABLE_DUMP_V2,
1572 entry_subtype: TableDumpV2Type::RibIpv6UnicastAddPath as u16,
1573 length: 0,
1574 },
1575 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::RibAfi(RibAfiEntries {
1576 rib_type: TableDumpV2Type::RibIpv6UnicastAddPath,
1577 sequence_number: 1,
1578 prefix: NetworkPrefix::from_str("2001:db8::/32").unwrap(),
1579 rib_entries: vec![RibEntry {
1580 peer_index,
1581 originated_time: 1_699_999_999,
1582 path_id: Some(1234),
1583 attributes: route_attributes([64500, 64501]),
1584 }],
1585 })),
1586 };
1587
1588 let mut bytes = pit_record.encode().unwrap().to_vec();
1589 bytes.extend_from_slice(&rib_record.encode().unwrap());
1590 let routes = assert_route_projection(bytes);
1591 assert_eq!(routes.len(), 1);
1592 assert_eq!(
1593 routes[0].prefix,
1594 NetworkPrefix::from_str("2001:db8::/32").unwrap()
1595 );
1596 }
1597
1598 #[test]
1599 fn route_iterator_matches_elem_projection_for_bgp4mp_ipv6_peer_update() {
1600 let record = MrtRecord {
1601 common_header: CommonHeader {
1602 timestamp: 1_700_000_000,
1603 microsecond_timestamp: None,
1604 entry_type: EntryType::BGP4MP,
1605 entry_subtype: Bgp4MpType::MessageAs4 as u16,
1606 length: 0,
1607 },
1608 message: MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(Bgp4MpMessage {
1609 msg_type: Bgp4MpType::MessageAs4,
1610 peer_asn: Asn::new_32bit(64496),
1611 local_asn: Asn::new_32bit(64497),
1612 interface_index: 0,
1613 peer_ip: IpAddr::from_str("2001:db8::1").unwrap(),
1614 local_ip: IpAddr::from_str("2001:db8::2").unwrap(),
1615 bgp_message: BgpMessage::Update(BgpUpdateMessage {
1616 withdrawn_prefixes: vec![],
1617 attributes: route_attributes([64500, 64501]),
1618 announced_prefixes: vec![NetworkPrefix::from_str("203.0.113.0/24").unwrap()],
1619 }),
1620 })),
1621 };
1622
1623 let routes = assert_route_projection(record.encode().unwrap().to_vec());
1624 assert_eq!(routes.len(), 1);
1625 assert_eq!(
1626 routes[0].peer_ip,
1627 IpAddr::from(Ipv6Addr::from_str("2001:db8::1").unwrap())
1628 );
1629 }
1630
1631 #[test]
1632 fn route_iterator_filters_match_elem_projection_for_table_dump_v2() {
1633 let bytes = table_dump_v2_records_bytes();
1634 let cases: &[&[(&str, &str)]] = &[
1635 &[("peer_ip", "192.0.2.10")],
1636 &[("peer_asn", "64496")],
1637 &[("type", "a")],
1638 &[("type", "w")],
1639 &[("prefix", "203.0.113.0/24")],
1640 &[("prefix_sub", "203.0.112.0/23")],
1641 &[("origin_asn", "64501")],
1642 &[("as_path", "64500 64501$")],
1643 &[("ts_start", "1699999999"), ("ts_end", "1699999999")],
1644 &[("peer_asn", "64496"), ("origin_asn", "64501")],
1645 ];
1646
1647 for filters in cases {
1648 assert_filtered_route_projection(bytes.clone(), filters);
1649 }
1650 }
1651
1652 #[test]
1653 fn route_parser_reports_bgp_message_shape_errors() {
1654 assert!(parse_bgp_message_routes(
1655 raw_bgp_message(18, BgpMessageType::KEEPALIVE, &[]),
1656 false,
1657 &AsnLength::Bits16,
1658 1_700_000_000.0,
1659 "192.0.2.1".parse().unwrap(),
1660 Asn::new_16bit(64496)
1661 )
1662 .is_err());
1663 assert!(parse_bgp_message_routes(
1664 raw_bgp_message(4097, BgpMessageType::OPEN, &[]),
1665 false,
1666 &AsnLength::Bits16,
1667 1_700_000_000.0,
1668 "192.0.2.1".parse().unwrap(),
1669 Asn::new_16bit(64496)
1670 )
1671 .is_err());
1672
1673 let routes = collect_route_record_iter(
1674 parse_bgp_message_routes(
1675 raw_bgp_message(30, BgpMessageType::KEEPALIVE, &[]),
1676 false,
1677 &AsnLength::Bits16,
1678 1_700_000_000.0,
1679 "192.0.2.1".parse().unwrap(),
1680 Asn::new_16bit(64496),
1681 )
1682 .unwrap(),
1683 )
1684 .unwrap();
1685 assert!(routes.is_empty());
1686
1687 let routes = collect_route_record_iter(
1688 parse_bgp_message_routes(
1689 raw_bgp_message(19, BgpMessageType::KEEPALIVE, &[0]),
1690 false,
1691 &AsnLength::Bits16,
1692 1_700_000_000.0,
1693 "192.0.2.1".parse().unwrap(),
1694 Asn::new_16bit(64496),
1695 )
1696 .unwrap(),
1697 )
1698 .unwrap();
1699 assert!(routes.is_empty());
1700
1701 let routes = collect_route_record_iter(
1702 parse_bgp_message_routes(
1703 raw_bgp_message_with_marker([0x00; 16], 19, BgpMessageType::KEEPALIVE, &[]),
1704 false,
1705 &AsnLength::Bits16,
1706 1_700_000_000.0,
1707 "192.0.2.1".parse().unwrap(),
1708 Asn::new_16bit(64496),
1709 )
1710 .unwrap(),
1711 )
1712 .unwrap();
1713 assert!(routes.is_empty());
1714 }
1715
1716 #[test]
1717 fn route_core_dump_write_respects_enabled_flag() {
1718 let dir = tempfile::tempdir().unwrap();
1719 let path = dir.path().join("mrt_core_dump");
1720
1721 write_mrt_core_dump_to_path(false, Some(vec![1, 2, 3]), &path);
1722 assert!(!path.exists());
1723
1724 write_mrt_core_dump_to_path(true, Some(vec![1, 2, 3]), &path);
1725 assert_eq!(std::fs::read(&path).unwrap(), vec![1, 2, 3]);
1726 }
1727
1728 #[test]
1729 fn route_parser_handles_table_dump_v2_error_edges() {
1730 let rib = RibAfiEntries {
1731 rib_type: TableDumpV2Type::RibIpv4Unicast,
1732 sequence_number: 1,
1733 prefix: NetworkPrefix::from_str("203.0.113.0/24").unwrap(),
1734 rib_entries: vec![RibEntry {
1735 peer_index: 99,
1736 originated_time: 1_699_999_999,
1737 path_id: None,
1738 attributes: route_attributes([64500, 64501]),
1739 }],
1740 };
1741 let mut no_peer_table = None;
1742 assert!(parse_table_dump_v2_routes(
1743 TableDumpV2Type::RibIpv4Unicast as u16,
1744 rib.encode().unwrap(),
1745 &mut no_peer_table,
1746 )
1747 .is_err());
1748
1749 let mut empty_peer_table = Some(RoutePeerTable::default());
1750 let routes = collect_route_record_iter(
1751 parse_table_dump_v2_routes(
1752 TableDumpV2Type::RibIpv4Unicast as u16,
1753 rib.encode().unwrap(),
1754 &mut empty_peer_table,
1755 )
1756 .unwrap(),
1757 )
1758 .unwrap();
1759 assert!(routes.is_empty());
1760
1761 let mut truncated = BytesMut::new();
1762 truncated.put_u32(1);
1763 truncated.extend(NetworkPrefix::from_str("203.0.113.0/24").unwrap().encode());
1764 truncated.put_u16(1);
1765 let mut empty_peer_table = Some(RoutePeerTable::default());
1766 let routes = collect_route_record_iter(
1767 parse_table_dump_v2_routes(
1768 TableDumpV2Type::RibIpv4Unicast as u16,
1769 truncated.freeze(),
1770 &mut empty_peer_table,
1771 )
1772 .unwrap(),
1773 )
1774 .unwrap();
1775 assert!(routes.is_empty());
1776
1777 let peer = Peer::new(
1778 "192.0.2.10".parse().unwrap(),
1779 "192.0.2.11".parse().unwrap(),
1780 Asn::new_32bit(64496),
1781 );
1782 let mut peer_table = PeerIndexTable::default();
1783 let peer_index = peer_table.add_peer(peer).unwrap();
1784
1785 let first_entry = RibEntry {
1786 peer_index,
1787 originated_time: 1_699_999_999,
1788 path_id: Some(1234),
1789 attributes: route_attributes([64500, 64501]),
1790 };
1791 let mut add_path_truncated = BytesMut::new();
1792 add_path_truncated.put_u32(1);
1793 add_path_truncated.extend(NetworkPrefix::from_str("203.0.113.0/24").unwrap().encode());
1794 add_path_truncated.put_u16(2);
1795 add_path_truncated.extend(first_entry.encode().unwrap());
1796 add_path_truncated.put_u16(peer_index);
1797 add_path_truncated.put_u32(1_699_999_998);
1798 add_path_truncated.put_u32(5678);
1799
1800 let mut peer_table = Some(route_peer_table_from_peer_index(peer_table));
1801 let routes = collect_route_record_iter(
1802 parse_table_dump_v2_routes(
1803 TableDumpV2Type::RibIpv4UnicastAddPath as u16,
1804 add_path_truncated.freeze(),
1805 &mut peer_table,
1806 )
1807 .unwrap(),
1808 )
1809 .unwrap();
1810 assert_eq!(routes.len(), 1);
1811 assert_eq!(
1812 routes[0].prefix,
1813 NetworkPrefix::from_str("203.0.113.0/24").unwrap()
1814 );
1815 }
1816
1817 #[test]
1818 fn route_parser_preserves_table_dump_v2_routes_before_truncated_attribute_payload() {
1819 let (_bytes, rib_body, peer_table) = table_dump_v2_truncated_attribute_payload();
1820 let mut peer_table = Some(route_peer_table_from_peer_index(peer_table));
1821
1822 let routes = collect_route_record_iter(
1823 parse_table_dump_v2_routes(
1824 TableDumpV2Type::RibIpv4Unicast as u16,
1825 rib_body,
1826 &mut peer_table,
1827 )
1828 .unwrap(),
1829 )
1830 .unwrap();
1831
1832 assert_eq!(routes.len(), 1);
1833 assert_eq!(
1834 routes[0].prefix,
1835 NetworkPrefix::from_str("203.0.113.0/24").unwrap()
1836 );
1837 assert_eq!(
1838 routes[0].as_path.as_ref().unwrap().to_u32_vec_opt(false),
1839 Some(vec![64500, 64501])
1840 );
1841 }
1842
1843 #[test]
1844 fn route_iterators_preserve_table_dump_v2_routes_before_truncated_attribute_payload() {
1845 let (bytes, _rib_body, _peer_table) = table_dump_v2_truncated_attribute_payload();
1846
1847 let routes = BgpkitParser::from_reader(Cursor::new(bytes.clone()))
1848 .into_route_iter()
1849 .collect::<Vec<_>>();
1850 assert_eq!(routes.len(), 1);
1851 assert_eq!(
1852 routes[0].prefix,
1853 NetworkPrefix::from_str("203.0.113.0/24").unwrap()
1854 );
1855
1856 let fallible_routes = BgpkitParser::from_reader(Cursor::new(bytes))
1857 .into_fallible_route_iter()
1858 .collect::<Result<Vec<_>, _>>()
1859 .unwrap();
1860 assert_eq!(fallible_routes, routes);
1861 }
1862
1863 fn table_dump_v2_rib_without_peer_table_record() -> MrtRecord {
1864 MrtRecord {
1865 common_header: CommonHeader {
1866 timestamp: 1_700_000_001,
1867 microsecond_timestamp: None,
1868 entry_type: EntryType::TABLE_DUMP_V2,
1869 entry_subtype: TableDumpV2Type::RibIpv4Unicast as u16,
1870 length: 0,
1871 },
1872 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::RibAfi(RibAfiEntries {
1873 rib_type: TableDumpV2Type::RibIpv4Unicast,
1874 sequence_number: 1,
1875 prefix: NetworkPrefix::from_str("203.0.113.0/24").unwrap(),
1876 rib_entries: vec![RibEntry {
1877 peer_index: 0,
1878 originated_time: 1_699_999_999,
1879 path_id: None,
1880 attributes: route_attributes([64500, 64501]),
1881 }],
1882 })),
1883 }
1884 }
1885
1886 #[test]
1887 fn route_iterator_skips_route_parse_errors() {
1888 let routes = BgpkitParser::from_reader(Cursor::new(
1889 table_dump_v2_rib_without_peer_table_record()
1890 .encode()
1891 .unwrap()
1892 .to_vec(),
1893 ))
1894 .into_route_iter()
1895 .collect::<Vec<_>>();
1896
1897 assert!(routes.is_empty());
1898 }
1899
1900 #[test]
1901 fn fallible_route_iterator_applies_filters_to_cached_routes() {
1902 let routes =
1903 BgpkitParser::from_reader(Cursor::new(update_record().encode().unwrap().to_vec()))
1904 .add_filter("type", "w")
1905 .unwrap()
1906 .into_fallible_route_iter()
1907 .collect::<Result<Vec<_>, _>>()
1908 .unwrap();
1909
1910 assert_eq!(routes.len(), 1);
1911 assert_eq!(routes[0].elem_type, ElemType::WITHDRAW);
1912 }
1913
1914 #[test]
1915 fn fallible_route_iterator_returns_route_parse_errors() {
1916 let bytes = table_dump_v2_rib_without_peer_table_record()
1917 .encode()
1918 .unwrap()
1919 .to_vec();
1920 let mut iter =
1921 BgpkitParser::from_reader(Cursor::new(bytes.clone())).into_fallible_route_iter();
1922
1923 let error = iter.next().unwrap().unwrap_err();
1924 assert_eq!(error.bytes.as_deref(), Some(bytes.as_slice()));
1925 }
1926
1927 #[test]
1928 fn fallible_route_iterator_yields_routes() {
1929 let bytes = update_record().encode().unwrap().to_vec();
1930 let routes = BgpkitParser::from_reader(Cursor::new(bytes))
1931 .into_fallible_route_iter()
1932 .collect::<Result<Vec<_>, _>>()
1933 .unwrap();
1934
1935 assert_eq!(routes.len(), 2);
1936 assert_eq!(routes[0].elem_type, ElemType::ANNOUNCE);
1937 assert_eq!(routes[1].elem_type, ElemType::WITHDRAW);
1938 }
1939
1940 #[test]
1941 fn fallible_route_iterator_returns_parse_errors() {
1942 let invalid_data = vec![
1943 0x00, 0x00, 0x00, 0x00, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, ];
1949
1950 let mut iter =
1951 BgpkitParser::from_reader(Cursor::new(invalid_data.clone())).into_fallible_route_iter();
1952
1953 let error = iter.next().unwrap().unwrap_err();
1956 assert_eq!(error.bytes.as_deref(), Some(invalid_data.as_slice()));
1957 assert!(iter.next().is_none());
1958 }
1959}