fix: correct NetFlow v9 Count and framing (#302)

Author: ktsaou

README.md

@@ -224,6 +224,26 @@ let parser = NetflowParser::builder()
 - Each parser instance maintains its own template cache
 - For multi-source deployments, use `RouterScopedParser` (see Template Management section)
 
+### NetFlow v9 Frame Boundaries
+
+NetFlow v9 does not carry a packet-length field. Its header `Count` is the
+number of Template, Options Template, and Data records, not the number of
+FlowSets and not a byte length. Pass exactly one complete, transport-delimited
+v9 export packet to each `parse_bytes` or `iter_packets` call.
+
+The default maximum v9 frame size is 65,535 bytes, including the 20-byte
+header. Callers using a transport that permits larger frames can configure a
+higher finite bound:
+
+```rust
+use netflow_parser::NetflowParser;
+
+let parser = NetflowParser::builder()
+    .with_v9_max_frame_size_bytes(128 * 1024)
+    .build()
+    .expect("valid frame limit");
+```
+
 ### Maximum Field Count (Security)
 
 Configure the maximum number of fields allowed per template to prevent DoS attacks via malicious packets with excessive field counts:

SECURITY.md

@@ -159,6 +159,7 @@ The parser includes several DoS mitigations:
 - **Template Field Count Limit:** Default 10,000 fields per template
 - **Template Total Size Validation:** Maximum 65,535 bytes per template
 - **Cumulative Decoded Output:** Defaults to 65,536 field values and 4 MiB of field content per message
+- **NetFlow v9 Frame Size Limit:** Default 65,535 bytes per caller-delimited packet
 - **Error Sample Size Limit:** Default 256 bytes to prevent memory exhaustion
 - **LRU Template Cache:** Prevents unbounded cache growth
 

src/lib.rs

@@ -61,7 +61,8 @@ pub use variable_versions::ttl::TtlConfig;
 pub use variable_versions::{
     Config, ConfigError, DEFAULT_MAX_DECODED_FIELD_PAYLOAD_BYTES_PER_MESSAGE,
     DEFAULT_MAX_DECODED_FIELD_VALUES_PER_MESSAGE, DEFAULT_MAX_RECORDS_PER_FLOWSET,
-    DecodedOutputLimit, DecodedOutputLimits, NoTemplateInfo, PendingFlowsConfig,
+    DEFAULT_MAX_V9_FRAME_SIZE_BYTES, DecodedOutputLimit, DecodedOutputLimits, NoTemplateInfo,
+    PendingFlowsConfig,
 };
 
 // Rust-idiomatic naming aliases
@@ -283,6 +284,7 @@ pub struct NetflowParserBuilder {
     /// Raw version numbers passed to `with_allowed_versions`, for validation
     requested_versions: Option<Vec<u16>>,
     max_error_sample_size: usize,
+    v9_max_frame_size_bytes: usize,
     template_hooks: TemplateHooks,
     template_store: Option<Arc<dyn TemplateStore>>,
     template_store_scope: Arc<str>,
@@ -312,6 +314,7 @@ impl std::fmt::Debug for NetflowParserBuilder {
             .field("ipfix_config", &self.ipfix_config)
             .field("allowed_versions", &self.allowed_versions)
             .field("max_error_sample_size", &self.max_error_sample_size)
+            .field("v9_max_frame_size_bytes", &self.v9_max_frame_size_bytes)
             .field(
                 "template_hooks",
                 &format!("{} hooks", self.template_hooks.len()),
@@ -337,6 +340,7 @@ impl Default for NetflowParserBuilder {
             allowed_versions: versions_to_array(&[5, 7, 9, 10]),
             requested_versions: None,
             max_error_sample_size: 256,
+            v9_max_frame_size_bytes: DEFAULT_MAX_V9_FRAME_SIZE_BYTES,
             template_hooks: TemplateHooks::new(),
             template_store: None,
             template_store_scope: Arc::from(""),
@@ -641,6 +645,16 @@ impl NetflowParserBuilder {
         self
     }
 
+    /// Sets the maximum size of one caller-delimited NetFlow v9 export packet.
+    ///
+    /// The size includes the complete 20-byte v9 header. The default is 65,535
+    /// bytes. Callers using a transport that permits larger frames may raise it.
+    #[must_use = "builder methods consume self and return a new builder; the return value must be used"]
+    pub fn with_v9_max_frame_size_bytes(mut self, size: usize) -> Self {
+        self.v9_max_frame_size_bytes = size;
+        self
+    }
+
     /// Registers a custom enterprise field definition for both V9 and IPFIX parsers.
     ///
     /// This allows library users to define their own enterprise-specific fields without
@@ -926,6 +940,9 @@ impl NetflowParserBuilder {
     pub fn validate(&self) -> Result<(), ConfigError> {
         V9Parser::validate_config(&self.v9_config)?;
         IPFixParser::validate_config(&self.ipfix_config)?;
+        if self.v9_max_frame_size_bytes == 0 {
+            return Err(ConfigError::InvalidV9FrameSize(0));
+        }
         // Check that all requested versions are supported (5, 7, 9, 10)
         if let Some(versions) = &self.requested_versions {
             if versions.is_empty() {
@@ -961,13 +978,15 @@ impl NetflowParserBuilder {
     /// ```
     pub fn build(self) -> Result<NetflowParser, ConfigError> {
         self.validate()?;
+        let v9_max_frame_size_bytes = self.v9_max_frame_size_bytes;
         let mut v9_config = self.v9_config;
         let mut ipfix_config = self.ipfix_config;
         v9_config.template_store = self.template_store.clone();
         v9_config.template_store_scope = Arc::clone(&self.template_store_scope);
         ipfix_config.template_store = self.template_store;
         ipfix_config.template_store_scope = self.template_store_scope;
-        let v9_parser = V9Parser::try_new(v9_config)?;
+        let mut v9_parser = V9Parser::try_new(v9_config)?;
+        v9_parser.set_max_frame_size_bytes(v9_max_frame_size_bytes)?;
         let ipfix_parser = IPFixParser::try_new(ipfix_config)?;
 
         Ok(NetflowParser {
@@ -1679,6 +1698,10 @@ impl NetflowParser {
     /// * `packets` - All successfully parsed packets (even if error occurred)
     /// * `error` - `None` if fully successful, `Some(error)` if parsing stopped
     ///
+    /// NetFlow v9 has no packet-length field. Pass exactly one complete,
+    /// transport-delimited v9 export packet per call. Concatenated v9 packets
+    /// cannot be separated from FlowSets and are not supported.
+    ///
     /// # Examples
     ///
     /// ## Basic usage
@@ -1750,6 +1773,8 @@ impl NetflowParser {
 
     /// Returns an iterator that yields NetflowPacket items without allocating a Vec.
     /// This is useful for processing large batches of packets without collecting all results in memory.
+    /// NetFlow v9 input must still contain exactly one transport-delimited
+    /// export packet because v9 has no packet-length field.
     ///
     /// # Examples
     ///

src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test.snap

@@ -5,7 +5,7 @@ expression: "parser.parse_bytes(&hex::decode(hex_hex1).unwrap()).packets"
 - V9:
     header:
       version: 9
-      count: 2
+      count: 1
       sys_up_time: 6
       unix_secs: 1672687345
       sequence_number: 2
@@ -36,9 +36,3 @@ expression: "parser.parse_bytes(&hex::decode(hex_hex1).unwrap()).packets"
                   - Ip4Addr: 0.0.0.1
                 - - Ipv4DstAddr
                   - Ip4Addr: 0.0.0.1
-      - header:
-          flowset_id: 256
-          length: 8
-        body:
-          Data:
-            fields: []

src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test_2.snap

@@ -5,7 +5,7 @@ expression: "parser.parse_bytes(&hex::decode(hex2).unwrap()).packets"
 - V9:
     header:
       version: 9
-      count: 2
+      count: 1
       sys_up_time: 6
       unix_secs: 1672687345
       sequence_number: 2
@@ -36,9 +36,3 @@ expression: "parser.parse_bytes(&hex::decode(hex2).unwrap()).packets"
                   - Ip4Addr: 0.0.0.1
                 - - Ipv4DstAddr
                   - Ip4Addr: 0.0.0.1
-      - header:
-          flowset_id: 256
-          length: 8
-        body:
-          Data:
-            fields: []

src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_multiple_packets.snap

@@ -1,6 +1,6 @@
 ---
 source: src/tests.rs
-expression: "NetflowParser::default().parse_bytes(&all).packets"
+expression: packets
 ---
 - V9:
     header:

src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_v9_ipv6flowlabel.snap

@@ -1,6 +1,6 @@
 ---
 source: src/tests.rs
-expression: "NetflowParser::default().parse_bytes(&packets).packets"
+expression: packets
 ---
 - V9:
     header:

src/tests.rs

@@ -122,33 +122,36 @@ mod base_tests {
         }));
     }
 
-    // Verify that combined v9 options template, data template, and data records parse together
+    // Verify that sequential v9 options template, data template, and data datagrams parse together
     #[test]
     fn can_read_v9_with_options_template_and_template() {
-        // Three v9 messages in one buffer: options template, data template,
+        // Three separately framed v9 messages: options template, data template,
         // and one data record using template 256.
         let hex_hex0 =
             "00090001000000000000000000000000000000010001001401000004000400010004002900020000";
         let hex_hex1 = "00090001000000000000000000000000000000010000000c0100000100080004";
         let hex_hex2 = "000900010000000000000000000000000000000101000008c0a80001";
 
-        let combined = format!("{}{}{}", hex_hex0, hex_hex1, hex_hex2);
-
         let mut parser = NetflowParser::builder()
             .with_cache_size(100)
             .build()
             .unwrap();
 
-        let packets = hex::decode(combined).unwrap();
-        let results = parser.parse_bytes(&packets).packets;
+        let mut results = Vec::new();
+        for packet_hex in [hex_hex0, hex_hex1, hex_hex2] {
+            let parsed = parser.parse_bytes(&hex::decode(packet_hex).unwrap());
+            assert!(parsed.error.is_none());
+            results.extend(parsed.packets);
+        }
         assert_yaml_snapshot!(results);
     }
 
     // Verify that a v9 template packet is parsed and can be serialized back to bytes
     #[test]
     fn can_read_v9() {
-        // Template
-        let hex = "0009000100000e1061db09bd000000010000000100000028010000080001000400020004000a00040004000400080004000c0004000700020015000400050001000600010016000400100004";
+        // One 40-byte Template FlowSet. An older fixture included 16 bytes
+        // beyond its declared FlowSet length; those bytes are not part of it.
+        let hex = "0009000100000e1061db09bd000000010000000100000028010000080001000400020004000a00040004000400080004000c00040007000200150004";
 
         let mut parser = NetflowParser::builder()
             .with_cache_size(100)
@@ -349,10 +352,11 @@ mod base_tests {
         }
     }
 
-    // Verify that v9 options template followed by a zeroed-out data record parses correctly
+    // Verify that a malformed v9 options template is not cached and its subsequent
+    // data record is retained as NoTemplate without relying on bytes outside either FlowSet.
     #[test]
     fn options_no_data() {
-        let hex = "0009000100000001639073f3000000010000000100010034010200210001000400020004000e000400160004001500040009000100070002001000040011000400180004000600010005000100b0000200b1000200b2000200b4000200b7000200b8000200ad000200ac00010038000200b9000200bd000200be000200c1000200c2000200c5000200c3000200c4000200c6000200c7000200c8000200c9000200ca000200cb000200ce000200";
+        let hex = "0009000100000001639073f3000000010000000100010034010200210001000400020004000e00040016000400150004000900010007000200100004001100040018000400060001";
 
         let mut parser = NetflowParser::builder()
             .with_cache_size(100)
@@ -362,7 +366,7 @@ mod base_tests {
         let packet = hex::decode(hex).unwrap();
         let _ = parser.parse_bytes(&packet);
 
-        let hex_data = "0009000100000002639073f300000002000000010102008400000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000";
+        let hex_data = "0009000100000002639073f30000000200000001010200840000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000";
 
         let packet = hex::decode(hex_data).unwrap();
         assert_yaml_snapshot!(parser.parse_bytes(&packet).packets);
@@ -395,9 +399,11 @@ mod base_tests {
     // Verify that v9 template followed by data in separate packets produces correct output
     #[test]
     fn v9_example_from_integration_test() {
-        let hex_template = "000900020000000563b32ef1000000010000000100000024010000060001000400020004000a00040004000400080004000c000400010038010000020001000400020004";
+        // Each packet ends at its declared FlowSet boundary. Older fixtures
+        // appended malformed bytes that the Count-as-FlowSet bug hid.
+        let hex_template = "000900010000000563b32ef1000000010000000100000024010000060001000400020004000a00040004000400080004000c000400010038";
 
-        let hex_hex1 = "000900020000000663b32ef10000000200000001010000240a70090a0a70090b00000001000000010000000100000001000000030000000601000008192a80e3192a80e4";
+        let hex_hex1 = "000900010000000663b32ef10000000200000001010000240a70090a0a70090b000000010000000100000001000000010000000300000006";
 
         let _hex_hex2 = "000900020000000763b32ef10000000300000001010000240a700a050a700a0600000001000000010000000100000001000000030000000601000008192a80e3192a80e4";
 
@@ -413,9 +419,10 @@ mod base_tests {
     // Verify that v9 multi-template data parsing works across sequential packets
     #[test]
     fn v9_example_from_integration_test_2() {
-        let hex1 = "000900020000000563b32ef1000000010000000100000024010000060001000400020004000a00040004000400080004000c000400010038010000020001000400020004";
+        // Same corrected caller-delimited fixture pair as the integration case above.
+        let hex1 = "000900010000000563b32ef1000000010000000100000024010000060001000400020004000a00040004000400080004000c000400010038";
 
-        let hex2 = "000900020000000663b32ef10000000200000001010000240a70090a0a70090b00000001000000010000000100000001000000030000000601000008192a80e3192a80e4";
+        let hex2 = "000900010000000663b32ef10000000200000001010000240a70090a0a70090b000000010000000100000001000000010000000300000006";
 
         let mut parser = NetflowParser::builder()
             .with_cache_size(100)
@@ -1080,9 +1087,14 @@ mod restored_legacy_tests {
     fn it_parses_v9_ipv6flowlabel() {
         let templates_hex = "0009000400a21e176658cb4600000155000000080000004c0102001100080004000c0004000f000400070002000b0002000a0002000e000200fc000400fd000400020004000100040016000400150004000400010005000101000002003d0001000000540103001300080004000c0004000f000400070002000b000200060001000a0002000e000200fc000400fd000400020004000100040016000400150004000400010005000100d1000801000002003d00010000005401050013001b0010001c0010003e001000070002000b000200060001000a0002000e000200fc000400fd00040002000400010004001600040015000400040001000500010050000601000002003d00010000005801060014001b0010001c0010003e0010001f000300070002000b000200060001000a0002000e000200fc000400fd00040002000400010004001600040015000400040001000500010050000601000002003d0001";
         let packets_hex = "0009000200a3a50e6658cbab000001640000000801020066c0a8120a8d180c0200000000c2c00035000200000000000000000000000000010000005200a31e7700a31e771100080001c0a8120a8d180c02000000009e230035000200000000000000000000000000010000005200a3197b00a3197b110008000101060063fd010008000000002a20235f1f7b9379fd00000000000000b2f208fffe2011800000000000000000000000000000000001f676e2b5003500000200000000000000000000000000010000006600a3197b00a3197b1100109027e0436d86dd01";
-        let combined = format!("{}{}", templates_hex, packets_hex);
-        let packets = hex::decode(combined).unwrap();
-        assert_yaml_snapshot!(NetflowParser::default().parse_bytes(&packets).packets);
+        let mut parser = NetflowParser::default();
+        let mut packets = Vec::new();
+        for packet_hex in [templates_hex, packets_hex] {
+            let parsed = parser.parse_bytes(&hex::decode(packet_hex).unwrap());
+            assert!(parsed.error.is_none());
+            packets.extend(parsed.packets);
+        }
+        assert_yaml_snapshot!(packets);
     }
 
     #[test]
@@ -1140,14 +1152,22 @@ mod restored_legacy_tests {
             4, 0, 12, 0, 4, 0, 2, 0, 4, 1, 0, 0, 28, 1, 2, 3, 4, 1, 2, 3, 3, 1, 2, 3, 2, 0, 2,
             0, 2, 0, 1, 2, 3, 4, 5, 6, 7,
         ];
-        let mut all = vec![];
-        all.extend_from_slice(&v9_packet);
-        all.extend_from_slice(&v5_packet);
-        all.extend_from_slice(&v7_packet);
-        all.extend_from_slice(&v9_packet);
-        all.extend_from_slice(&ipfix_packet);
-        all.extend_from_slice(&v5_packet);
-        assert_yaml_snapshot!(NetflowParser::default().parse_bytes(&all).packets);
+        let mut parser = NetflowParser::default();
+        let mut packets = parser.parse_bytes(&v9_packet).packets;
+
+        let mut legacy_batch = Vec::new();
+        legacy_batch.extend_from_slice(&v5_packet);
+        legacy_batch.extend_from_slice(&v7_packet);
+        packets.extend(parser.parse_bytes(&legacy_batch).packets);
+
+        packets.extend(parser.parse_bytes(&v9_packet).packets);
+
+        let mut sized_batch = Vec::new();
+        sized_batch.extend_from_slice(&ipfix_packet);
+        sized_batch.extend_from_slice(&v5_packet);
+        packets.extend(parser.parse_bytes(&sized_batch).packets);
+
+        assert_yaml_snapshot!(packets);
     }
 
     #[test]

src/variable_versions/config.rs

@@ -27,6 +27,9 @@ pub const MAX_FIELD_COUNT: usize = 10_000;
 /// This prevents CPU-bound DoS from maliciously large flowsets.
 pub const DEFAULT_MAX_RECORDS_PER_FLOWSET: usize = 1024;
 
+/// Default maximum size of one caller-delimited NetFlow v9 export packet.
+pub const DEFAULT_MAX_V9_FRAME_SIZE_BYTES: usize = 65_535;
+
 /// Configuration for V9 and IPFIX parsers.
 ///
 /// Controls template cache size, field limits, TTL, enterprise field definitions,
@@ -104,6 +107,8 @@ pub enum ConfigError {
     InvalidDecodedFieldValueLimit(usize),
     /// Decoded field-payload-byte message limit must be greater than 0.
     InvalidDecodedFieldPayloadByteLimit(usize),
+    /// NetFlow v9 frame size must be greater than 0
+    InvalidV9FrameSize(usize),
     /// Pending flow max_total_bytes must be >= max_entry_size_bytes
     InvalidPendingTotalBytes {
         max_total_bytes: usize,
@@ -193,6 +198,13 @@ impl std::fmt::Display for ConfigError {
                     bytes
                 )
             }
+            ConfigError::InvalidV9FrameSize(size) => {
+                write!(
+                    f,
+                    "Invalid NetFlow v9 frame size: {}. Must be greater than 0.",
+                    size
+                )
+            }
             ConfigError::EmptyAllowedVersions => {
                 write!(
                     f,

src/variable_versions/mod.rs

@@ -87,7 +87,7 @@ pub use config::ParserConfig;
 pub(crate) use config::ParserFields;
 pub use config::{
     Config, ConfigError, DEFAULT_MAX_RECORDS_PER_FLOWSET, DEFAULT_MAX_TEMPLATE_CACHE_SIZE,
-    MAX_FIELD_COUNT,
+    DEFAULT_MAX_V9_FRAME_SIZE_BYTES, MAX_FIELD_COUNT,
 };
 pub use output_budget::{
     DEFAULT_MAX_DECODED_FIELD_PAYLOAD_BYTES_PER_MESSAGE,