From 48eaaa4795d00c46f03837e0733016b871643637 Mon Sep 17 00:00:00 2001 From: Costa Tsaousis Date: Mon, 20 Jul 2026 04:56:17 +0300 Subject: [PATCH] fix: correct NetFlow v9 Count and framing --- README.md | 20 ++ SECURITY.md | 1 + src/lib.rs | 29 ++- ...sts__v9_example_from_integration_test.snap | 8 +- ...s__v9_example_from_integration_test_2.snap | 8 +- ...acy_tests__it_parses_multiple_packets.snap | 2 +- ...acy_tests__it_parses_v9_ipv6flowlabel.snap | 2 +- src/tests.rs | 72 +++--- src/variable_versions/config.rs | 12 + src/variable_versions/mod.rs | 2 +- src/variable_versions/v9/parser.rs | 76 +++++-- src/variable_versions/v9/serializer.rs | 34 ++- src/variable_versions/v9/types.rs | 5 +- tests/v9_count_framing.rs | 210 ++++++++++++++++++ 14 files changed, 407 insertions(+), 74 deletions(-) create mode 100644 tests/v9_count_framing.rs diff --git a/README.md b/README.md index 6e5a31d..2315996 100644 --- a/README.md +++ b/README.md @@ -222,6 +222,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: diff --git a/SECURITY.md b/SECURITY.md index 3330c5f..a498211 100644 --- a/SECURITY.md +++ b/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 diff --git a/src/lib.rs b/src/lib.rs index 482df1d..8c65e4d 100644 --- a/src/lib.rs +++ b/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>, max_error_sample_size: usize, + v9_max_frame_size_bytes: usize, template_hooks: TemplateHooks, template_store: Option>, template_store_scope: Arc, @@ -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 { 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 /// diff --git a/src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test.snap b/src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test.snap index 0f9a020..9b8ce8c 100644 --- a/src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test.snap +++ b/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: [] diff --git a/src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test_2.snap b/src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test_2.snap index 0f340c2..9161ffd 100644 --- a/src/snapshots/netflow_parser__tests__base_tests__v9_example_from_integration_test_2.snap +++ b/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: [] diff --git a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_multiple_packets.snap b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_multiple_packets.snap index 3d913aa..7d5bf5c 100644 --- a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_multiple_packets.snap +++ b/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: diff --git a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_v9_ipv6flowlabel.snap b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_v9_ipv6flowlabel.snap index 6ecf2fb..f68fa61 100644 --- a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_v9_ipv6flowlabel.snap +++ b/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: diff --git a/src/tests.rs b/src/tests.rs index 4789120..deaf0bf 100644 --- a/src/tests.rs +++ b/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] diff --git a/src/variable_versions/config.rs b/src/variable_versions/config.rs index cb224de..d031c39 100644 --- a/src/variable_versions/config.rs +++ b/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, diff --git a/src/variable_versions/mod.rs b/src/variable_versions/mod.rs index 95b9fdc..1ebf6cc 100644 --- a/src/variable_versions/mod.rs +++ b/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, diff --git a/src/variable_versions/v9/parser.rs b/src/variable_versions/v9/parser.rs index 404b772..40e1d2b 100644 --- a/src/variable_versions/v9/parser.rs +++ b/src/variable_versions/v9/parser.rs @@ -16,7 +16,9 @@ use crate::template_store::{ TemplateKind, TemplateStore, TemplateStoreKey, decode_v9_options_template, decode_v9_template, encode_v9_options_template, encode_v9_template, }; -use crate::variable_versions::config::DEFAULT_MAX_RECORDS_PER_FLOWSET; +use crate::variable_versions::config::{ + DEFAULT_MAX_RECORDS_PER_FLOWSET, DEFAULT_MAX_V9_FRAME_SIZE_BYTES, +}; use crate::variable_versions::enterprise_registry::EnterpriseFieldRegistry; use crate::variable_versions::field_value::FieldValue; use crate::variable_versions::lazy_lru::LazyLruCache; @@ -51,6 +53,7 @@ pub struct V9Parser { pub(crate) max_error_sample_size: usize, pub(crate) max_records_per_flowset: usize, pub(crate) decoded_output_budget: DecodedOutputBudget, + max_frame_size_bytes: usize, pub(crate) metrics: CacheMetricsInner, pub(crate) pending_flows: Option, /// Optional secondary-tier template store. See [`crate::template_store`]. @@ -139,6 +142,7 @@ impl V9Parser { config.max_decoded_field_values_per_message, config.max_decoded_field_payload_bytes_per_message, ), + max_frame_size_bytes: DEFAULT_MAX_V9_FRAME_SIZE_BYTES, metrics: CacheMetricsInner::new(), pending_flows, template_store: config.template_store, @@ -154,6 +158,20 @@ impl V9Parser { self.template_store_scope = scope; } + /// Returns the maximum accepted size of one caller-delimited v9 frame. + pub fn max_frame_size_bytes(&self) -> usize { + self.max_frame_size_bytes + } + + /// Sets the maximum accepted size of one caller-delimited v9 frame. + pub fn set_max_frame_size_bytes(&mut self, size: usize) -> Result<(), ConfigError> { + if size == 0 { + return Err(ConfigError::InvalidV9FrameSize(size)); + } + self.max_frame_size_bytes = size; + Ok(()) + } + /// Write-through: persist a freshly learned data template. No-op when /// no store is configured. Backend failures are recorded in metrics /// but do not abort packet parsing. @@ -426,8 +444,34 @@ impl ParserConfig for V9Parser { } impl V9Parser { - /// Parse a NetFlow V9 packet from raw bytes, using cached templates to decode data records. + /// Parse a NetFlow v9 packet after the version field has been consumed. pub(crate) fn parse<'a>(&mut self, packet: &'a [u8]) -> ParsedNetflow<'a> { + // Restore the two-byte version field consumed by the version router so + // the configured limit applies to the complete wire frame. + let Some(frame_size) = packet.len().checked_add(2) else { + return ParsedNetflow::Error { + error: NetflowError::Partial { + message: "V9 frame size overflow".to_string(), + }, + }; + }; + if frame_size > self.max_frame_size_bytes { + return ParsedNetflow::Error { + error: NetflowError::Partial { + message: format!( + "V9 frame size {} exceeds configured maximum {}", + frame_size, self.max_frame_size_bytes + ), + }, + }; + } + if frame_size == 20 { + return ParsedNetflow::Error { + error: NetflowError::Partial { + message: "V9 export packet must contain at least one FlowSet".to_string(), + }, + }; + } // Reset the per-parse restored-templates buffer so the next call // sees only what was restored during *this* packet. self.restored_templates.clear(); @@ -572,7 +616,6 @@ impl V9Parser { } } } - v9.header.count = u16::try_from(v9.flowsets.len()).unwrap_or(u16::MAX); } /// Try to replay a pending flow entry using available templates. @@ -1089,23 +1132,20 @@ impl ScopeDataField { impl FlowSetParser { pub(super) fn parse_flowsets<'a>( - i: &'a [u8], + mut remaining: &'a [u8], parser: &mut V9Parser, - record_count: u16, ) -> IResult<&'a [u8], Vec> { - // Cap pre-allocation to avoid memory amplification from untrusted header.count - let cap = (record_count as usize).min(64); - let (remaining, flowsets) = (0..record_count).try_fold( - (i, Vec::with_capacity(cap)), - |(remaining, mut flowsets), _| { - if remaining.is_empty() { - return Ok((remaining, flowsets)); - } - let (i, flowset) = FlowSet::parse(remaining, parser)?; - flowsets.push(flowset); - Ok((i, flowsets)) - }, - )?; + let capacity = (remaining.len() / 4).min(64); + let mut flowsets = Vec::with_capacity(capacity); + while !remaining.is_empty() { + let before = remaining; + let (next, flowset) = FlowSet::parse(remaining, parser)?; + if next.len() >= before.len() { + return Err(nom::Err::Error(NomError::new(remaining, ErrorKind::Verify))); + } + flowsets.push(flowset); + remaining = next; + } Ok((remaining, flowsets)) } diff --git a/src/variable_versions/v9/serializer.rs b/src/variable_versions/v9/serializer.rs index 79d4c2a..9a1912d 100644 --- a/src/variable_versions/v9/serializer.rs +++ b/src/variable_versions/v9/serializer.rs @@ -110,7 +110,9 @@ impl V9 { /// Convert the V9 struct to a `Vec` of bytes in big-endian order for exporting. /// /// `NoTemplate` flowsets are omitted from the output, and `header.count` - /// is recomputed to match the number of actually-serialized flowsets. + /// is recomputed to match the number of actually serialized records. + /// Returns an error if no FlowSet can be emitted or a FlowSet length or + /// total record count exceeds the corresponding 16-bit wire field. pub fn to_be_bytes(&self) -> Result, Box> { let mut result = Vec::new(); @@ -123,12 +125,19 @@ impl V9 { result.extend_from_slice(&self.header.source_id.to_be_bytes()); let mut emitted_count: u16 = 0; + let mut emitted_flowset = false; for set in self.flowsets.iter() { - let body_bytes = match &set.body { - FlowSetBody::Template(t) => Self::serialize_template_body(t), - FlowSetBody::OptionsTemplate(o) => Self::serialize_options_template_body(o), - FlowSetBody::Data(d) => Self::serialize_data_body(d)?, - FlowSetBody::OptionsData(o) => Self::serialize_options_data_body(o)?, + let (body_bytes, record_count) = match &set.body { + FlowSetBody::Template(t) => { + (Self::serialize_template_body(t), t.templates.len()) + } + FlowSetBody::OptionsTemplate(o) => { + (Self::serialize_options_template_body(o), o.templates.len()) + } + FlowSetBody::Data(d) => (Self::serialize_data_body(d)?, d.fields.len()), + FlowSetBody::OptionsData(o) => { + (Self::serialize_options_data_body(o)?, o.fields.len()) + } FlowSetBody::NoTemplate(_) | FlowSetBody::Empty => continue, }; // Compute flowset length from actual serialized body instead @@ -143,12 +152,19 @@ impl V9 { result.extend_from_slice(&set.header.flowset_id.to_be_bytes()); result.extend_from_slice(&flowset_length.to_be_bytes()); result.extend_from_slice(&body_bytes); + emitted_flowset = true; + let record_count = u16::try_from(record_count) + .map_err(|_| format!("V9 record count exceeds u16::MAX ({})", u16::MAX))?; emitted_count = emitted_count - .checked_add(1) - .ok_or_else(|| format!("V9 flowset count exceeds u16::MAX ({})", u16::MAX))?; + .checked_add(record_count) + .ok_or_else(|| format!("V9 record count exceeds u16::MAX ({})", u16::MAX))?; + } + + if !emitted_flowset { + return Err("V9 export packet must contain at least one FlowSet".into()); } - // Patch header.count with actual number of serialized flowsets + // Patch header.count with the total number of serialized records. result[2..4].copy_from_slice(&emitted_count.to_be_bytes()); Ok(result) diff --git a/src/variable_versions/v9/types.rs b/src/variable_versions/v9/types.rs index 93f5120..c4a152c 100644 --- a/src/variable_versions/v9/types.rs +++ b/src/variable_versions/v9/types.rs @@ -35,7 +35,7 @@ pub struct V9 { /// V9 Header pub header: Header, /// Flowsets - #[nom(Parse = "{ |i| FlowSetParser::parse_flowsets(i, parser, header.count) }")] + #[nom(Parse = "{ |i| FlowSetParser::parse_flowsets(i, parser) }")] pub flowsets: Vec, } @@ -45,7 +45,8 @@ pub struct Header { /// The version of NetFlow records exported in this packet; for Version 9, this value is 9 #[nom(Value = "9")] pub version: u16, - /// Number of FlowSet records (both template and data) contained within this packet + /// Total number of Template, Options Template, and Data records in this packet. + /// This is exporter-declared metadata and does not delimit the FlowSet sequence. pub count: u16, /// Time in milliseconds since this device was first booted pub sys_up_time: u32, diff --git a/tests/v9_count_framing.rs b/tests/v9_count_framing.rs new file mode 100644 index 0000000..260157f --- /dev/null +++ b/tests/v9_count_framing.rs @@ -0,0 +1,210 @@ +use netflow_parser::variable_versions::v9::{ + Data, FlowSet, FlowSetBody, FlowSetHeader, Header, V9, +}; +use netflow_parser::{NetflowPacket, NetflowParser, PendingFlowsConfig}; + +fn v9_header(count: u16) -> Vec { + let mut packet = Vec::new(); + packet.extend_from_slice(&9u16.to_be_bytes()); + packet.extend_from_slice(&count.to_be_bytes()); + packet.extend_from_slice(&0u32.to_be_bytes()); + packet.extend_from_slice(&0u32.to_be_bytes()); + packet.extend_from_slice(&1u32.to_be_bytes()); + packet.extend_from_slice(&1u32.to_be_bytes()); + packet +} + +fn append_flowset(packet: &mut Vec, id: u16, body: &[u8]) { + packet.extend_from_slice(&id.to_be_bytes()); + packet.extend_from_slice(&u16::try_from(4 + body.len()).unwrap().to_be_bytes()); + packet.extend_from_slice(body); +} + +fn template_record(template_id: u16, field_type: u16) -> Vec { + let mut record = Vec::new(); + record.extend_from_slice(&template_id.to_be_bytes()); + record.extend_from_slice(&1u16.to_be_bytes()); + record.extend_from_slice(&field_type.to_be_bytes()); + record.extend_from_slice(&4u16.to_be_bytes()); + record +} + +fn two_template_packet(count: u16) -> Vec { + let mut packet = v9_header(count); + let mut templates = template_record(256, 1); + templates.extend_from_slice(&template_record(257, 2)); + append_flowset(&mut packet, 0, &templates); + packet +} + +fn mixed_record_packet() -> Vec { + let mut packet = v9_header(6); + + append_flowset(&mut packet, 0, &template_record(256, 1)); + + let mut options_template = Vec::new(); + options_template.extend_from_slice(&257u16.to_be_bytes()); + options_template.extend_from_slice(&4u16.to_be_bytes()); + options_template.extend_from_slice(&4u16.to_be_bytes()); + options_template.extend_from_slice(&1u16.to_be_bytes()); + options_template.extend_from_slice(&4u16.to_be_bytes()); + options_template.extend_from_slice(&42u16.to_be_bytes()); + options_template.extend_from_slice(&4u16.to_be_bytes()); + options_template.extend_from_slice(&[0; 2]); + append_flowset(&mut packet, 1, &options_template); + + let mut data = Vec::new(); + data.extend_from_slice(&11u32.to_be_bytes()); + data.extend_from_slice(&22u32.to_be_bytes()); + append_flowset(&mut packet, 256, &data); + + let mut options_data = Vec::new(); + for value in [1u32, 2, 3, 4] { + options_data.extend_from_slice(&value.to_be_bytes()); + } + append_flowset(&mut packet, 257, &options_data); + + packet +} + +fn reserved_frame(size: usize) -> Vec { + assert!((24..=65_555).contains(&size)); + let mut packet = v9_header(0); + let body_len = size - 24; + append_flowset(&mut packet, 2, &vec![0; body_len]); + assert_eq!(packet.len(), size); + packet +} + +fn parsed_v9(result: &netflow_parser::ParseResult) -> &V9 { + assert!(result.error.is_none(), "{:?}", result.error); + assert_eq!(result.packets.len(), 1); + let NetflowPacket::V9(packet) = &result.packets[0] else { + panic!("expected NetFlow v9 packet"); + }; + packet +} + +#[test] +fn parses_flowsets_to_the_caller_delimited_frame_boundary() { + let result = NetflowParser::default().parse_bytes(&two_template_packet(0)); + let packet = parsed_v9(&result); + + assert_eq!(packet.header.count, 0); + assert_eq!(packet.flowsets.len(), 1); + let FlowSetBody::Template(templates) = &packet.flowsets[0].body else { + panic!("expected template flowset"); + }; + assert_eq!(templates.templates.len(), 2); +} + +#[test] +fn pending_flow_processing_preserves_the_exporter_declared_count() { + let mut parser = NetflowParser::builder() + .with_pending_flows(PendingFlowsConfig::default()) + .build() + .unwrap(); + + let result = parser.parse_bytes(&two_template_packet(2)); + let packet = parsed_v9(&result); + + assert_eq!(packet.header.count, 2); + assert_eq!(packet.flowsets.len(), 1); +} + +#[test] +fn serializer_counts_records_instead_of_flowsets() { + let result = NetflowParser::default().parse_bytes(&mixed_record_packet()); + let packet = parsed_v9(&result); + assert_eq!(packet.flowsets.len(), 4); + + let serialized = packet.to_be_bytes().unwrap(); + assert_eq!(u16::from_be_bytes([serialized[2], serialized[3]]), 6); +} + +#[test] +fn serializer_rejects_record_count_overflow() { + let packet = V9 { + header: Header { + version: 9, + count: 0, + sys_up_time: 0, + unix_secs: 0, + sequence_number: 0, + source_id: 0, + }, + flowsets: vec![FlowSet { + header: FlowSetHeader { + flowset_id: 256, + length: 4, + }, + body: FlowSetBody::Data(Data::new(vec![Vec::new(); 65_536])), + }], + }; + + assert!(packet.to_be_bytes().is_err()); +} + +#[test] +fn parser_and_serializer_reject_header_only_v9_packets() { + let result = NetflowParser::default().parse_bytes(&v9_header(0)); + assert!(result.packets.is_empty()); + assert!(result.error.is_some()); + + let packet = V9 { + header: Header { + version: 9, + count: 0, + sys_up_time: 0, + unix_secs: 0, + sequence_number: 0, + source_id: 0, + }, + flowsets: Vec::new(), + }; + assert!(packet.to_be_bytes().is_err()); +} + +#[test] +fn default_v9_frame_limit_accepts_the_limit_and_rejects_one_more_byte() { + let accepted = NetflowParser::default().parse_bytes(&reserved_frame(65_535)); + assert!(accepted.error.is_none(), "{:?}", accepted.error); + + let rejected = NetflowParser::default().parse_bytes(&reserved_frame(65_536)); + assert!(rejected.packets.is_empty()); + assert!(rejected.error.is_some()); +} + +#[test] +fn configured_v9_frame_limit_accepts_larger_caller_delimited_frames() { + let mut parser = NetflowParser::builder() + .with_v9_max_frame_size_bytes(65_536) + .build() + .unwrap(); + + let result = parser.parse_bytes(&reserved_frame(65_536)); + assert!(result.error.is_none(), "{:?}", result.error); +} + +#[test] +fn zero_v9_frame_limit_is_rejected() { + let result = NetflowParser::builder() + .with_v9_max_frame_size_bytes(0) + .build(); + assert!(result.is_err()); +} + +#[test] +fn oversized_v9_frame_is_rejected_before_learning_templates() { + let packet = two_template_packet(2); + let mut parser = NetflowParser::builder() + .with_v9_max_frame_size_bytes(packet.len() - 1) + .build() + .unwrap(); + + let result = parser.parse_bytes(&packet); + assert!(result.packets.is_empty()); + assert!(result.error.is_some()); + assert!(!parser.has_v9_template(256)); + assert!(!parser.has_v9_template(257)); +}