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")); +}