From e2b3c67ec34fad43798b6790268ae69ee4ab42f7 Mon Sep 17 00:00:00 2001 From: mikemiles-dev Date: Mon, 21 Sep 2026 23:40:54 -0500 Subject: [PATCH] fix: route each chained message to its own scope in AutoScopedParser parse_from_source derived the scope once, from the first message's header, then handed the entire buffer to that single child parser. A call carrying several chained IPFIX messages with different Observation Domain IDs therefore cached every domain's templates against the first domain. That is silent data corruption, not just a miscount. When two domains reuse a Template ID with different field layouts, the later definition overwrites the earlier one in the wrong cache and data records decode to the wrong fields with no error reported: a sourceIPv4Address of 192.0.2.1 decodes as packetDeltaCount 3221225985, its own bytes read as a counter. Re-derive the scope from the head of the remaining buffer on every iteration so each message reaches its own (source_addr, observation_domain_id) parser, as RFC 7011 section 8 requires. Error handling mirrors NetflowParser::parse_bytes: parsing stops at the first failure and already-parsed packets are retained. Only IPFIX was affected in practice. NetFlow v9 has no packet-length field, so a chained v9 buffer already fails to parse rather than misrouting, and RouterScopedParser is keyed entirely by the caller's key and never derives scope from packet contents. iter_packets_from_source still borrows one child parser for the iterator's whole lifetime and keeps this limitation, which is now documented on the method. Lifting it needs an iterator able to own several child parsers. Reproducer from Costa Tsaousis (#325), extended here with a regression test that pins the decoded field rather than the source count, and a test that chained messages sharing one domain still share a parser. --- RELEASES.md | 19 ++++ src/lib.rs | 10 +- src/scoped_parser.rs | 71 ++++++++++++-- tests/auto_scoped_chained_domains.rs | 135 +++++++++++++++++++++++++++ 4 files changed, 224 insertions(+), 11 deletions(-) create mode 100644 tests/auto_scoped_chained_domains.rs diff --git a/RELEASES.md b/RELEASES.md index cd9ec90..1cf836b 100644 --- a/RELEASES.md +++ b/RELEASES.md @@ -1,5 +1,24 @@ # 1.1.0 +## Fixes + +* **Route each chained message to its own scope in `AutoScopedParser`.** + `parse_from_source` derived the scope once from the first message's header + and then parsed the entire buffer with that one child parser. A single call + carrying several chained IPFIX messages with different Observation Domain IDs + therefore cached every domain's templates against the first domain. When two + domains reused a Template ID with different field layouts, data records + decoded to the wrong fields and no error was reported — a `sourceIPv4Address` + would silently decode as a `packetDeltaCount`. The scope is now re-derived + from each message in the buffer, so every message reaches its own + `(source_addr, observation_domain_id)` parser, as + [RFC 7011 section 8](https://www.rfc-editor.org/rfc/rfc7011.html#section-8) + requires. Reported by Costa Tsaousis (#325). + + `iter_packets_from_source` still borrows a single child parser for the + iterator's lifetime and retains this limitation; it is now documented. Fixing + it requires an iterator able to own several child parsers. + ## Dependencies * Bumped `etherparse` from 0.20 to 0.21 (dev-dependency, used by the pcap diff --git a/src/lib.rs b/src/lib.rs index 8c65e4d..fcfa3fd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1801,8 +1801,16 @@ impl NetflowParser { } } + /// Parse exactly one export packet, returning the unconsumed remainder. + /// + /// `AutoScopedParser` uses this to route each message in a chained buffer + /// to its own scoped child parser, which requires advancing one message at + /// a time rather than handing the whole buffer to a single parser. #[inline] - fn parse_packet_by_version<'a>(&mut self, packet: &'a [u8]) -> ParsedNetflow<'a> { + pub(crate) fn parse_packet_by_version<'a>( + &mut self, + packet: &'a [u8], + ) -> ParsedNetflow<'a> { // Snapshot metrics before parsing to detect collisions/evictions/expirations let hooks_active = !self.template_hooks.is_empty(); let v9_metrics_before = if hooks_active { diff --git a/src/scoped_parser.rs b/src/scoped_parser.rs index 40dd2f5..50e2869 100644 --- a/src/scoped_parser.rs +++ b/src/scoped_parser.rs @@ -5,7 +5,7 @@ use crate::{ ConfigError, NetflowError, NetflowPacket, NetflowParser, NetflowParserBuilder, ParseResult, - ParserCacheInfo, + ParsedNetflow, ParserCacheInfo, }; use lru::LruCache; use std::hash::Hash; @@ -816,16 +816,51 @@ impl AutoScopedParser { where F: FnMut(&SourceRemoval) -> Result<(), SourceRemovalReporterError>, { - let parser = match self.get_or_create_parser(source, data, reporter) { - Ok(p) => p, - Err(e) => { - return ParseResult { - packets: vec![], - error: Some(e), - }; + if data.is_empty() { + return ParseResult { + packets: vec![], + error: None, + }; + } + + // A single call may carry several chained export packets, and each one + // declares its own scope. Re-derive the scope from the head of the + // remaining buffer on every iteration so a message is never parsed by + // another domain's parser, which would mix its templates into the wrong + // cache and silently misdecode data records that reuse a Template ID. + let mut packets = Vec::new(); + let mut remaining = data; + let mut error = None; + + while !remaining.is_empty() { + let parser = match self.get_or_create_parser(source, remaining, reporter) { + Ok(p) => p, + Err(e) => { + error = Some(e); + break; + } + }; + + match parser.parse_packet_by_version(remaining) { + ParsedNetflow::Success { + packet, + remaining: new_remaining, + } => { + packets.push(packet); + remaining = new_remaining; + } + ParsedNetflow::UnallowedVersion { version } => { + error = Some(NetflowError::FilteredVersion { version }); + break; + } + ParsedNetflow::Error { error: e } => { + error = Some(e); + break; + } } - }; - parser.parse_bytes(data) + } + + ParseResult { packets, error } } /// Parse NetFlow data from a source using the iterator API. @@ -841,6 +876,22 @@ impl AutoScopedParser { /// # Returns /// /// An iterator over parsed NetFlow packets. + /// + /// # Chained messages and scoping + /// + /// Unlike [`parse_from_source`](Self::parse_from_source), this method + /// selects one scoped parser from the *first* message and borrows it for + /// the iterator's whole lifetime. If `data` chains several IPFIX messages + /// carrying different Observation Domain IDs, every message is parsed by + /// the first domain's parser, so templates from later domains are cached + /// against the wrong scope. When those domains reuse a Template ID with + /// different field layouts, data records decode to the wrong fields + /// without reporting an error. + /// + /// Pass one message per call, or use + /// [`parse_from_source`](Self::parse_from_source), which routes each + /// chained message to its own scope. Lifting this limitation requires an + /// iterator able to own several child parsers. pub fn iter_packets_from_source<'a>( &'a mut self, source: SocketAddr, diff --git a/tests/auto_scoped_chained_domains.rs b/tests/auto_scoped_chained_domains.rs new file mode 100644 index 0000000..630a275 --- /dev/null +++ b/tests/auto_scoped_chained_domains.rs @@ -0,0 +1,135 @@ +//! Chained export packets must each be routed to their own scoped parser. +//! +//! The reproducer in `chained_ipfix_messages_are_scoped_by_each_domain` comes +//! from Costa Tsaousis (netflow_parser#325). + +use std::net::SocketAddr; + +use netflow_parser::{AutoScopedParser, NetflowPacket}; + +/// One IPFIX message carrying a single Template Set with one field spec. +fn template_message(domain: u32, template_id: u16, ie: u16, field_len: u16) -> Vec { + let mut message = Vec::new(); + message.extend_from_slice(&10u16.to_be_bytes()); + message.extend_from_slice(&28u16.to_be_bytes()); + message.extend_from_slice(&1u32.to_be_bytes()); + message.extend_from_slice(&2u32.to_be_bytes()); + message.extend_from_slice(&domain.to_be_bytes()); + message.extend_from_slice(&2u16.to_be_bytes()); + message.extend_from_slice(&12u16.to_be_bytes()); + message.extend_from_slice(&template_id.to_be_bytes()); + message.extend_from_slice(&1u16.to_be_bytes()); + message.extend_from_slice(&ie.to_be_bytes()); + message.extend_from_slice(&field_len.to_be_bytes()); + message +} + +/// One IPFIX message carrying a single Data Set with a 4-byte record. +fn data_message(domain: u32, template_id: u16, payload: [u8; 4]) -> Vec { + let mut message = Vec::new(); + message.extend_from_slice(&10u16.to_be_bytes()); + message.extend_from_slice(&24u16.to_be_bytes()); + message.extend_from_slice(&3u32.to_be_bytes()); + message.extend_from_slice(&4u32.to_be_bytes()); + message.extend_from_slice(&domain.to_be_bytes()); + message.extend_from_slice(&template_id.to_be_bytes()); + message.extend_from_slice(&8u16.to_be_bytes()); + message.extend_from_slice(&payload); + message +} + +fn source() -> SocketAddr { + "192.0.2.1:2055".parse().unwrap() +} + +/// Render the first decoded field of the first data record as `"Name=Value"`. +fn first_field(packets: &[NetflowPacket]) -> String { + let data = packets + .iter() + .filter_map(|p| match p { + NetflowPacket::IPFix(i) => Some(i), + _ => None, + }) + .flat_map(|ipfix| ipfix.flowsets.iter()) + .find_map(|fs| match &fs.body { + netflow_parser::variable_versions::ipfix::FlowSetBody::Data(d) => Some(d), + _ => None, + }) + .expect("expected a Data FlowSet"); + + let (key, value) = data + .fields + .first() + .and_then(|record| record.first()) + .expect("expected one decoded field"); + + format!("{key:?}={value:?}") +} + +#[test] +fn chained_ipfix_messages_are_scoped_by_each_domain() { + let mut batch = template_message(1, 256, 1, 4); + batch.extend_from_slice(&template_message(2, 257, 1, 4)); + + let mut parser = AutoScopedParser::new(); + let result = parser.parse_from_source(source(), &batch); + + assert!(result.error.is_none(), "{:#?}", result.error); + assert_eq!(result.packets.len(), 2); + assert_eq!(parser.ipfix_source_count(), 2); +} + +/// Two domains reusing one Template ID with different layouts must not share a +/// template cache. Before the routing fix, the chained batch parsed without +/// error but decoded domain 1's `sourceIPv4Address` as `packetDeltaCount`. +#[test] +fn chained_domains_reusing_a_template_id_do_not_corrupt_decoding() { + // IE 8 = sourceIPv4Address, IE 2 = packetDeltaCount. Both 4 bytes wide, so + // the layouts differ only in meaning and the misdecode is silent. + let mut batch = template_message(1, 256, 8, 4); + batch.extend_from_slice(&template_message(2, 256, 2, 4)); + + let mut chained = AutoScopedParser::new(); + let setup = chained.parse_from_source(source(), &batch); + assert!(setup.error.is_none(), "{:#?}", setup.error); + assert_eq!(chained.ipfix_source_count(), 2); + + let chained_out = + chained.parse_from_source(source(), &data_message(1, 256, [192, 0, 2, 1])); + assert!(chained_out.error.is_none(), "{:#?}", chained_out.error); + + // Same templates delivered one per call: the already-correct path. + let mut separate = AutoScopedParser::new(); + let _ = separate.parse_from_source(source(), &template_message(1, 256, 8, 4)); + let _ = separate.parse_from_source(source(), &template_message(2, 256, 2, 4)); + let separate_out = + separate.parse_from_source(source(), &data_message(1, 256, [192, 0, 2, 1])); + assert!(separate_out.error.is_none(), "{:#?}", separate_out.error); + + // Chaining must not change how domain 1's record decodes. + assert_eq!( + first_field(&chained_out.packets), + first_field(&separate_out.packets), + ); + assert!( + first_field(&chained_out.packets).contains("SourceIpv4address"), + "domain 1 decoded with the wrong template: {}", + first_field(&chained_out.packets), + ); +} + +#[test] +fn chained_messages_from_one_domain_still_share_a_parser() { + let mut batch = template_message(7, 256, 8, 4); + batch.extend_from_slice(&data_message(7, 256, [192, 0, 2, 9])); + + let mut parser = AutoScopedParser::new(); + let result = parser.parse_from_source(source(), &batch); + + assert!(result.error.is_none(), "{:#?}", result.error); + assert_eq!(result.packets.len(), 2); + // The data record resolved against the template from the same batch, which + // only works if both messages shared one parser. + assert_eq!(parser.ipfix_source_count(), 1); + assert!(first_field(&result.packets).contains("SourceIpv4address")); +}