Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
5 changes: 5 additions & 0 deletions SECURITY.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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

Expand Down
214 changes: 170 additions & 44 deletions benches/hot_path_bench.rs
Original file line number Diff line number Diff line change
@@ -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![
Expand Down Expand Up @@ -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);
Loading
Loading