fix: bound cumulative decoded output per message (#305)

* Bound cumulative decoded output per message

* Add decoded-output performance gates

* Verify benchmark decoded record counts

* Reject decoded-output accounting overflow

* Resolve decoded-output test lint warnings

* fix pending replay body validation

* isolate pending replay validation changes

* fix: harden cumulative decoded-output limits

* fix: classify IPFIX replay trigger boundaries

Author: ktsaou

README.md

@@ -258,6 +258,35 @@ The parser also automatically validates:
   - Each occurrence is decoded as a distinct value in the record
   - Descriptor order is significant and preserved
 
+### Cumulative Decoded Output Limits
+
+A small NetFlow v9 or IPFIX message can expand into many decoded field values,
+especially when a template contains zero-width or very small fields. The parser
+therefore applies two cumulative limits to each message, including pending data
+replayed when a template arrives:
+
+- 65,536 decoded field values
+- 4 MiB of decoded field content
+
+The content-byte limit counts field contents, not IPFIX variable-length prefixes.
+Both limits must be greater than zero. A message that exceeds either limit is
+rejected with `NetflowError::DecodedOutputLimitExceeded`; no packet from that
+message is returned.
+
+```rust
+use netflow_parser::NetflowParser;
+
+let parser = NetflowParser::builder()
+    .with_max_decoded_field_values_per_message(32_768)
+    .with_max_decoded_field_payload_bytes_per_message(2 * 1024 * 1024)
+    .build()
+    .expect("valid limits");
+```
+
+Use the `with_v9_*` and `with_ipfix_*` variants to configure the protocols
+independently. These limits complement the per-FlowSet record limit; they bound
+the combined decoded output of all Sets or FlowSets in one message.
+
 ### Template TTL (Time-to-Live)
 
 > **Note:** Only time-based TTL is supported. See [RELEASES.md](RELEASES.md) for details.

SECURITY.md

@@ -124,6 +124,10 @@ let parser = NetflowParser::builder()
     // Limit fields per template (DoS protection)
     .with_max_field_count(5000)
 
+    // Bound cumulative decoded output from each message
+    .with_max_decoded_field_values_per_message(65_536)
+    .with_max_decoded_field_payload_bytes_per_message(4 * 1024 * 1024)
+
     // Limit error sample size (prevents memory exhaustion)
     .with_max_error_sample_size(256)
 
@@ -154,6 +158,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
 - **Error Sample Size Limit:** Default 256 bytes to prevent memory exhaustion
 - **LRU Template Cache:** Prevents unbounded cache growth
 

benches/hot_path_bench.rs

@@ -1,6 +1,8 @@
 use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
-use netflow_parser::NetflowParser;
+use netflow_parser::scoped_parser::AutoScopedParser;
+use netflow_parser::{NetflowPacket, NetflowParser};
 use std::hint::black_box;
+use std::net::SocketAddr;
 
 fn v9_template_packet() -> Vec<u8> {
     vec![
@@ -167,55 +169,179 @@ fn ipfix_data_packet(flow_count: u16) -> Vec<u8> {
     packet
 }
 
-fn bench_warm_v9_data_hot_path(c: &mut Criterion) {
-    let template = v9_template_packet();
-    let mut group = c.benchmark_group("Hot Path V9 Data");
-
-    for flow_count in [100u16, 500, 1000] {
-        let data = v9_data_packet(flow_count);
-        group.throughput(Throughput::Bytes(data.len() as u64));
-        group.bench_with_input(BenchmarkId::from_parameter(flow_count), &data, |b, pkt| {
-            let mut parser = NetflowParser::default();
-            let template_result = parser.parse_bytes(&template);
-            assert!(template_result.error.is_none());
-            assert_eq!(template_result.packets.len(), 1);
-
-            b.iter(|| {
-                let result = parser.parse_bytes(black_box(pkt));
-                black_box(result.packets.len());
-            });
-        });
+#[derive(Clone, Copy)]
+enum Protocol {
+    V9,
+    Ipfix,
+}
+
+impl Protocol {
+    fn name(self) -> &'static str {
+        match self {
+            Self::V9 => "v9",
+            Self::Ipfix => "ipfix",
+        }
+    }
+
+    fn template_packet(self) -> Vec<u8> {
+        match self {
+            Self::V9 => v9_template_packet(),
+            Self::Ipfix => ipfix_template_packet(),
+        }
     }
 
-    group.finish();
+    fn data_packet(self, flow_count: u16) -> Vec<u8> {
+        match self {
+            Self::V9 => v9_data_packet(flow_count),
+            Self::Ipfix => ipfix_data_packet(flow_count),
+        }
+    }
+
+    fn assert_decoded_records(self, packets: &[NetflowPacket], expected: usize) {
+        assert_eq!(packets.len(), 1, "fixture must decode one outer packet");
+        let actual: usize = match (self, &packets[0]) {
+            (Self::V9, NetflowPacket::V9(packet)) => packet
+                .flowsets
+                .iter()
+                .map(|flowset| match &flowset.body {
+                    netflow_parser::variable_versions::v9::FlowSetBody::Data(data) => {
+                        data.fields.len()
+                    }
+                    _ => 0,
+                })
+                .sum(),
+            (Self::Ipfix, NetflowPacket::IPFix(packet)) => packet
+                .flowsets
+                .iter()
+                .map(|flowset| match &flowset.body {
+                    netflow_parser::variable_versions::ipfix::FlowSetBody::Data(data) => {
+                        data.fields.len()
+                    }
+                    _ => 0,
+                })
+                .sum(),
+            _ => panic!("fixture decoded as the wrong protocol"),
+        };
+        assert_eq!(actual, expected, "fixture decoded-record count mismatch");
+    }
 }
 
-fn bench_warm_ipfix_data_hot_path(c: &mut Criterion) {
-    let template = ipfix_template_packet();
-    let mut group = c.benchmark_group("Hot Path IPFIX Data");
-
-    for flow_count in [100u16, 500, 1000] {
-        let data = ipfix_data_packet(flow_count);
-        group.throughput(Throughput::Bytes(data.len() as u64));
-        group.bench_with_input(BenchmarkId::from_parameter(flow_count), &data, |b, pkt| {
-            let mut parser = NetflowParser::default();
-            let template_result = parser.parse_bytes(&template);
-            assert!(template_result.error.is_none());
-            assert_eq!(template_result.packets.len(), 1);
-
-            b.iter(|| {
-                let result = parser.parse_bytes(black_box(pkt));
-                black_box(result.packets.len());
-            });
-        });
+#[derive(Clone, Copy)]
+enum Scenario {
+    DirectParse,
+    DirectIterator,
+    AutoParse,
+    AutoIterator,
+}
+
+impl Scenario {
+    fn name(self) -> &'static str {
+        match self {
+            Self::DirectParse => "direct/parse",
+            Self::DirectIterator => "direct/iterator",
+            Self::AutoParse => "auto/parse",
+            Self::AutoIterator => "auto/iterator",
+        }
     }
+}
 
-    group.finish();
+fn bench_warmed_hot_paths(c: &mut Criterion) {
+    let source = SocketAddr::from(([192, 0, 2, 1], 2055));
+
+    for protocol in [Protocol::V9, Protocol::Ipfix] {
+        let template = protocol.template_packet();
+        for scenario in [
+            Scenario::DirectParse,
+            Scenario::DirectIterator,
+            Scenario::AutoParse,
+            Scenario::AutoIterator,
+        ] {
+            let mut group =
+                c.benchmark_group(format!("Hot Path/{}/{}", protocol.name(), scenario.name()));
+
+            for flow_count in [1u16, 1000] {
+                let data = protocol.data_packet(flow_count);
+                group.throughput(Throughput::Elements(u64::from(flow_count)));
+                group.bench_with_input(
+                    BenchmarkId::from_parameter(flow_count),
+                    &data,
+                    |b, packet| match scenario {
+                        Scenario::DirectParse => {
+                            let mut parser = NetflowParser::default();
+                            assert!(parser.parse_bytes(&template).is_ok());
+                            let result = parser.parse_bytes(packet);
+                            assert!(result.is_ok());
+                            protocol.assert_decoded_records(
+                                &result.packets,
+                                usize::from(flow_count),
+                            );
+                            b.iter(|| {
+                                drop(black_box(
+                                    parser.parse_bytes(black_box(packet.as_slice())),
+                                ));
+                            });
+                        }
+                        Scenario::DirectIterator => {
+                            let mut parser = NetflowParser::default();
+                            assert!(parser.parse_bytes(&template).is_ok());
+                            let packets = parser
+                                .iter_packets(packet)
+                                .map(Result::unwrap)
+                                .collect::<Vec<_>>();
+                            protocol.assert_decoded_records(&packets, usize::from(flow_count));
+                            b.iter(|| {
+                                for result in parser.iter_packets(black_box(packet.as_slice()))
+                                {
+                                    black_box(result.unwrap());
+                                }
+                            });
+                        }
+                        Scenario::AutoParse => {
+                            let mut parser = AutoScopedParser::new();
+                            assert!(parser.parse_from_source(source, &template).is_ok());
+                            let result = parser.parse_from_source(source, packet);
+                            assert!(result.is_ok());
+                            protocol.assert_decoded_records(
+                                &result.packets,
+                                usize::from(flow_count),
+                            );
+                            b.iter(|| {
+                                drop(black_box(
+                                    parser.parse_from_source(
+                                        source,
+                                        black_box(packet.as_slice()),
+                                    ),
+                                ));
+                            });
+                        }
+                        Scenario::AutoIterator => {
+                            let mut parser = AutoScopedParser::new();
+                            assert!(parser.parse_from_source(source, &template).is_ok());
+                            let packets = parser
+                                .iter_packets_from_source(source, packet)
+                                .unwrap()
+                                .map(Result::unwrap)
+                                .collect::<Vec<_>>();
+                            protocol.assert_decoded_records(&packets, usize::from(flow_count));
+                            b.iter(|| {
+                                let iterator = parser
+                                    .iter_packets_from_source(
+                                        source,
+                                        black_box(packet.as_slice()),
+                                    )
+                                    .unwrap();
+                                for result in iterator {
+                                    black_box(result.unwrap());
+                                }
+                            });
+                        }
+                    },
+                );
+            }
+            group.finish();
+        }
+    }
 }
 
-criterion_group!(
-    benches,
-    bench_warm_v9_data_hot_path,
-    bench_warm_ipfix_data_hot_path
-);
+criterion_group!(benches, bench_warmed_hot_paths);
 criterion_main!(benches);