fix: route each chained message to its own scope in AutoScopedParser

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

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 {

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<AutoSourceKey>) -> 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,

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