diff --git a/CHANGELOG.md b/CHANGELOG.md index 6025678..d8e97c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,7 @@ All notable changes to this project will be documented in this file. #### Test data * **Historical RIPE regression fixtures**: Added original RRC00 update and bview gzip files from 1999 and January 2000 as repository-only, offline integration fixtures. +* **Opt-in MRT framing recovery**: Added `into_recovering_record_iter`, `into_recovering_elem_iter`, and CLI `--recover` support for salvaging records after damaged MRT framing. Recovery validates a three-record chain, uses exact embedded BGP headers as BGP4MP anchors, and reports every skipped decompressed byte range as a typed gap event. A record of any MRT type whose framing is intact but whose body fails to parse is skipped exactly (`AlignedRecordChain` evidence) when intact records follow its declared boundary, and damage extending to the end of the stream — e.g. a truncated final record — is reported as a terminal gap (`EndOfStream` evidence) instead of an error. Text-dump parsers, which have no MRT record representation, yield an explicit `Unsupported` error. #### Examples @@ -66,6 +67,7 @@ All notable changes to this project will be documented in this file. * **BGP OPEN parameter type 255 rejected**: RFC 9072 reserves type 255 as the extended-length marker; encoding it as a real parameter produced output that round-tripped to a structurally different message. * **BGP OPEN optional-parameter encoding**: Encode the Optional Parameters Length as the total byte length required by RFC 4271 instead of the number of parameters. OPEN messages now also use the extended length format from RFC 9072 when requested or required. * **Historical Quagga state changes**: Parse BGP4MP state-change records containing Quagga's `Clearing` (7) and `Deleted` (8) FSM states instead of logging an error and dropping the MRT record. +* **CLI broken-pipe exit status**: the CLI now exits 0 when its stdout consumer closes the pipe early (e.g. `bgpkit-parser updates.gz | head`), matching Unix convention; previous releases exited 1. Other write errors still exit 1. ### Contributors diff --git a/Cargo.toml b/Cargo.toml index b6ca253..72b818e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,8 @@ exclude = [ "tests/fixtures/ripe/**", "tests/legacy_ripe_2000.rs", "tests/quagga_bgp_states.rs", + "tests/recovery.rs", + "tests/torn_mrt_records.rs", ] [[bin]] diff --git a/README.md b/README.md index ba389a4..728fe1e 100644 --- a/README.md +++ b/README.md @@ -246,6 +246,32 @@ match process_mrt_file("http://example.com/updates.bz2") { } ``` +**Recovering After Damaged MRT Framing** + +Recovery is opt-in and never reconstructs a damaged record. It reports the skipped decompressed +byte range before resuming at a conservatively validated record chain. A correctly framed record +whose body fails to parse is skipped exactly, and damage extending to the end of the stream +(e.g. a truncated final record) is reported as a terminal gap rather than an error. Use +`into_recovering_elem_iter` for the same events at the `BgpElem` level. + +```rust +use bgpkit_parser::{BgpkitParser, RecoveryConfig, RecoveryEvent}; + +fn recover(path: &str) -> Result<(), Box> { + let parser = BgpkitParser::new(path)?; + for event in parser.into_recovering_record_iter(RecoveryConfig::default()) { + match event? { + RecoveryEvent::Item(record) => println!("{}", record), + RecoveryEvent::Gap(gap) => eprintln!( + "skipped bytes {}..{}: {}", + gap.start_offset, gap.end_offset, gap.cause + ), + } + } + Ok(()) +} +``` + ### Advanced Examples #### Parsing Real-time Data Streams @@ -481,6 +507,7 @@ Options: --pretty Pretty-print JSON output -e, --elems-count Count BGP elems -r, --records-count Count MRT records + --recover Recover after damaged MRT framing and report skipped byte ranges on stderr -o, --origin-asn Filter by origin AS Number -f, --filter Generic filter expression (key=value or key!=value) -p, --prefix Filter by network prefix @@ -526,6 +553,11 @@ bgpkit-parser --json updates.20211001.0000.bz2 > output.json bgpkit-parser -e updates.20211001.0000.bz2 ``` +#### Recover records after damaged framing +```bash +bgpkit-parser --recover -e damaged-updates.gz +``` + #### Cache remote files for faster repeated access ```bash bgpkit-parser -c ~/.bgpkit-cache http://example.com/updates.mrt.bz2 diff --git a/src/bin/main.rs b/src/bin/main.rs index 74f62de..e5b597f 100644 --- a/src/bin/main.rs +++ b/src/bin/main.rs @@ -4,7 +4,7 @@ use std::io::Write; use std::net::IpAddr; use std::path::PathBuf; -use bgpkit_parser::{BgpElem, BgpkitParser, Elementor}; +use bgpkit_parser::{BgpElem, BgpkitParser, Elementor, RecoveryConfig, RecoveryEvent, RecoveryGap}; use clap::{Parser, ValueEnum}; use ipnet::IpNet; @@ -72,6 +72,10 @@ struct Opts { #[clap(short, long)] records_count: bool, + /// Recover after damaged MRT framing and report skipped byte ranges on stderr + #[clap(long)] + recover: bool, + #[clap(flatten)] filters: Filters, } @@ -248,52 +252,202 @@ fn main() { opts.format }; - match (opts.elems_count, opts.records_count) { - (true, true) => { - let mut elementor = Elementor::new(); - let (mut records_count, mut elems_count) = (0, 0); - for record in parser.into_record_iter() { - records_count += 1; - elems_count += elementor.record_to_elems(record).len(); - } - println!("total records: {records_count}"); - println!("total elems: {elems_count}"); + let recovery_config = RecoveryConfig::default(); + // Element-level runs (element output or counting only elements) use the elem + // iterators, which apply filters per element; everything else stays at the record + // level. Counting both (-e -r) iterates records and converts once per record. + let use_elem_stream = (opts.elems_count && !opts.records_count) + || (!opts.elems_count && !opts.records_count && matches!(opts.level, OutputLevel::Elems)); + + let result = match (opts.recover, use_elem_stream) { + (true, true) => run_elems( + parser + .into_recovering_elem_iter(recovery_config) + .map(|event| event.map_err(|error| error.to_string())), + output_format, + opts.elems_count, + true, + ), + (false, true) => run_elems( + parser + .into_elem_iter() + .map(|elem| Ok(RecoveryEvent::Item(elem))), + output_format, + opts.elems_count, + false, + ), + (true, false) => run_records( + parser + .into_recovering_record_iter(recovery_config) + .map(|event| event.map_err(|error| error.to_string())), + output_format, + opts.elems_count, + opts.records_count, + true, + ), + (false, false) => run_records( + parser + .into_record_iter() + .map(|record| Ok(RecoveryEvent::Item(record))), + output_format, + opts.elems_count, + opts.records_count, + false, + ), + }; + if let Err(error) = result { + eprintln!("{error}"); + std::process::exit(1); + } +} + +/// Per-gap reporting and end-of-run summary for `--recover`. +struct RecoveryStats { + enabled: bool, + gap_count: usize, + skipped_bytes: u64, +} + +impl RecoveryStats { + fn new(enabled: bool) -> Self { + Self { + enabled, + gap_count: 0, + skipped_bytes: 0, } - (false, true) => { - println!("total records: {}", parser.into_record_iter().count()); + } + + fn observe(&mut self, gap: &RecoveryGap) { + self.gap_count += 1; + self.skipped_bytes += gap.skipped_bytes(); + eprintln!( + "recovered MRT framing: skipped bytes {}..{} ({} bytes, {:?}, {} confirming records): {}", + gap.start_offset, + gap.end_offset, + gap.skipped_bytes(), + gap.evidence, + gap.confirmed_records, + gap.cause + ); + } + + fn print_summary(&self) { + if self.enabled { + eprintln!( + "recovery summary: {} gaps, {} bytes skipped", + self.gap_count, self.skipped_bytes + ); } - (true, false) => { - println!("total elems: {}", parser.into_elem_iter().count()); + } +} + +fn run_elems( + events: I, + output_format: OutputFormat, + count_requested: bool, + report_recovery: bool, +) -> Result<(), String> +where + I: IntoIterator, String>>, +{ + let mut stdout = std::io::stdout(); + let mut elems_count = 0usize; + let mut elem_index = 0usize; + let mut stats = RecoveryStats::new(report_recovery); + let mut terminal_error = None; + + for event in events { + match event { + Err(error) => { + terminal_error = Some(error); + break; + } + Ok(RecoveryEvent::Gap(gap)) => stats.observe(&gap), + Ok(RecoveryEvent::Item(elem)) => { + elems_count += 1; + if count_requested { + continue; + } + let output = format_elem(&elem, output_format, elem_index); + elem_index += 1; + if !write_output(&mut stdout, &output)? { + return Ok(()); + } + } } - (false, false) => { - let mut stdout = std::io::stdout(); - - match opts.level { - OutputLevel::Elems => { - for (index, elem) in parser.into_elem_iter().enumerate() { - let output_str = format_elem(&elem, output_format, index); - if let Err(e) = writeln!(stdout, "{}", output_str) { - if e.kind() != std::io::ErrorKind::BrokenPipe { - eprintln!("{e}"); - } - std::process::exit(1); - } - } + } + + if count_requested { + println!("total elems: {elems_count}"); + } + stats.print_summary(); + terminal_error.map_or(Ok(()), Err) +} + +fn run_records( + events: I, + output_format: OutputFormat, + elems_count_requested: bool, + records_count_requested: bool, + report_recovery: bool, +) -> Result<(), String> +where + I: IntoIterator, String>>, +{ + let mut stdout = std::io::stdout(); + let mut elementor = Elementor::new(); + let mut records_count = 0usize; + let mut elems_count = 0usize; + let mut stats = RecoveryStats::new(report_recovery); + let mut terminal_error = None; + + for event in events { + match event { + Err(error) => { + terminal_error = Some(error); + break; + } + Ok(RecoveryEvent::Gap(gap)) => stats.observe(&gap), + Ok(RecoveryEvent::Item(record)) => { + records_count += 1; + if elems_count_requested { + // Counting both (-e -r): count every element of records that passed + // record-level filtering, matching the historical CLI behavior. + elems_count += elementor.record_to_elems(record).len(); + continue; } - OutputLevel::Records => { - for record in parser.into_record_iter() { - let output_str = format_record(&record, output_format); - if let Err(e) = writeln!(stdout, "{}", output_str) { - if e.kind() != std::io::ErrorKind::BrokenPipe { - eprintln!("{e}"); - } - std::process::exit(1); - } - } + if records_count_requested { + continue; + } + let output = format_record(&record, output_format); + if !write_output(&mut stdout, &output)? { + return Ok(()); } } } } + + match (elems_count_requested, records_count_requested) { + (true, true) => { + println!("total records: {records_count}"); + println!("total elems: {elems_count}"); + } + (false, true) => println!("total records: {records_count}"), + (true, false) => println!("total elems: {elems_count}"), + (false, false) => {} + } + stats.print_summary(); + terminal_error.map_or(Ok(()), Err) +} + +fn write_output(stdout: &mut std::io::Stdout, output: &str) -> Result { + if let Err(error) = writeln!(stdout, "{output}") { + if error.kind() == std::io::ErrorKind::BrokenPipe { + return Ok(false); + } + return Err(error.to_string()); + } + Ok(true) } fn format_elem(elem: &BgpElem, format: OutputFormat, index: usize) -> String { diff --git a/src/lib.rs b/src/lib.rs index 291840d..facf60c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -236,6 +236,29 @@ match process_mrt_file("http://example.com/updates.bz2") { } ``` +**Recovering After Damaged MRT Framing** + +Recovery is opt-in and never reconstructs a damaged record. It reports the skipped decompressed +byte range before resuming at a conservatively validated record chain. + +```no_run +use bgpkit_parser::{BgpkitParser, RecoveryConfig, RecoveryEvent}; + +fn recover(path: &str) -> Result<(), Box> { + let parser = BgpkitParser::new(path)?; + for event in parser.into_recovering_record_iter(RecoveryConfig::default()) { + match event? { + RecoveryEvent::Item(record) => println!("{}", record), + RecoveryEvent::Gap(gap) => eprintln!( + "skipped bytes {}..{}: {}", + gap.start_offset, gap.end_offset, gap.cause + ), + } + } + Ok(()) +} +``` + ## Advanced Examples ### Parsing Real-time Data Streams @@ -477,6 +500,7 @@ Options: --pretty Pretty-print JSON output -e, --elems-count Count BGP elems -r, --records-count Count MRT records + --recover Recover after damaged MRT framing and report skipped byte ranges on stderr -o, --origin-asn Filter by origin AS Number -f, --filter Generic filter expression (key=value or key!=value) -p, --prefix Filter by network prefix @@ -522,6 +546,11 @@ bgpkit-parser --json updates.20211001.0000.bz2 > output.json bgpkit-parser -e updates.20211001.0000.bz2 ``` +### Recover records after damaged framing +```bash +bgpkit-parser --recover -e damaged-updates.gz +``` + ### Cache remote files for faster repeated access ```bash bgpkit-parser -c ~/.bgpkit-cache http://example.com/updates.mrt.bz2 diff --git a/src/parser/iters/default.rs b/src/parser/iters/default.rs index a10334c..0edb466 100644 --- a/src/parser/iters/default.rs +++ b/src/parser/iters/default.rs @@ -3,7 +3,7 @@ Default iterator implementations that skip errors and return successfully parsed */ use crate::error::ParserError; use crate::models::*; -use crate::parser::iters::write_mrt_core_dump; +use crate::parser::iters::{record_matches_filters, write_mrt_core_dump}; use crate::parser::BgpkitParser; use crate::{Elementor, Filterable}; use log::{error, warn}; @@ -42,24 +42,10 @@ impl Iterator for RecordIterator { loop { return match self.parser.next_record() { Ok(v) => { - // if None, the reaches EoF. - let filters = &self.parser.filters; - if filters.is_empty() { + if record_matches_filters(&v, &self.parser.filters, &mut self.elementor) { Some(v) } else { - if let MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable( - _, - )) = &v.message - { - let _ = self.elementor.record_to_elems(v.clone()); - return Some(v); - } - let elems = self.elementor.record_to_elems(v.clone()); - if elems.iter().any(|e| e.match_filters(&self.parser.filters)) { - Some(v) - } else { - continue; - } + continue; } } Err(e) => { diff --git a/src/parser/iters/mod.rs b/src/parser/iters/mod.rs index 3f5ffc5..f58afad 100644 --- a/src/parser/iters/mod.rs +++ b/src/parser/iters/mod.rs @@ -14,6 +14,7 @@ pub mod default; mod diagnostic; pub mod fallible; mod raw; +mod recovery; mod route; mod update; @@ -22,6 +23,10 @@ pub use default::{ElemIterator, RecordIterator}; pub use diagnostic::{DiagnosticEvent, DiagnosticIterator}; pub use fallible::{FallibleElemIterator, FallibleRecordIterator}; pub use raw::RawRecordIterator; +pub use recovery::{ + RecoveringElemIterator, RecoveringRecordIterator, RecoveryConfig, RecoveryError, RecoveryEvent, + RecoveryEvidence, RecoveryGap, +}; pub use route::{FallibleRouteIterator, RouteIterator}; pub use update::{ Bgp4MpUpdate, FallibleUpdateIterator, LegacyBgpUpdate, MrtUpdate, TableDumpV2Entry, @@ -31,11 +36,33 @@ pub use update::{ use crate::models::BgpElem; use crate::models::{MrtMessage, MrtRecord, TableDumpV2Message}; use crate::parser::BgpkitParser; -use crate::Elementor; use crate::RawMrtRecord; +use crate::{Elementor, Filter, Filterable}; use std::io::Read; use std::path::Path; +#[inline] +pub(crate) fn record_matches_filters( + record: &MrtRecord, + filters: &[Filter], + elementor: &mut Elementor, +) -> bool { + if filters.is_empty() { + return true; + } + if matches!( + &record.message, + MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(_)) + ) { + let _ = elementor.record_to_elems(record.clone()); + return true; + } + elementor + .record_to_elems(record.clone()) + .iter() + .any(|element| element.match_filters(filters)) +} + pub(crate) fn write_mrt_core_dump(enabled: bool, bytes: Option>) { write_mrt_core_dump_to_path(enabled, bytes, "mrt_core_dump"); } @@ -75,6 +102,29 @@ impl BgpkitParser { RawRecordIterator::new(self) } + /// Creates an opt-in iterator that reports skipped byte ranges while recovering MRT framing. + /// + /// Recovery never reconstructs a damaged record. It scans for a structurally valid boundary, + /// confirms a chain of records, emits [`RecoveryEvent::Gap`], and then resumes normal parsing. + /// Damage extending to the end of the stream is reported as a terminal gap. Offsets in + /// recovery events refer to the decompressed MRT byte stream. + pub fn into_recovering_record_iter( + self, + config: RecoveryConfig, + ) -> RecoveringRecordIterator { + RecoveringRecordIterator::new(self, config) + } + + /// Creates an opt-in iterator over BGP elements that reports skipped byte ranges while + /// recovering MRT framing. + /// + /// Behaves like [`into_recovering_record_iter`](Self::into_recovering_record_iter) but + /// converts each recovered record to [`BgpElem`]s, applying the parser's filters per + /// element. + pub fn into_recovering_elem_iter(self, config: RecoveryConfig) -> RecoveringElemIterator { + RecoveringElemIterator::new(self, config) + } + /// Creates an iterator over BGP announcements from MRT data. /// /// This iterator yields `MrtUpdate` items from both UPDATES files (BGP4MP messages) diff --git a/src/parser/iters/recovery.rs b/src/parser/iters/recovery.rs new file mode 100644 index 0000000..c839217 --- /dev/null +++ b/src/parser/iters/recovery.rs @@ -0,0 +1,1164 @@ +//! Opt-in MRT framing recovery. +//! +//! Recovery is deliberately separate from the default iterators. It never attempts to +//! reconstruct a damaged record: bytes are skipped until a conservatively validated chain of +//! records is found, and the skipped range is reported as a [`RecoveryEvent::Gap`]. +//! +//! Damage is classified before scanning. When a record frames correctly — its header and +//! declared length were consumed exactly — but its body fails to parse, and intact records +//! (or a clean end of stream) follow at the declared boundary, exactly that record is +//! skipped without scanning. Damage that extends to the end of the stream is reported as a +//! terminal gap rather than an error, so trailing truncation — the most common real-world +//! corruption — still yields every intact record plus an explicit account of the discarded +//! tail. A [`RecoveryError`] is reserved for I/O failures, unsupported input, and scan +//! windows exhausted without finding a boundary mid-stream. +//! +//! The undamaged fast path reads straight from the underlying reader; bytes are only +//! buffered while a recovery scan is in progress. + +use crate::models::{Bgp4MpType, BgpElem, EntryType, MrtRecord}; +use crate::parser::iters::{record_matches_filters, write_mrt_core_dump}; +use crate::parser::mrt::messages::bgp4mp::uses_zebra_compat; +use crate::parser::mrt::mrt_header::parse_common_header_with_bytes; +use crate::parser::mrt::mrt_record::{ + chunk_mrt_record, parse_mrt_record_with_zebra_compat, raw_record_uses_zebra_compat, +}; +use crate::parser::{ + BgpkitParser, Elementor, Filter, ParserError, ParserErrorWithBytes, ParserOptions, +}; +use crate::Filterable; +use bytes::Bytes; +use std::fmt::{Display, Formatter}; +use std::io::{self, Read}; + +const DEFAULT_MAX_SCAN_BYTES: usize = 1024 * 1024; +const DEFAULT_CONFIRMATION_RECORDS: u8 = 3; +const MAX_RECOVERY_RECORD_LEN: u32 = 65_599; +const SCAN_FILL_CHUNK: usize = 8_192; + +/// Settings for opt-in MRT framing recovery. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RecoveryConfig { + max_scan_bytes: usize, + confirmation_records: u8, +} + +impl Default for RecoveryConfig { + fn default() -> Self { + Self { + max_scan_bytes: DEFAULT_MAX_SCAN_BYTES, + confirmation_records: DEFAULT_CONFIRMATION_RECORDS, + } + } +} + +impl RecoveryConfig { + /// Set the maximum number of bytes searched after a damaged record. + pub const fn with_max_scan_bytes(mut self, max_scan_bytes: usize) -> Self { + self.max_scan_bytes = max_scan_bytes; + self + } + + /// Set the number of consecutive records required to confirm a recovered boundary. + /// + /// A value of zero is treated as one. + pub const fn with_confirmation_records(mut self, confirmation_records: u8) -> Self { + self.confirmation_records = confirmation_records; + self + } + + pub const fn max_scan_bytes(&self) -> usize { + self.max_scan_bytes + } + + pub const fn confirmation_records(&self) -> u8 { + self.confirmation_records + } +} + +/// Evidence used to validate the first record following a recovered gap. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +#[non_exhaustive] +pub enum RecoveryEvidence { + /// A deprecated MRT Type-5 record and its confirmation chain parsed structurally. + LegacyMrtChain, + /// A BGP4MP message contained an exact embedded BGP marker and length. + BgpMarkerChain, + /// A BGP4MP state-change record, which has no embedded BGP message header. + Bgp4MpStateChangeChain, + /// The damaged record's framing was intact: complete MRT records of any type (or a + /// clean end of stream) followed at its declared end offset, so exactly that record + /// was skipped without scanning. + AlignedRecordChain, + /// No boundary was validated before the stream ended; the gap extends to the end of + /// the input. + EndOfStream, +} + +/// A byte range discarded while restoring MRT record framing. +#[derive(Debug, Clone, PartialEq, Eq)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +pub struct RecoveryGap { + /// Inclusive offset in the decompressed MRT byte stream. + pub start_offset: u64, + /// Exclusive offset in the decompressed MRT byte stream. + pub end_offset: u64, + /// Error raised while parsing at `start_offset`. + pub cause: String, + /// Structural evidence used to accept `end_offset` as a new boundary. + pub evidence: RecoveryEvidence, + /// Number of consecutive records validated at the recovered boundary. Zero when the + /// gap ends at the end of the stream. + pub confirmed_records: u8, +} + +impl RecoveryGap { + /// Return the number of skipped bytes, or zero for an invalid inverted range. + pub const fn skipped_bytes(&self) -> u64 { + self.end_offset.saturating_sub(self.start_offset) + } +} + +/// An item produced by a recovering iterator. +#[derive(Debug)] +pub enum RecoveryEvent { + Item(T), + Gap(RecoveryGap), +} + +/// A framing error for which no recovery boundary was found within the scan window, an +/// I/O failure, or unsupported input. +/// +/// Damage that extends to the end of the stream is reported as a terminal +/// [`RecoveryEvent::Gap`] instead of this error. +#[derive(Debug)] +pub struct RecoveryError { + pub offset: u64, + pub scanned_bytes: u64, + pub error: ParserErrorWithBytes, +} + +impl Display for RecoveryError { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!( + f, + "MRT recovery failed at decompressed offset {} after scanning {} bytes: {}", + self.offset, self.scanned_bytes, self.error + ) + } +} + +impl std::error::Error for RecoveryError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&self.error) + } +} + +/// Iterator over parsed MRT records and explicit recovery gaps. +pub struct RecoveringRecordIterator { + reader: CarryoverReader, + config: RecoveryConfig, + filters: Vec, + elementor: Elementor, + options: ParserOptions, + core_dump: bool, + unsupported_input: Option, + finished: bool, +} + +impl RecoveringRecordIterator { + pub(crate) fn new(parser: BgpkitParser, config: RecoveryConfig) -> Self { + let unsupported_input = parser.text_dump_iter.is_some().then(|| { + "text-dump parsers have no MRT record representation; iterate elements instead" + .to_string() + }); + Self { + reader: CarryoverReader::new(parser.reader), + config, + filters: parser.filters, + elementor: Elementor::new(), + options: parser.options, + core_dump: parser.core_dump, + unsupported_input, + finished: false, + } + } +} + +impl Iterator for RecoveringRecordIterator { + type Item = Result, RecoveryError>; + + fn next(&mut self) -> Option { + if self.finished { + return None; + } + if let Some(message) = self.unsupported_input.take() { + self.finished = true; + return Some(Err(RecoveryError { + offset: 0, + scanned_bytes: 0, + error: ParserErrorWithBytes::from(ParserError::Unsupported(message)), + })); + } + + loop { + let record_start = self.reader.position(); + let raw_record = match chunk_mrt_record(&mut self.reader) { + Ok(raw_record) => raw_record, + Err(error) if matches!(error.error, ParserError::EofExpected) => { + self.finished = true; + return None; + } + Err(error) if is_non_eof_io_error(&error.error) => { + self.finished = true; + return Some(Err(RecoveryError { + offset: record_start, + scanned_bytes: 0, + error, + })); + } + Err(error) => return self.recover(record_start, None, error), + }; + + let used_zebra_compat = raw_record_uses_zebra_compat(&raw_record); + match raw_record.clone().parse() { + Ok(record) => { + if used_zebra_compat { + self.options.warn_zebra_compat_once(); + } + if record_matches_filters(&record, &self.filters, &mut self.elementor) { + return Some(Ok(RecoveryEvent::Item(record))); + } + } + Err(error) => { + // The header and declared length were consumed exactly, so the + // stream may still be aligned even though the body is unparsable. + let framed_end = self.reader.position(); + let error = ParserErrorWithBytes { + error, + bytes: Some(raw_record.raw_bytes().to_vec()), + }; + return self.recover(record_start, Some(framed_end), error); + } + } + } + } +} + +impl RecoveringRecordIterator { + fn recover( + &mut self, + record_start: u64, + framed_end: Option, + error: ParserErrorWithBytes, + ) -> Option, RecoveryError>> { + write_mrt_core_dump(self.core_dump, error.bytes.clone()); + let consumed = error.bytes.clone().unwrap_or_default(); + debug_assert_eq!(record_start + consumed.len() as u64, self.reader.position()); + let confirmations = self.config.confirmation_records.max(1); + let max_scan_bytes = self.config.max_scan_bytes; + let mut window = ReplayReader::seeded(&mut self.reader, consumed, record_start); + + if let Some(framed_end) = framed_end { + match confirm_aligned_boundary(&mut window, framed_end, confirmations) { + Ok(Some(confirmed_records)) => { + let leftover = window.into_leftover(framed_end); + self.reader.resume_with(leftover, framed_end); + return Some(Ok(RecoveryEvent::Gap(RecoveryGap { + start_offset: record_start, + end_offset: framed_end, + cause: error.to_string(), + evidence: RecoveryEvidence::AlignedRecordChain, + confirmed_records, + }))); + } + Ok(None) => {} + Err(io_error) => { + self.finished = true; + return Some(Err(RecoveryError { + offset: record_start, + scanned_bytes: 0, + error: ParserErrorWithBytes::from(ParserError::IoError(io_error)), + })); + } + } + } + + match find_recovery(&mut window, record_start, max_scan_bytes, confirmations) { + Ok(ScanOutcome::Found { + offset, + evidence, + confirmed_records, + }) => { + let leftover = window.into_leftover(offset); + self.reader.resume_with(leftover, offset); + Some(Ok(RecoveryEvent::Gap(RecoveryGap { + start_offset: record_start, + end_offset: offset, + cause: error.to_string(), + evidence, + confirmed_records, + }))) + } + Ok(ScanOutcome::EndOfStream { end_offset }) => { + let leftover = window.into_leftover(end_offset); + self.reader.resume_with(leftover, end_offset); + Some(Ok(RecoveryEvent::Gap(RecoveryGap { + start_offset: record_start, + end_offset, + cause: error.to_string(), + evidence: RecoveryEvidence::EndOfStream, + confirmed_records: 0, + }))) + } + Ok(ScanOutcome::WindowExhausted) => { + self.finished = true; + Some(Err(RecoveryError { + offset: record_start, + scanned_bytes: max_scan_bytes as u64, + error, + })) + } + Err(io_error) => { + self.finished = true; + Some(Err(RecoveryError { + offset: record_start, + scanned_bytes: window.position().saturating_sub(record_start), + error: ParserErrorWithBytes::from(ParserError::IoError(io_error)), + })) + } + } + } +} + +/// Iterator over BGP elements and explicit recovery gaps. +/// +/// Filters are applied per element; each record is converted to elements exactly once. +pub struct RecoveringElemIterator { + inner: RecoveringRecordIterator, + elementor: Elementor, + filters: Vec, + cache_elems: Vec, +} + +impl RecoveringElemIterator { + pub(crate) fn new(mut parser: BgpkitParser, config: RecoveryConfig) -> Self { + // Elements are filtered here; strip the parser filters so the inner record + // iterator does not also convert every record for record-level matching. + let filters = std::mem::take(&mut parser.filters); + Self { + inner: RecoveringRecordIterator::new(parser, config), + elementor: Elementor::new(), + filters, + cache_elems: Vec::new(), + } + } +} + +impl Iterator for RecoveringElemIterator { + type Item = Result, RecoveryError>; + + fn next(&mut self) -> Option { + loop { + if let Some(elem) = self.cache_elems.pop() { + return Some(Ok(RecoveryEvent::Item(elem))); + } + match self.inner.next()? { + Ok(RecoveryEvent::Item(record)) => { + let mut elems = self.elementor.record_to_elems(record); + elems.retain(|elem| elem.match_filters(&self.filters)); + elems.reverse(); + self.cache_elems = elems; + } + Ok(RecoveryEvent::Gap(gap)) => return Some(Ok(RecoveryEvent::Gap(gap))), + Err(error) => return Some(Err(error)), + } + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StreamFamily { + Legacy, + Bgp4Mp, +} + +struct Candidate { + family: StreamFamily, + evidence: RecoveryEvidence, +} + +enum ScanOutcome { + Found { + offset: u64, + evidence: RecoveryEvidence, + confirmed_records: u8, + }, + EndOfStream { + end_offset: u64, + }, + WindowExhausted, +} + +/// Confirm that intact records (of any MRT type) parse at the failed record's declared +/// end offset, distinguishing an unparsable-but-correctly-framed record from framing +/// damage. A clean end of stream on the boundary is consistent with intact framing. +fn confirm_aligned_boundary( + window: &mut ReplayReader, + boundary: u64, + required: u8, +) -> io::Result> { + if !window.move_to(boundary)? { + return Ok(None); + } + let mut confirmed = 0u8; + while confirmed < required { + match parse_mrt_record_with_zebra_compat(window) { + Ok(_) => confirmed += 1, + Err(error) => { + return match error.error { + ParserError::EofExpected => Ok(Some(confirmed)), + ParserError::IoError(inner) | ParserError::EofError(inner) + if inner.kind() != io::ErrorKind::UnexpectedEof => + { + Err(inner) + } + _ => Ok(None), + }; + } + } + } + Ok(Some(confirmed)) +} + +fn is_anchor_entry_type(bytes: [u8; 2]) -> bool { + let value = u16::from_be_bytes(bytes); + value == EntryType::BGP as u16 + || value == EntryType::BGP4MP as u16 + || value == EntryType::BGP4MP_ET as u16 +} + +fn find_recovery( + window: &mut ReplayReader, + failed_start: u64, + max_scan_bytes: usize, + confirmations: u8, +) -> io::Result { + for distance in 1..=max_scan_bytes { + let candidate = failed_start + distance as u64; + // Cheap anchor pre-filter: only offsets whose entry-type field matches a + // recoverable stream family warrant header parsing and chain validation. + let Some(entry_type) = window.peek_two_at(candidate + 4)? else { + return Ok(ScanOutcome::EndOfStream { + end_offset: window.buffered_end(), + }); + }; + if !is_anchor_entry_type(entry_type) { + continue; + } + if let Some((evidence, confirmed_records)) = + validate_chain(window, candidate, confirmations)? + { + return Ok(ScanOutcome::Found { + offset: candidate, + evidence, + confirmed_records, + }); + } + } + Ok(ScanOutcome::WindowExhausted) +} + +fn validate_chain( + reader: &mut ReplayReader, + offset: u64, + required: u8, +) -> io::Result> { + if !reader.move_to(offset)? { + return Ok(None); + } + let mut family = None; + let mut evidence = None; + let mut confirmed = 0u8; + + while confirmed < required { + let record_start = reader.position(); + let Some(candidate) = read_candidate(reader)? else { + reader.move_to(record_start)?; + return Ok(None); + }; + if family.is_some_and(|expected| expected != candidate.family) { + return Ok(None); + } + family.get_or_insert(candidate.family); + evidence.get_or_insert(candidate.evidence); + confirmed += 1; + } + + Ok(evidence.map(|evidence| (evidence, confirmed))) +} + +fn read_candidate(reader: &mut ReplayReader) -> io::Result> { + let parsed_header = match parse_common_header_with_bytes(reader) { + Ok(header) => header, + Err(error) => return parser_error_as_candidate(error), + }; + let header = parsed_header.header; + + let family = match header.entry_type { + EntryType::BGP if legacy_header_is_plausible(header.entry_subtype, header.length) => { + StreamFamily::Legacy + } + EntryType::BGP4MP | EntryType::BGP4MP_ET + if Bgp4MpType::try_from(header.entry_subtype).is_ok() + && header.length <= MAX_RECOVERY_RECORD_LEN => + { + StreamFamily::Bgp4Mp + } + _ => return Ok(None), + }; + + if header + .microsecond_timestamp + .is_some_and(|value| value >= 1_000_000) + { + return Ok(None); + } + + let mut body = Vec::with_capacity(header.length as usize); + reader + .by_ref() + .take(header.length as u64) + .read_to_end(&mut body)?; + if body.len() != header.length as usize { + // The stream ended inside the candidate body. + return Ok(None); + } + let raw_record = crate::RawMrtRecord { + common_header: header, + header_bytes: parsed_header.raw_bytes, + message_bytes: Bytes::from(body), + }; + + let evidence = match family { + StreamFamily::Legacy => raw_record + .clone() + .parse() + .ok() + .map(|_| RecoveryEvidence::LegacyMrtChain), + StreamFamily::Bgp4Mp => strict_bgp4mp_evidence(&raw_record), + }; + Ok(evidence.map(|evidence| Candidate { family, evidence })) +} + +fn parser_error_as_candidate(error: ParserError) -> io::Result> { + match error { + ParserError::IoError(error) | ParserError::EofError(error) + if error.kind() != io::ErrorKind::UnexpectedEof => + { + Err(error) + } + _ => Ok(None), + } +} + +fn legacy_header_is_plausible(subtype: u16, length: u32) -> bool { + match subtype { + 1 => (16..=65_535).contains(&length), + 3 => length == 10, + 5 => (22..=4_089).contains(&length), + 6 => (14..=65_535).contains(&length), + 7 => length == 12, + _ => false, + } +} + +fn strict_bgp4mp_evidence(raw_record: &crate::RawMrtRecord) -> Option { + let msg_type = Bgp4MpType::try_from(raw_record.common_header.entry_subtype).ok()?; + if matches!( + msg_type, + Bgp4MpType::StateChange | Bgp4MpType::StateChangeAs4 + ) { + let body = &raw_record.message_bytes; + if uses_zebra_compat(raw_record.common_header.entry_subtype, body) { + if body.len() != 8 { + return None; + } + } else { + let asn_pair_len = if matches!(msg_type, Bgp4MpType::StateChange) { + 4 + } else { + 8 + }; + let afi_offset = asn_pair_len + 2; + let afi = u16::from_be_bytes(body.get(afi_offset..afi_offset + 2)?.try_into().ok()?); + let address_len = match afi { + 1 => 4, + 2 => 16, + _ => return None, + }; + if body.len() != asn_pair_len + 4 + address_len * 2 + 4 { + return None; + } + } + raw_record.clone().parse().ok()?; + return Some(RecoveryEvidence::Bgp4MpStateChangeChain); + } + + let body = &raw_record.message_bytes; + let asn_pair_len = match msg_type { + Bgp4MpType::Message + | Bgp4MpType::MessageLocal + | Bgp4MpType::MessageAddpath + | Bgp4MpType::MessageLocalAddpath => 4, + Bgp4MpType::MessageAs4 + | Bgp4MpType::MessageAs4Local + | Bgp4MpType::MessageAs4Addpath + | Bgp4MpType::MessageLocalAs4Addpath => 8, + Bgp4MpType::StateChange | Bgp4MpType::StateChangeAs4 => return None, + }; + + let marker_offset = if uses_zebra_compat(raw_record.common_header.entry_subtype, body) { + asn_pair_len + } else { + let afi_offset = asn_pair_len + 2; + let afi = u16::from_be_bytes(body.get(afi_offset..afi_offset + 2)?.try_into().ok()?); + let address_len = match afi { + 1 => 4, + 2 => 16, + _ => return None, + }; + asn_pair_len + 4 + address_len * 2 + }; + + let bgp_header = body.get(marker_offset..marker_offset + 19)?; + if bgp_header[..16] != [0xff; 16] { + return None; + } + let bgp_length = u16::from_be_bytes([bgp_header[16], bgp_header[17]]) as usize; + let bgp_type = bgp_header[18]; + if !(19..=65_535).contains(&bgp_length) + || !(1..=4).contains(&bgp_type) + || marker_offset + bgp_length != body.len() + { + return None; + } + match bgp_type { + 1 if bgp_length > 4_096 => return None, + 2 if bgp_length < 23 => return None, + 3 if bgp_length < 21 => return None, + 4 if bgp_length != 19 => return None, + _ => {} + } + + raw_record.clone().parse().ok()?; + Some(RecoveryEvidence::BgpMarkerChain) +} + +fn is_non_eof_io_error(error: &ParserError) -> bool { + matches!( + error, + ParserError::IoError(error) | ParserError::EofError(error) + if error.kind() != io::ErrorKind::UnexpectedEof + ) +} + +/// Reader adapter that tracks the absolute decompressed-stream offset and can be handed +/// back unconsumed bytes after a recovery scan read past the resume boundary. +struct CarryoverReader { + inner: R, + carryover: Vec, + carry_pos: usize, + offset: u64, +} + +impl CarryoverReader { + fn new(inner: R) -> Self { + Self { + inner, + carryover: Vec::new(), + carry_pos: 0, + offset: 0, + } + } + + fn position(&self) -> u64 { + self.offset + } + + /// Resume reading at `offset`, serving `bytes` (followed by any bytes already held + /// but not yet served) before the underlying reader. + fn resume_with(&mut self, mut bytes: Vec, offset: u64) { + bytes.extend_from_slice(&self.carryover[self.carry_pos..]); + self.carryover = bytes; + self.carry_pos = 0; + self.offset = offset; + } +} + +impl Read for CarryoverReader { + fn read(&mut self, output: &mut [u8]) -> io::Result { + if output.is_empty() { + return Ok(0); + } + if self.carry_pos < self.carryover.len() { + let count = output.len().min(self.carryover.len() - self.carry_pos); + output[..count] + .copy_from_slice(&self.carryover[self.carry_pos..self.carry_pos + count]); + self.carry_pos += count; + self.offset += count as u64; + if self.carry_pos == self.carryover.len() { + self.carryover.clear(); + self.carry_pos = 0; + } + return Ok(count); + } + let read = self.inner.read(output)?; + self.offset += read as u64; + Ok(read) + } +} + +/// Bounded lookahead buffer used only while scanning for a recovery boundary. +/// +/// It is seeded with the bytes already consumed by the failed record and buffers further +/// bytes on demand so candidate boundaries can be revisited. It exists only for the +/// duration of one recovery attempt; the undamaged fast path never copies through it. +struct ReplayReader { + inner: R, + data: Vec, + cursor: usize, + base_offset: u64, +} + +impl ReplayReader { + fn seeded(inner: R, data: Vec, base_offset: u64) -> Self { + Self { + inner, + data, + cursor: 0, + base_offset, + } + } + + fn position(&self) -> u64 { + self.base_offset + self.cursor as u64 + } + + fn buffered_end(&self) -> u64 { + self.base_offset + self.data.len() as u64 + } + + /// Consume the window, returning the buffered bytes at and beyond `offset`. + fn into_leftover(mut self, offset: u64) -> Vec { + let index = (offset.saturating_sub(self.base_offset) as usize).min(self.data.len()); + self.data.split_off(index) + } +} + +impl ReplayReader { + /// Buffer through `offset` and place the cursor there. Returns false when the stream + /// ends first or `offset` precedes the window. + fn move_to(&mut self, offset: u64) -> io::Result { + if offset < self.base_offset { + return Ok(false); + } + let target = (offset - self.base_offset) as usize; + while self.data.len() < target { + let filled = self.data.len(); + let chunk = (target - filled).min(SCAN_FILL_CHUNK); + self.data.resize(filled + chunk, 0); + let read = self.inner.read(&mut self.data[filled..])?; + self.data.truncate(filled + read); + if read == 0 { + self.cursor = self.data.len(); + return Ok(false); + } + } + self.cursor = target; + Ok(true) + } + + /// Read two bytes at `offset`, buffering as needed. Returns `None` when the stream + /// ends first. Callers re-position with [`Self::move_to`] before parsing. + fn peek_two_at(&mut self, offset: u64) -> io::Result> { + if !self.move_to(offset + 2)? { + return Ok(None); + } + let index = (offset - self.base_offset) as usize; + Ok(Some([self.data[index], self.data[index + 1]])) + } +} + +impl Read for ReplayReader { + fn read(&mut self, output: &mut [u8]) -> io::Result { + if output.is_empty() { + return Ok(0); + } + if self.cursor < self.data.len() { + let count = output.len().min(self.data.len() - self.cursor); + output[..count].copy_from_slice(&self.data[self.cursor..self.cursor + count]); + self.cursor += count; + return Ok(count); + } + + let read = self.inner.read(output)?; + self.data.extend_from_slice(&output[..read]); + self.cursor += read; + Ok(read) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::models::{Asn, Bgp4MpEnum, Bgp4MpMessage, BgpMessage, CommonHeader, MrtMessage}; + use bytes::{BufMut, BytesMut}; + use std::io::Cursor; + use std::net::{IpAddr, Ipv4Addr}; + + fn legacy_state(timestamp: u32) -> Vec { + let mut bytes = BytesMut::new(); + bytes.put_u32(timestamp); + bytes.put_u16(EntryType::BGP as u16); + bytes.put_u16(3); + bytes.put_u32(10); + bytes.put_u16(64512); + bytes.put_slice(&[192, 0, 2, 1]); + bytes.put_u16(1); + bytes.put_u16(2); + bytes.to_vec() + } + + fn bgp4mp_keepalive(timestamp: u32) -> Vec { + MrtRecord { + common_header: CommonHeader { + timestamp, + microsecond_timestamp: None, + entry_type: EntryType::BGP4MP, + entry_subtype: Bgp4MpType::Message as u16, + length: 0, + }, + message: MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(Bgp4MpMessage { + msg_type: Bgp4MpType::Message, + peer_asn: Asn::new_16bit(64_496), + local_asn: Asn::new_16bit(64_497), + interface_index: 0, + peer_ip: IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), + local_ip: IpAddr::V4(Ipv4Addr::new(192, 0, 2, 2)), + bgp_message: BgpMessage::KeepAlive, + })), + } + .encode() + .unwrap() + .to_vec() + } + + /// A record with a well-formed common header and declared length, whose body cannot + /// be parsed as a BGP4MP message. + fn framed_record_with_garbage_body(timestamp: u32, body: &[u8]) -> Vec { + let mut bytes = BytesMut::new(); + bytes.put_u32(timestamp); + bytes.put_u16(EntryType::BGP4MP as u16); + bytes.put_u16(Bgp4MpType::Message as u16); + bytes.put_u32(body.len() as u32); + bytes.put_slice(body); + bytes.to_vec() + } + + #[test] + fn recovers_at_three_record_legacy_chain() { + let first = legacy_state(100); + let mut input = first.clone(); + input.extend_from_slice(&[0xde, 0xad, 0xbe, 0xef, 0x01]); + input.extend_from_slice(&legacy_state(101)); + input.extend_from_slice(&legacy_state(102)); + input.extend_from_slice(&legacy_state(103)); + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + assert_eq!(events.len(), 5); + assert!(matches!(events[0], RecoveryEvent::Item(_))); + let RecoveryEvent::Gap(gap) = &events[1] else { + panic!("expected recovery gap") + }; + assert_eq!(gap.start_offset, first.len() as u64); + assert_eq!(gap.end_offset, first.len() as u64 + 5); + assert_eq!(gap.confirmed_records, 3); + assert_eq!(gap.evidence, RecoveryEvidence::LegacyMrtChain); + assert!(events[2..] + .iter() + .all(|event| matches!(event, RecoveryEvent::Item(_)))); + } + + #[test] + fn emits_terminal_gap_when_chain_too_short_before_non_eof_garbage() { + let mut input = vec![0xff; 12]; + input.extend_from_slice(&legacy_state(101)); + input.extend_from_slice(&legacy_state(102)); + input.extend_from_slice(&[1, 2, 3]); + let total = input.len() as u64; + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + // A two-record chain is below the configured confirmation count, so nothing is + // recovered; the damage extends to the end of the stream and is reported as a + // terminal gap rather than an error. + assert_eq!(events.len(), 1); + let RecoveryEvent::Gap(gap) = &events[0] else { + panic!("expected terminal gap") + }; + assert_eq!(gap.start_offset, 0); + assert_eq!(gap.end_offset, total); + assert_eq!(gap.evidence, RecoveryEvidence::EndOfStream); + assert_eq!(gap.confirmed_records, 0); + } + + #[test] + fn emits_terminal_gap_when_chain_too_short_at_clean_eof() { + let mut input = vec![0xff; 12]; + input.extend_from_slice(&legacy_state(101)); + input.extend_from_slice(&legacy_state(102)); + let total = input.len() as u64; + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + assert_eq!(events.len(), 1); + let RecoveryEvent::Gap(gap) = &events[0] else { + panic!("expected terminal gap") + }; + assert_eq!(gap.start_offset, 0); + assert_eq!(gap.end_offset, total); + assert_eq!(gap.evidence, RecoveryEvidence::EndOfStream); + assert_eq!(gap.confirmed_records, 0); + } + + #[test] + fn truncated_final_record_yields_terminal_gap() { + let mut input = bgp4mp_keepalive(100); + input.extend_from_slice(&bgp4mp_keepalive(101)); + let boundary = input.len() as u64; + let tail = bgp4mp_keepalive(102); + input.extend_from_slice(&tail[..10]); + let total = input.len() as u64; + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + assert_eq!(events.len(), 3); + assert!(matches!(events[0], RecoveryEvent::Item(_))); + assert!(matches!(events[1], RecoveryEvent::Item(_))); + let RecoveryEvent::Gap(gap) = &events[2] else { + panic!("expected terminal gap") + }; + assert_eq!(gap.start_offset, boundary); + assert_eq!(gap.end_offset, total); + assert_eq!(gap.evidence, RecoveryEvidence::EndOfStream); + assert_eq!(gap.confirmed_records, 0); + } + + #[test] + fn skips_exactly_one_framed_record_with_unparsable_body() { + let first = bgp4mp_keepalive(100); + let bad = framed_record_with_garbage_body(101, &[0xde, 0xad, 0xbe]); + let mut input = first.clone(); + input.extend_from_slice(&bad); + input.extend_from_slice(&bgp4mp_keepalive(102)); + input.extend_from_slice(&bgp4mp_keepalive(103)); + input.extend_from_slice(&bgp4mp_keepalive(104)); + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + // The bad record framed correctly, so exactly its bytes are skipped without a + // boundary scan and all following records survive. + assert_eq!(events.len(), 5); + assert!(matches!(events[0], RecoveryEvent::Item(_))); + let RecoveryEvent::Gap(gap) = &events[1] else { + panic!("expected aligned-boundary gap") + }; + assert_eq!(gap.start_offset, first.len() as u64); + assert_eq!(gap.end_offset, (first.len() + bad.len()) as u64); + assert_eq!(gap.evidence, RecoveryEvidence::AlignedRecordChain); + assert_eq!(gap.confirmed_records, 3); + assert!(events[2..] + .iter() + .all(|event| matches!(event, RecoveryEvent::Item(_)))); + } + + #[test] + fn skips_framed_record_with_unparsable_body_at_clean_eof() { + let first = bgp4mp_keepalive(100); + let bad = framed_record_with_garbage_body(101, &[0xde, 0xad, 0xbe]); + let mut input = first.clone(); + input.extend_from_slice(&bad); + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + assert_eq!(events.len(), 2); + assert!(matches!(events[0], RecoveryEvent::Item(_))); + let RecoveryEvent::Gap(gap) = &events[1] else { + panic!("expected aligned-boundary gap") + }; + assert_eq!(gap.start_offset, first.len() as u64); + assert_eq!(gap.end_offset, (first.len() + bad.len()) as u64); + assert_eq!(gap.evidence, RecoveryEvidence::AlignedRecordChain); + assert_eq!(gap.confirmed_records, 0); + } + + #[test] + fn misframed_record_falls_back_to_boundary_scan() { + let first = bgp4mp_keepalive(100); + // A header whose declared length overlaps the next record: the declared + // boundary is misaligned, so aligned-boundary confirmation must fail and the + // byte scan must find the true boundary. + let mut bad = BytesMut::new(); + bad.put_u32(101); + bad.put_u16(EntryType::BGP4MP as u16); + bad.put_u16(Bgp4MpType::Message as u16); + bad.put_u32(5); + let mut input = first.clone(); + input.extend_from_slice(&bad); + input.extend_from_slice(&[0xde, 0xad, 0xbe]); + input.extend_from_slice(&bgp4mp_keepalive(102)); + input.extend_from_slice(&bgp4mp_keepalive(103)); + input.extend_from_slice(&bgp4mp_keepalive(104)); + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + assert_eq!(events.len(), 5); + let RecoveryEvent::Gap(gap) = &events[1] else { + panic!("expected recovery gap") + }; + assert_eq!(gap.start_offset, first.len() as u64); + assert_eq!(gap.end_offset, (first.len() + bad.len() + 3) as u64); + assert_eq!(gap.evidence, RecoveryEvidence::BgpMarkerChain); + assert_eq!(gap.confirmed_records, 3); + assert!(events[2..] + .iter() + .all(|event| matches!(event, RecoveryEvent::Item(_)))); + } + + #[test] + fn text_dump_parser_yields_unsupported_error() { + let dump = "BGP table version is 1, local router ID is 1.2.3.4, vrf id 0\n\ +Default local pref 100, local AS 65001\n\n\ + Network Next Hop Metric LocPrf Weight Path\n\ + *> 1.0.0.0/24 10.0.0.1 0 0 13335 i\n"; + let parser = BgpkitParser::from_text_reader(dump.as_bytes()).unwrap(); + let mut iter = parser.into_recovering_record_iter(RecoveryConfig::default()); + + let error = iter.next().expect("one error").unwrap_err(); + assert!(matches!(error.error.error, ParserError::Unsupported(_))); + assert!(iter.next().is_none()); + } + + #[test] + fn recovering_elem_iter_passes_gaps_through() { + let first = legacy_state(100); + let mut input = first.clone(); + input.extend_from_slice(&[0xde, 0xad, 0xbe, 0xef, 0x01]); + input.extend_from_slice(&legacy_state(101)); + input.extend_from_slice(&legacy_state(102)); + input.extend_from_slice(&legacy_state(103)); + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_elem_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + // State-change records yield no elements, so only the gap surfaces. + assert_eq!(events.len(), 1); + let RecoveryEvent::Gap(gap) = &events[0] else { + panic!("expected recovery gap") + }; + assert_eq!(gap.start_offset, first.len() as u64); + assert_eq!(gap.end_offset, first.len() as u64 + 5); + } + + #[test] + fn skipped_bytes_is_zero_for_inverted_range() { + let gap = RecoveryGap { + start_offset: 10, + end_offset: 5, + cause: String::new(), + evidence: RecoveryEvidence::LegacyMrtChain, + confirmed_records: 3, + }; + + assert_eq!(gap.skipped_bytes(), 0); + } + + #[test] + fn recovers_bgp4mp_using_exact_embedded_bgp_headers() { + let first = bgp4mp_keepalive(100); + let mut input = first.clone(); + input.extend_from_slice(&[0xde, 0xad, 0xbe, 0xef]); + input.extend_from_slice(&bgp4mp_keepalive(101)); + input.extend_from_slice(&bgp4mp_keepalive(102)); + input.extend_from_slice(&bgp4mp_keepalive(103)); + + let parser = BgpkitParser::from_reader(Cursor::new(input)); + let events = parser + .into_recovering_record_iter(RecoveryConfig::default()) + .collect::, _>>() + .unwrap(); + + let RecoveryEvent::Gap(gap) = &events[1] else { + panic!("expected recovery gap") + }; + assert_eq!(gap.start_offset, first.len() as u64); + assert_eq!(gap.end_offset, first.len() as u64 + 4); + assert_eq!(gap.evidence, RecoveryEvidence::BgpMarkerChain); + assert_eq!(gap.confirmed_records, 3); + } + + #[test] + fn rejects_bgp4mp_candidate_with_inexact_embedded_length() { + let encoded = bgp4mp_keepalive(100); + let header = CommonHeader { + timestamp: 100, + microsecond_timestamp: None, + entry_type: EntryType::BGP4MP, + entry_subtype: Bgp4MpType::Message as u16, + length: (encoded.len() - 12) as u32, + }; + let mut body = encoded[12..].to_vec(); + // 16-bit ASN/IPv4 BGP4MP envelope is 16 bytes; corrupt the BGP length field. + body[32..34].copy_from_slice(&20u16.to_be_bytes()); + let raw = crate::RawMrtRecord { + common_header: header, + header_bytes: Bytes::copy_from_slice(&encoded[..12]), + message_bytes: Bytes::from(body), + }; + assert!(strict_bgp4mp_evidence(&raw).is_none()); + } +} diff --git a/src/parser/mod.rs b/src/parser/mod.rs index e486c81..bc8c44e 100644 --- a/src/parser/mod.rs +++ b/src/parser/mod.rs @@ -63,6 +63,17 @@ impl Default for ParserOptions { } } +impl ParserOptions { + pub(crate) fn warn_zebra_compat_once(&mut self) { + if self.show_warnings && !self.warned_zebra_compat { + warn!( + "recovered shortened Zebra BGP4MP records with missing envelope fields; substituting IPv4 zero addresses and interface index 0 (further occurrences for this parser will not be logged)" + ); + self.warned_zebra_compat = true; + } + } +} + #[cfg(feature = "oneio")] impl BgpkitParser> { /// Creating a new parser from a object that implements [Read] trait. @@ -296,12 +307,7 @@ impl BgpkitParser> { impl BgpkitParser { pub(crate) fn warn_zebra_compat_once(&mut self) { - if self.options.show_warnings && !self.options.warned_zebra_compat { - warn!( - "recovered shortened Zebra BGP4MP records with missing envelope fields; substituting IPv4 zero addresses and interface index 0 (further occurrences for this parser will not be logged)" - ); - self.options.warned_zebra_compat = true; - } + self.options.warn_zebra_compat_once(); } pub fn enable_core_dump(self) -> Self { diff --git a/tests/fixtures/ripe/README.md b/tests/fixtures/ripe/README.md index ad4371c..425b21e 100644 --- a/tests/fixtures/ripe/README.md +++ b/tests/fixtures/ripe/README.md @@ -10,6 +10,7 @@ must not download data. | `rrc00/1999.12/updates.19991214.1621.gz` | | 305,644 bytes | `8e4b0ed378464f68397d60e2775a310117b23ffe6dc7f79fbb0d5039a743c91e` | | `rrc00/2000.01/updates.20000102.2014.gz` | | 32,455 bytes | `e15119ada15bed7b524f9cef91a2ae1002f28e65360b32a8f40531d76fdc0a5f` | | `rrc00/2000.01/bview.20000111.0032.gz` | | 2,932,008 bytes | `1456fd58551374c6222c3f2f88606bac327ffdb3bdd931dd150fabae68800009` | +| `rrc00/2000.03/updates.20000325.0345.gz` | | 28,128 bytes | `f5acf0a6d5bc2610c05350c75b2eaf4e3bc100701836e6cf004fbcf635f3fc78` | | `rrc01/2000.11/updates.20001104.0124.gz` | | 8,815 bytes | `89985919e4a1f8726b1e1a6391b12ce1c5576f9d066d84e729414d922e4052f1` | | `rrc03/2010.02/updates.20100227.1600.first-796-records.gz` | | 12,953 bytes | `cca59f359818c3dacc18014e50ce3b656cbc4d86be97cbfbdd6771047a2e71fd` | | `rrc15/2010.02/updates.20100227.1600.gz` | | 213,264 bytes | `68e03340030f3c44a9eb3dd22baa7739f910466cd82ee844fc8c0147f86b7028` | @@ -17,7 +18,8 @@ must not download data. The September and December 1999 and January 2000 update files use deprecated MRT Type 5 BGP records. The December fixture includes one BGP OPEN and one BGP -NOTIFY record. The bview uses early TABLE_DUMP records that batch many entries +NOTIFY record. The March 2000 fixture contains two damaged MRT boundaries and +is retained to test opt-in framing recovery. The bview uses early TABLE_DUMP records that batch many entries and declare each physical record four bytes shorter than the bytes written by the historical MRT producer. The rrc01 update file contains shortened BGP4MP state-change and OPEN records diff --git a/tests/fixtures/ripe/rrc00/2000.03/updates.20000325.0345.gz b/tests/fixtures/ripe/rrc00/2000.03/updates.20000325.0345.gz new file mode 100644 index 0000000..bd34ea0 Binary files /dev/null and b/tests/fixtures/ripe/rrc00/2000.03/updates.20000325.0345.gz differ diff --git a/tests/recovery.rs b/tests/recovery.rs new file mode 100644 index 0000000..1d92e60 --- /dev/null +++ b/tests/recovery.rs @@ -0,0 +1,91 @@ +use bgpkit_parser::{BgpkitParser, RecoveryConfig, RecoveryEvent, RecoveryEvidence, RecoveryGap}; +use flate2::read::GzDecoder; +use std::io::{Cursor, Read}; + +const FIXTURE: &str = "tests/fixtures/ripe/rrc00/2000.03/updates.20000325.0345.gz"; + +fn fixture_path() -> String { + format!("{}/{}", env!("CARGO_MANIFEST_DIR"), FIXTURE) +} + +#[test] +fn recovers_damaged_ripe_type_5_fixture() { + let mut records = 0usize; + let mut gaps = Vec::::new(); + + for event in BgpkitParser::new(&fixture_path()) + .unwrap() + .into_recovering_record_iter(RecoveryConfig::default()) + { + match event.unwrap() { + RecoveryEvent::Item(_) => records += 1, + RecoveryEvent::Gap(gap) => gaps.push(gap), + } + } + + assert_eq!(records, 2_580); + assert_eq!(gaps.len(), 2); + assert_eq!(gaps.iter().map(RecoveryGap::skipped_bytes).sum::(), 24); + assert_eq!( + gaps.iter() + .map(|gap| ( + gap.start_offset, + gap.end_offset, + gap.evidence, + gap.confirmed_records, + )) + .collect::>(), + vec![ + (188, 198, RecoveryEvidence::LegacyMrtChain, 3), + (4_554, 4_568, RecoveryEvidence::LegacyMrtChain, 3), + ] + ); +} + +#[test] +fn recovering_elem_iter_matches_fixture_totals() { + let mut elements = 0usize; + let mut gaps = 0usize; + + for event in BgpkitParser::new(&fixture_path()) + .unwrap() + .into_recovering_elem_iter(RecoveryConfig::default()) + { + match event.unwrap() { + RecoveryEvent::Item(_) => elements += 1, + RecoveryEvent::Gap(_) => gaps += 1, + } + } + + assert_eq!(elements, 5_378); + assert_eq!(gaps, 2); +} + +/// A truncated final record — the most common real-world corruption — must yield every +/// intact record plus a terminal gap, not a hard error that discards the whole file. +#[test] +fn truncated_fixture_tail_yields_terminal_gap() { + let mut bytes = Vec::new(); + GzDecoder::new(std::fs::File::open(fixture_path()).unwrap()) + .read_to_end(&mut bytes) + .unwrap(); + bytes.truncate(bytes.len() - 20); + let total = bytes.len() as u64; + + let mut records = 0usize; + let mut gaps = Vec::::new(); + for event in BgpkitParser::from_reader(Cursor::new(bytes)) + .into_recovering_record_iter(RecoveryConfig::default()) + { + match event.unwrap() { + RecoveryEvent::Item(_) => records += 1, + RecoveryEvent::Gap(gap) => gaps.push(gap), + } + } + + assert_eq!(records, 2_579); + assert_eq!(gaps.len(), 3); + let terminal = gaps.last().unwrap(); + assert_eq!(terminal.evidence, RecoveryEvidence::EndOfStream); + assert_eq!(terminal.end_offset, total); +} diff --git a/tests/torn_mrt_records.rs b/tests/torn_mrt_records.rs new file mode 100644 index 0000000..d474286 --- /dev/null +++ b/tests/torn_mrt_records.rs @@ -0,0 +1,270 @@ +//! Regression coverage for torn MRT writes in the early RIPE RIS archive. +//! +//! `rrc00/2000.03/updates.20000325.0345.gz` is physically corrupt: the historical +//! collector emitted two short writes into an otherwise well-formed stream. The +//! damage is confined to the first ~4.6 KiB of the 151,804-byte uncompressed body, +//! but the parser cannot resynchronize past either one, so it recovers 4 of the +//! file's 2,580 records. +//! +//! **Offset 188 — a 10-byte truncated record header:** +//! +//! ```text +//! 38 dc 36 68 00 05 00 01 00 00 | 38 dc 36 71 00 05 00 03 ... +//! ^timestamp ^BGP ^UPDATE ^only 2 of the 4 Length bytes +//! 953955944 (5) (1) ^next record starts here +//! ``` +//! +//! The writer stopped after 10 of the 12 header bytes. The parser therefore reads +//! `Length` as the four bytes `00 00 38 dc`, whose low half is stolen from the next +//! record's timestamp (`0x38dc3671`), giving 14,556. It consumes 14,556 bytes — +//! roughly 200 real records — and fails with [`ParserError::TruncatedMsg`], leaving +//! the reader at offset 14,756, which is not a record boundary. +//! +//! **Offset 4554 — 14 orphan body bytes:** +//! +//! ```text +//! cb 2d 13 | 40 03 04 cb 25 ff 7e | 18 d1 f7 aa +//! ^attr tail ^NEXT_HOP 203.37.255.126 ^NLRI 209.247.170.0/24 +//! ``` +//! +//! The tail of a BGP UPDATE body left over after a complete STATE_CHANGE record. +//! Timestamps run backwards across this point (953956818 before, 953955969 after), +//! consistent with interleaved buffer flushes. +//! +//! Deleting those 24 bytes makes the file parse cleanly — 2,580 records (2,239 +//! UPDATE, 307 STATE_CHANGE, 34 KEEPALIVE), 5,378 elements, zero errors — so there +//! is no body-level parsing defect here. What is missing is stream +//! resynchronization: on a framing error the reader has no pushback and cannot scan +//! forward for the next plausible common header, so every iterator either retries +//! at the same wrong offset or stops. +//! +//! The assertions below therefore pin **current, lossy** behaviour rather than +//! desired behaviour. Every count is a deterministic function of the fixture bytes. +//! If a change adds resynchronization, these numbers are expected to rise to the +//! recoverable totals quoted above. +//! +//! The same signature appears in every neighbouring rrc00 2000.03 file — exactly two +//! torn writes, one in the first ~200 bytes and one just past 4 KiB — so this is a +//! property of the historical writer, not of this one file. + +use bgpkit_parser::models::EntryType; +use bgpkit_parser::{BgpkitParser, DiagnosticEvent, MrtUpdate, ParserError}; +use std::collections::BTreeMap; + +const FIXTURE: &str = "tests/fixtures/ripe/rrc00/2000.03/updates.20000325.0345.gz"; + +/// Records ahead of the first torn write, all of which parse normally. The fixture +/// holds 2,580 recoverable records and 5,378 recoverable elements, so all but these +/// four are lost. +const RECORDS_BEFORE_FIRST_TORN_WRITE: usize = 4; +/// Errors produced while the parser hunts forward 12 bytes at a time after +/// desynchronizing. It never re-aligns, because the true record boundary is not a +/// multiple of 12 from where the oversized read left it. +const DESYNCHRONIZED_ERRORS: usize = 128; + +/// Header the parser manufactures from the 10-byte torn write at offset 188. +const TORN_HEADER_TIMESTAMP: u32 = 953_955_944; +const TORN_HEADER_LENGTH: u32 = 14_556; + +fn repo_fixture(path: &str) -> String { + format!("{}/{path}", env!("CARGO_MANIFEST_DIR")) +} + +#[test] +fn torn_writes_desynchronize_raw_record_framing() { + let source = repo_fixture(FIXTURE); + let mut raw_records = 0usize; + let mut headers = BTreeMap::<(u16, u16), usize>::new(); + + for raw_record in BgpkitParser::new(&source).unwrap().into_raw_record_iter() { + raw_records += 1; + *headers + .entry(( + raw_record.common_header.entry_type as u16, + raw_record.common_header.entry_subtype, + )) + .or_default() += 1; + } + + // Only 5 of the fixture's 2,580 records reach framing — 4 genuine ones plus the + // 14,556-byte torn "record" — after which the parser frames 3 more headers out of + // misaligned bytes. + assert_eq!(raw_records, 8); + assert_eq!( + headers, + BTreeMap::from([ + ( + (EntryType::NULL as u16, 256), + 3 // pure garbage, framed mid-record after desynchronizing + ), + ( + (EntryType::BGP as u16, 1), + 4 // 3 genuine UPDATEs plus the torn 14,556-byte "record" + ), + ((EntryType::BGP as u16, 7), 1), + ]) + ); +} + +#[test] +fn torn_writes_lose_all_but_four_records() { + let source = repo_fixture(FIXTURE); + + // Skipping iterators: the error arms can only retry at the same wrong offset or + // stop, so 2,576 of 2,580 records and 5,374 of 5,378 elements are lost. + assert_eq!( + BgpkitParser::new(&source) + .unwrap() + .into_record_iter() + .count(), + RECORDS_BEFORE_FIRST_TORN_WRITE + ); + assert_eq!( + BgpkitParser::new(&source).unwrap().into_elem_iter().count(), + RECORDS_BEFORE_FIRST_TORN_WRITE + ); + assert_eq!( + BgpkitParser::new(&source) + .unwrap() + .into_route_iter() + .count(), + RECORDS_BEFORE_FIRST_TORN_WRITE + ); + // One of the four surviving records is a KEEPALIVE, so the update iterator sees + // one fewer. + assert_eq!( + BgpkitParser::new(&source) + .unwrap() + .into_update_iter() + .count(), + RECORDS_BEFORE_FIRST_TORN_WRITE - 1 + ); + + // Fallible iterators surface the same loss as errors instead of silence. + let mut records = 0usize; + let mut record_errors = 0usize; + for result in BgpkitParser::new(&source) + .unwrap() + .into_fallible_record_iter() + { + match result { + Ok(_) => records += 1, + Err(_) => record_errors += 1, + } + } + assert_eq!(records, RECORDS_BEFORE_FIRST_TORN_WRITE); + assert_eq!(record_errors, DESYNCHRONIZED_ERRORS); + + let mut elements = 0usize; + let mut element_errors = 0usize; + for result in BgpkitParser::new(&source) + .unwrap() + .into_fallible_elem_iter() + { + match result { + Ok(_) => elements += 1, + Err(_) => element_errors += 1, + } + } + assert_eq!(elements, RECORDS_BEFORE_FIRST_TORN_WRITE); + assert_eq!(element_errors, DESYNCHRONIZED_ERRORS); + + let mut updates = 0usize; + let mut update_errors = 0usize; + for result in BgpkitParser::new(&source) + .unwrap() + .into_fallible_update_iter() + { + match result { + Ok(MrtUpdate::LegacyBgpUpdate(_)) => updates += 1, + Ok(update) => panic!("unexpected MRT update: {update:?}"), + Err(_) => update_errors += 1, + } + } + assert_eq!(updates, RECORDS_BEFORE_FIRST_TORN_WRITE - 1); + assert_eq!(update_errors, DESYNCHRONIZED_ERRORS); + + let mut routes = 0usize; + let mut route_errors = 0usize; + for result in BgpkitParser::new(&source) + .unwrap() + .into_fallible_route_iter() + { + match result { + Ok(_) => routes += 1, + Err(_) => route_errors += 1, + } + } + assert_eq!(routes, RECORDS_BEFORE_FIRST_TORN_WRITE); + assert_eq!(route_errors, DESYNCHRONIZED_ERRORS); +} + +#[test] +fn diagnostic_iterator_reports_both_torn_writes() { + let source = repo_fixture(FIXTURE); + let events: Vec = BgpkitParser::new(&source) + .unwrap() + .into_diagnostic_iter() + .collect(); + + // Four clean records, then one body error, then one framing error which + // terminates the iterator by design. + assert_eq!(events.len(), RECORDS_BEFORE_FIRST_TORN_WRITE + 2); + for event in &events[..RECORDS_BEFORE_FIRST_TORN_WRITE] { + let DiagnosticEvent::Record(record) = event else { + panic!("expected a clean record, got {event:?}"); + }; + assert_eq!(record.common_header.entry_type, EntryType::BGP); + } + + // The 14,556-byte body read fails, but the manufactured header itself framed, so + // the iterator reports it and continues. + let DiagnosticEvent::ParseError { + error, + common_header, + raw_bytes, + } = &events[RECORDS_BEFORE_FIRST_TORN_WRITE] + else { + panic!( + "expected a parse error for the torn header, got {:?}", + events[RECORDS_BEFORE_FIRST_TORN_WRITE] + ); + }; + let ParserError::TruncatedMsg(message) = error else { + panic!("expected TruncatedMsg, got {error:?}"); + }; + assert_eq!( + message, + "not enough bytes to read. remaining: 14542, required: 33274" + ); + let header = common_header.expect("the torn header parsed, so it is retained"); + assert_eq!(header.timestamp, TORN_HEADER_TIMESTAMP); + assert_eq!(header.entry_type, EntryType::BGP); + assert_eq!(header.entry_subtype, 1); + assert_eq!(header.length, TORN_HEADER_LENGTH); + assert_eq!( + raw_bytes.as_deref().map(<[u8]>::len), + Some(12 + TORN_HEADER_LENGTH as usize) + ); + + // The oversized read left the stream mid-record, so the next 12 bytes are not a + // header. There is no header to report, and the framing error terminates + // iteration rather than reinterpreting the remaining bytes. + let DiagnosticEvent::ParseError { + error, + common_header, + raw_bytes, + } = &events[RECORDS_BEFORE_FIRST_TORN_WRITE + 1] + else { + panic!( + "expected a framing error after desynchronizing, got {:?}", + events[RECORDS_BEFORE_FIRST_TORN_WRITE + 1] + ); + }; + let ParserError::ParseError(message) = error else { + panic!("expected ParseError, got {error:?}"); + }; + assert_eq!(message, "cannot parse entry type: 32305"); + assert!(common_header.is_none()); + assert_eq!(raw_bytes.as_deref().map(<[u8]>::len), Some(12)); +}