From 3f06a1521ff612a370cec16811a14a8dd978a2d4 Mon Sep 17 00:00:00 2001 From: mikemiles-dev Date: Sun, 11 Oct 2026 17:44:46 -0500 Subject: [PATCH] fix: skip an undecodable IPFIX Set instead of dropping the Sets behind it This is the remaining half of #310. The Set-framing check added in #330 could not reach it: a Set whose declared length is sound but whose body cannot be decoded is perfectly well framed, so framing validation passes and many0 still stopped there, dropping every Set behind it while the message reported success. A Template Set declaring more fields than its octets carry silently destroyed the valid Data Sets that followed. The framing pre-pass is what makes the fix possible. Every declared length in the chain is proved before any Set is parsed, so the next Set's offset is known even when the current one fails to decode. Replace many0 with a walk over that validated chain which steps over a failing Set and keeps going. validate_set_framing now returns the offset where the chain ends rather than unit. The walk needs that boundary instead of re-deriving the trailing-padding rule, because a disagreement between the two would be an infinite loop: a tolerated 4-to-7 octet zero tail declares length 0, and splitting at 0 never advances. Reported by the walk's own guard as well, but the boundary removes the duplication that would make them drift. The skipped Set is reported as FlowSetBody::NoTemplate carrying a new NoTemplateInfo::reason of NoTemplateReason::UndecodableBody, so the loss is visible in the data model without consuming the single ParseResult::error slot. reason also names the pre-existing case: a flowset whose template is merely absent reports MissingTemplate. That distinction is load-bearing, not cosmetic. Both V9 and IPFIX cache NoTemplate flowsets for pending-flow replay, and an undecodable Set must never enter that cache -- no template arriving later changes its octets, so it would be replayed and fail forever. Both caching paths now skip anything that is not replayable, and is_replayable() exposes the rule. NoTemplateInfo is #[non_exhaustive], so adding a field is not a breaking change and this stays on the unreleased 1.1.0 alongside the framing fix. A FlowSetBody variant would have forced 2.0.0. The walk also stops early once the decoded-output budget is exhausted. The PostExec hook rejects such a message wholesale, so continuing to decode Sets would spend work on output about to be discarded -- and would let a crafted message with many Sets amplify that work. Three bundled snapshots recorded flowsets: [] before this change and now show the Sets that were being dropped, so none of them was testing what its name claimed. Reviewed each diff before accepting. Verified with the main fuzz target: 2.6M runs, no crashes. That matters here because many0 was replaced with a hand-rolled loop, where a non-advancing split would hang rather than fail. Co-Authored-By: Claude Opus 5 (1M context) --- RELEASES.md | 26 +++ ...r__tests__base_tests__options_no_data.snap | 1 + ...ests__it_parses_0_length_fields_ipfix.snap | 44 +++- ...nterprise_bit_in_non_options_template.snap | 33 ++- ...ipfix_with_v9_options_template_packet.snap | 108 +++++++++- src/variable_versions/ipfix/parser.rs | 8 + src/variable_versions/ipfix/types.rs | 108 ++++++++-- src/variable_versions/mod.rs | 47 ++++- src/variable_versions/v9/parser.rs | 8 + tests/ipfix_undecodable_set.rs | 199 ++++++++++++++++++ 10 files changed, 557 insertions(+), 25 deletions(-) create mode 100644 tests/ipfix_undecodable_set.rs diff --git a/RELEASES.md b/RELEASES.md index f4fcc48..a7d75f7 100644 --- a/RELEASES.md +++ b/RELEASES.md @@ -44,6 +44,32 @@ elsewhere, `pcaps/IPFIX.pcap` included, so retiring both captures loses no coverage. +* **A Set this parser cannot decode no longer discards the Sets behind it.** + Set iteration stopped at the first Set whose body failed to decode and + dropped the rest of the message without reporting anything — the remaining + half of #310, which the Set-framing check could not reach, because such a Set + is perfectly well framed and only its *body* is undecodable. A Template Set + declaring more fields than its octets carry, for instance, silently destroyed + every valid Data Set behind it while the message reported success. + + Because Set framing is now validated across the whole chain before any Set is + parsed, the next Set's offset is known even when the current one fails, so the + walk steps over it and keeps decoding. The skipped Set is reported as + `FlowSetBody::NoTemplate` carrying the new + `NoTemplateInfo::reason` of `NoTemplateReason::UndecodableBody`, so the loss + stays visible without consuming the single `ParseResult::error` slot. + + `reason` also distinguishes the pre-existing case: a flowset whose template is + merely absent reports `NoTemplateReason::MissingTemplate` and remains eligible + for pending-flow replay. An undecodable Set never is — no template arriving + later changes its octets, so caching it would retry it forever. + `NoTemplateInfo::is_replayable()` exposes the distinction. + + Three bundled test snapshots recorded `flowsets: []` before this change and + now show the Sets that were being dropped, so they were not testing what their + names claimed. `NoTemplateInfo` is `#[non_exhaustive]`, so adding `reason` is + not a breaking change. + * **A malformed IPFIX message no longer discards the rest of the datagram.** RFC 7011 section 9.1 discards the *Message*, but parsing stopped at the first message-level error and abandoned every message chained behind it — so one diff --git a/src/snapshots/netflow_parser__tests__base_tests__options_no_data.snap b/src/snapshots/netflow_parser__tests__base_tests__options_no_data.snap index cf86692..ff09aa8 100644 --- a/src/snapshots/netflow_parser__tests__base_tests__options_no_data.snap +++ b/src/snapshots/netflow_parser__tests__base_tests__options_no_data.snap @@ -147,3 +147,4 @@ expression: parser.parse_bytes(&packet).packets - 0 - 0 truncated: false + reason: MissingTemplate diff --git a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_0_length_fields_ipfix.snap b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_0_length_fields_ipfix.snap index 46236ed..1af65bb 100644 --- a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_0_length_fields_ipfix.snap +++ b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_0_length_fields_ipfix.snap @@ -9,4 +9,46 @@ expression: "NetflowParser::default().parse_bytes(&packet).packets" export_time: 16909060 sequence_number: 0 observation_domain_id: 16909060 - flowsets: [] + flowsets: + - header: + header_id: 2 + length: 20 + body: + NoTemplate: + template_id: 2 + raw_data: + - 1 + - 0 + - 0 + - 3 + - 0 + - 8 + - 0 + - 4 + - 0 + - 12 + - 0 + - 4 + - 0 + - 65 + - 0 + - 0 + truncated: false + reason: UndecodableBody + - header: + header_id: 256 + length: 12 + body: + NoTemplate: + template_id: 256 + raw_data: + - 1 + - 2 + - 3 + - 4 + - 1 + - 2 + - 3 + - 4 + truncated: false + reason: MissingTemplate diff --git a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_enterprise_bit_in_non_options_template.snap b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_enterprise_bit_in_non_options_template.snap index 88d62f3..58f0001 100644 --- a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_enterprise_bit_in_non_options_template.snap +++ b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_enterprise_bit_in_non_options_template.snap @@ -9,4 +9,35 @@ expression: "NetflowParser::default().parse_bytes(&packet).packets" export_time: 1670052913 sequence_number: 0 observation_domain_id: 0 - flowsets: [] + flowsets: + - header: + header_id: 2 + length: 26 + body: + NoTemplate: + template_id: 2 + raw_data: + - 1 + - 4 + - 0 + - 2 + - 128 + - 103 + - 255 + - 255 + - 24 + - 77 + - 128 + - 103 + - 255 + - 255 + - 0 + - 0 + - 24 + - 77 + - 129 + - 64 + - 0 + - 4 + truncated: false + reason: UndecodableBody diff --git a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_with_v9_options_template_packet.snap b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_with_v9_options_template_packet.snap index 5bf7391..9ef18bc 100644 --- a/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_with_v9_options_template_packet.snap +++ b/src/snapshots/netflow_parser__tests__restored_legacy_tests__it_parses_ipfix_with_v9_options_template_packet.snap @@ -9,4 +9,110 @@ expression: "NetflowParser::default().parse_bytes(&packet).packets" export_time: 1744871721 sequence_number: 100000 observation_domain_id: 0 - flowsets: [] + flowsets: + - header: + header_id: 1 + length: 14 + body: + NoTemplate: + template_id: 1 + raw_data: + - 4 + - 232 + - 0 + - 1 + - 0 + - 6 + - 0 + - 34 + - 0 + - 4 + truncated: false + reason: UndecodableBody + - header: + header_id: 1256 + length: 8 + body: + NoTemplate: + template_id: 1256 + raw_data: + - 0 + - 0 + - 0 + - 1 + truncated: false + reason: MissingTemplate + - header: + header_id: 2 + length: 44 + body: + Template: + template_id: 258 + field_count: 9 + fields: + - field_type_number: 8 + field_length: 4 + field_type: + IANA: SourceIpv4address + - field_type_number: 12 + field_length: 4 + field_type: + IANA: DestinationIpv4address + - field_type_number: 7 + field_length: 2 + field_type: + IANA: SourceTransportPort + - field_type_number: 11 + field_length: 2 + field_type: + IANA: DestinationTransportPort + - field_type_number: 4 + field_length: 1 + field_type: + IANA: ProtocolIdentifier + - field_type_number: 2 + field_length: 4 + field_type: + IANA: PacketDeltaCount + - field_type_number: 1 + field_length: 4 + field_type: + IANA: OctetDeltaCount + - field_type_number: 22 + field_length: 4 + field_type: + IANA: FlowStartSysUpTime + - field_type_number: 21 + field_length: 4 + field_type: + IANA: FlowEndSysUpTime + - header: + header_id: 258 + length: 33 + body: + Data: + fields: + - - - IANA: SourceIpv4address + - Ip4Addr: 70.70.1.1 + - - IANA: DestinationIpv4address + - Ip4Addr: 120.120.1.1 + - - IANA: SourceTransportPort + - DataNumber: + U16: 12345 + - - IANA: DestinationTransportPort + - DataNumber: + U16: 32777 + - - IANA: ProtocolIdentifier + - ProtocolType: Tcp + - - IANA: PacketDeltaCount + - DataNumber: + U32: 100 + - - IANA: OctetDeltaCount + - DataNumber: + U32: 125000000 + - - IANA: FlowStartSysUpTime + - DataNumber: + U32: 1744871721 + - - IANA: FlowEndSysUpTime + - DataNumber: + U32: 1744871721 diff --git a/src/variable_versions/ipfix/parser.rs b/src/variable_versions/ipfix/parser.rs index 0698ca3..0be7220 100644 --- a/src/variable_versions/ipfix/parser.rs +++ b/src/variable_versions/ipfix/parser.rs @@ -16,6 +16,7 @@ use crate::template_store::{ encode_ipfix_options_template, encode_ipfix_template, encode_v9_options_template, encode_v9_template, }; +use crate::variable_versions::NoTemplateReason; use crate::variable_versions::config::DEFAULT_MAX_RECORDS_PER_FLOWSET; use crate::variable_versions::enterprise_registry::EnterpriseFieldRegistry; use crate::variable_versions::field_value::FieldValue; @@ -645,6 +646,12 @@ impl IPFixParser { for (i, flowset) in ipfix.flowsets.iter_mut().enumerate() { match &mut flowset.body { FlowSetBody::NoTemplate(info) => { + // A Set whose body could not be decoded is not waiting on a + // template: no template that arrives later changes its + // octets, so caching it for replay would retry it forever. + if !info.is_replayable() { + continue; + } // Reject flowsets with impossibly small headers (RFC minimum is 4). // Also reject truncated raw_data (oversized entry at parse time). // The flowset is kept in output as diagnostic data. @@ -1747,6 +1754,7 @@ impl FlowSetBody { template_id: id, raw_data, truncated, + reason: NoTemplateReason::MissingTemplate, }; Ok((&[] as &[u8], FlowSetBody::NoTemplate(info))) } else { diff --git a/src/variable_versions/ipfix/types.rs b/src/variable_versions/ipfix/types.rs index b6bf890..b339d83 100644 --- a/src/variable_versions/ipfix/types.rs +++ b/src/variable_versions/ipfix/types.rs @@ -16,8 +16,8 @@ use crate::variable_versions::template_events::TemplateProtocol; use crate::variable_versions::ttl::{TemplateWithTtl, TtlConfig}; use nom::bytes::complete::take; -use nom::combinator::{complete, map_res}; -use nom::multi::{count, many0}; +use nom::combinator::map_res; +use nom::multi::count; use nom_derive::{Nom, Parse}; use serde::Serialize; @@ -121,13 +121,18 @@ const MAX_TRAILING_PADDING: usize = 7; /// A trailing run of fewer than 4 octets cannot begin a Set, and a trailing /// run of up to [`MAX_TRAILING_PADDING`] zero octets is alignment padding; /// both are accepted. -pub(crate) fn validate_set_framing(region: &[u8]) -> Result<(), SetFramingLocation> { +/// +/// Returns the octet offset where the Set chain ends, i.e. where tolerated +/// padding begins. [`walk_set_chain`] relies on that boundary rather than +/// re-deriving the padding rule, which would risk disagreeing with it and +/// stalling on a zero tail. +pub(crate) fn validate_set_framing(region: &[u8]) -> Result { let mut offset = 0usize; loop { let tail = ®ion[offset..]; if tail.len() < 4 { // Too short to begin a Set; tolerated as trailing padding. - return Ok(()); + return Ok(offset); } let declared = usize::from(u16::from_be_bytes([tail[2], tail[3]])); if declared < 4 { @@ -135,7 +140,7 @@ pub(crate) fn validate_set_framing(region: &[u8]) -> Result<(), SetFramingLocati // run followed by more data is not, so the `all` check below is // what keeps this from excusing a zero gap mid-message. if tail.len() <= MAX_TRAILING_PADDING && tail.iter().all(|&b| b == 0) { - return Ok(()); + return Ok(offset); } return Err(SetFramingLocation { offset, @@ -155,6 +160,67 @@ pub(crate) fn validate_set_framing(region: &[u8]) -> Result<(), SetFramingLocati } } +/// Decodes each Set in a framing-validated chain, skipping any whose body +/// cannot be decoded rather than abandoning the rest of the message. +/// +/// `many0` stopped at the first Set it could not parse, and the Sets behind it +/// were dropped with no indication — the half of netflow_parser#310 that the +/// framing check could not reach, because such a Set is perfectly well framed. +/// Since [`validate_set_framing`] has already proved every declared length in +/// `chain`, the next Set's offset is known even when the current one fails, so +/// the walk can step over it. +/// +/// A skipped Set is recorded as [`FlowSetBody::NoTemplate`] carrying +/// [`NoTemplateReason::UndecodableBody`], which keeps the loss visible without +/// making it eligible for pending-flow replay. +pub(crate) fn walk_set_chain(chain: &[u8], parser: &mut IPFixParser) -> Vec { + let mut flow_sets = Vec::new(); + let mut rest = chain; + + while rest.len() >= 4 { + // A message that has already blown its decoded-output budget is + // rejected wholesale by the PostExec hook, so continuing to decode + // Sets would be work spent on output that is about to be discarded. + if parser.decoded_output_limit_was_exceeded() { + break; + } + + // Framing is validated, so `declared` is within 4..=rest.len() and the + // split always advances. + let declared = usize::from(u16::from_be_bytes([rest[2], rest[3]])); + if declared < 4 || declared > rest.len() { + break; + } + let (set_bytes, remainder) = rest.split_at(declared); + + match FlowSet::parse(set_bytes, parser) { + Ok((_, flow_set)) => flow_sets.push(flow_set), + Err(_) => flow_sets.push(undecodable_set(set_bytes, parser.max_error_sample_size)), + } + rest = remainder; + } + + flow_sets +} + +/// Records a well-framed Set whose body could not be decoded, keeping a +/// bounded sample of the octets for diagnosis. +fn undecodable_set(set_bytes: &[u8], max_error_sample_size: usize) -> FlowSet { + let header_id = u16::from_be_bytes([set_bytes[0], set_bytes[1]]); + let length = u16::from_be_bytes([set_bytes[2], set_bytes[3]]); + let body = &set_bytes[4..]; + let limit = body.len().min(max_error_sample_size); + + FlowSet { + header: FlowSetHeader { header_id, length }, + body: FlowSetBody::NoTemplate(NoTemplateInfo::undecodable( + header_id, + body[..limit].to_vec(), + limit < body.len(), + )), + } +} + /// A parsed IPFIX message containing a header and a list of flowsets. #[derive(Nom, Debug, PartialEq, Clone, Serialize)] #[nom( @@ -185,21 +251,25 @@ pub struct IPFix { // // map_res discards the nom error constructed here, so the precise // violation is stashed on the parser for IPFixParser::parse. - if let Err(location) = validate_set_framing(i) { - parser.set_framing_error = Some(SetFramingViolation { - offset: location.offset, - declared: location.declared, - available: location.available, - export_time: header.export_time, - sequence_number: header.sequence_number, - observation_domain_id: header.observation_domain_id, - }); - return Err(nom::Err::Error(nom::error::Error::new(i, nom::error::ErrorKind::Verify))); + match validate_set_framing(i) { + Err(location) => { + parser.set_framing_error = Some(SetFramingViolation { + offset: location.offset, + declared: location.declared, + available: location.available, + export_time: header.export_time, + sequence_number: header.sequence_number, + observation_domain_id: header.observation_domain_id, + }); + Err(nom::Err::Error(nom::error::Error::new(i, nom::error::ErrorKind::Verify))) + } + // Walk the validated chain rather than many0, which halted at + // the first Set it could not decode and silently dropped the + // Sets behind it. See walk_set_chain. + Ok(chain_end) => Ok::<_, nom::Err>>( + walk_set_chain(&i[..chain_end], parser) + ), } - many0(complete(|i| FlowSet::parse(i, parser) - .map(|(i, flow_set)| (i, flow_set)) - ))(i) - .map(|(_, flow_sets)| flow_sets) // Extract the Vec })" )] pub flowsets: Vec, diff --git a/src/variable_versions/mod.rs b/src/variable_versions/mod.rs index 1ebf6cc..f5a8795 100644 --- a/src/variable_versions/mod.rs +++ b/src/variable_versions/mod.rs @@ -103,28 +103,68 @@ pub(crate) use pending_flows::{PendingFlowCache, PendingFlowEntry}; use crate::variable_versions::metrics::CacheMetricsInner; use crate::variable_versions::ttl::TtlConfig; -/// Information about a data flowset that couldn't be parsed due to missing template. +/// Why a flowset was not decoded into records. +/// +/// Distinguishes the two cases that both leave a [`NoTemplateInfo`] behind, +/// because only one of them can ever succeed on a retry. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize)] +pub enum NoTemplateReason { + /// The flowset's template ID was not in the cache. The records may still + /// decode once the template arrives, so these are eligible for + /// pending-flow replay. + #[default] + MissingTemplate, + /// The Set was well framed but its body could not be decoded — a record + /// count or field layout that does not match the octets present, for + /// instance. Replaying it cannot help, because no later template changes + /// the bytes, so these are never cached for retry. + UndecodableBody, +} + +/// Information about a data flowset that couldn't be parsed. /// This provides context to help diagnose template-related issues. #[non_exhaustive] #[derive(Debug, Clone, serde::Serialize)] pub struct NoTemplateInfo { - /// The template ID that was requested but not found + /// The template ID that was requested but not found, or the Set ID of a + /// Set whose body could not be decoded pub template_id: u16, /// The unparsed flowset data (for potential retry after template arrives) pub raw_data: Vec, /// Whether the raw_data was truncated due to size limits pub truncated: bool, + /// Why the flowset was not decoded. Only [`NoTemplateReason::MissingTemplate`] + /// is eligible for pending-flow replay. + pub reason: NoTemplateReason, } impl NoTemplateInfo { - /// Create a new NoTemplateInfo with the given template ID and raw data + /// Create a new NoTemplateInfo for a flowset whose template is missing pub fn new(template_id: u16, raw_data: Vec) -> Self { Self { template_id, raw_data, truncated: false, + reason: NoTemplateReason::MissingTemplate, + } + } + + /// Create a new NoTemplateInfo for a well-framed Set whose body could not + /// be decoded. Never eligible for pending-flow replay. + pub fn undecodable(set_id: u16, raw_data: Vec, truncated: bool) -> Self { + Self { + template_id: set_id, + raw_data, + truncated, + reason: NoTemplateReason::UndecodableBody, } } + + /// Whether this flowset could decode on a retry once its template arrives. + pub fn is_replayable(&self) -> bool { + matches!(self.reason, NoTemplateReason::MissingTemplate) + } } impl PartialEq for NoTemplateInfo { @@ -132,6 +172,7 @@ impl PartialEq for NoTemplateInfo { self.template_id == other.template_id && self.raw_data == other.raw_data && self.truncated == other.truncated + && self.reason == other.reason } } diff --git a/src/variable_versions/v9/parser.rs b/src/variable_versions/v9/parser.rs index 0bd619c..f9b9546 100644 --- a/src/variable_versions/v9/parser.rs +++ b/src/variable_versions/v9/parser.rs @@ -16,6 +16,7 @@ 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::NoTemplateReason; use crate::variable_versions::config::{ DEFAULT_MAX_RECORDS_PER_FLOWSET, DEFAULT_MAX_V9_FRAME_SIZE_BYTES, }; @@ -621,6 +622,12 @@ impl V9Parser { for (i, flowset) in v9.flowsets.iter_mut().enumerate() { match &mut flowset.body { FlowSetBody::NoTemplate(info) => { + // A Set whose body could not be decoded is not waiting on a + // template: no template that arrives later changes its + // octets, so caching it for replay would retry it forever. + if !info.is_replayable() { + continue; + } // Reject flowsets with impossibly small headers (RFC minimum is 4). // Also reject truncated raw_data (oversized entry at parse time). // The flowset is kept in output as diagnostic data. @@ -951,6 +958,7 @@ impl FlowSetBody { template_id: id, raw_data, truncated, + reason: NoTemplateReason::MissingTemplate, }; Ok((&[] as &[u8], FlowSetBody::NoTemplate(info))) } else { diff --git a/tests/ipfix_undecodable_set.rs b/tests/ipfix_undecodable_set.rs new file mode 100644 index 0000000..5d97866 --- /dev/null +++ b/tests/ipfix_undecodable_set.rs @@ -0,0 +1,199 @@ +//! A Set this parser cannot decode must not take the Sets behind it down. +//! +//! The remaining half of netflow_parser#310. The framing check could not reach +//! this case, because such a Set is perfectly well framed — only its body is +//! undecodable — so `many0` stopped there and every Set behind it was dropped +//! while the message reported success. + +use netflow_parser::variable_versions::NoTemplateReason; +use netflow_parser::variable_versions::ipfix::FlowSetBody; +use netflow_parser::{NetflowPacket, NetflowParser, PendingFlowsConfig}; + +fn message(body: &[u8]) -> Vec { + let mut m = Vec::new(); + m.extend_from_slice(&10u16.to_be_bytes()); + m.extend_from_slice(&0u16.to_be_bytes()); // patched below + m.extend_from_slice(&1u32.to_be_bytes()); + m.extend_from_slice(&2u32.to_be_bytes()); + m.extend_from_slice(&3u32.to_be_bytes()); + m.extend_from_slice(body); + let len = u16::try_from(m.len()).unwrap(); + m[2..4].copy_from_slice(&len.to_be_bytes()); + m +} + +/// Template Set defining Template 256 with one 4-octet sourceIPv4Address. +fn template_set() -> Vec { + let mut s = Vec::new(); + for v in [2u16, 12, 256, 1, 8, 4] { + s.extend_from_slice(&v.to_be_bytes()); + } + s +} + +/// A Template Set whose declared `field_count` exceeds the octets present, so +/// the Set is well framed but its body cannot be decoded. +fn undecodable_set() -> Vec { + let mut s = Vec::new(); + for v in [2u16, 12, 257, 100, 8, 4] { + s.extend_from_slice(&v.to_be_bytes()); + } + s +} + +fn data_set(payload: [u8; 4]) -> Vec { + let mut s = Vec::new(); + for v in [256u16, 8] { + s.extend_from_slice(&v.to_be_bytes()); + } + s.extend_from_slice(&payload); + s +} + +fn data_record_count(packets: &[NetflowPacket]) -> usize { + packets + .iter() + .filter_map(|p| match p { + NetflowPacket::IPFix(i) => Some(i), + _ => None, + }) + .flat_map(|i| i.flowsets.iter()) + .map(|fs| match &fs.body { + FlowSetBody::Data(d) => d.fields.len(), + _ => 0, + }) + .sum() +} + +fn no_template_bodies(packets: &[NetflowPacket]) -> Vec<(u16, NoTemplateReason)> { + packets + .iter() + .filter_map(|p| match p { + NetflowPacket::IPFix(i) => Some(i), + _ => None, + }) + .flat_map(|i| i.flowsets.iter()) + .filter_map(|fs| match &fs.body { + FlowSetBody::NoTemplate(info) => Some((info.template_id, info.reason)), + _ => None, + }) + .collect() +} + +/// The case with teeth: a valid Data Set behind an undecodable Set. +#[test] +fn data_behind_an_undecodable_set_survives() { + let mut body = template_set(); + body.extend_from_slice(&undecodable_set()); + body.extend_from_slice(&data_set([192, 0, 2, 3])); + + let mut parser = NetflowParser::default(); + let result = parser.parse_bytes(&message(&body)); + + assert!(result.error.is_none(), "{:?}", result.error); + assert_eq!( + data_record_count(&result.packets), + 1, + "the Data Set behind the undecodable Set must still decode" + ); +} + +/// The skipped Set stays visible, and is marked as something a retry cannot fix. +#[test] +fn the_undecodable_set_is_reported_as_undecodable() { + let mut body = template_set(); + body.extend_from_slice(&undecodable_set()); + body.extend_from_slice(&data_set([192, 0, 2, 3])); + + let mut parser = NetflowParser::default(); + let result = parser.parse_bytes(&message(&body)); + + assert_eq!( + no_template_bodies(&result.packets), + vec![(2, NoTemplateReason::UndecodableBody)], + "exactly the skipped Set must be reported, as undecodable" + ); +} + +/// A genuinely missing template keeps its own reason and stays replayable; +/// the new variant must not swallow the pre-existing case. +#[test] +fn a_missing_template_is_still_reported_as_missing() { + let mut parser = NetflowParser::default(); + let result = parser.parse_bytes(&message(&data_set([1, 2, 3, 4]))); + + assert_eq!( + no_template_bodies(&result.packets), + vec![(256, NoTemplateReason::MissingTemplate)] + ); +} + +/// The subtle part: an undecodable Set must never enter the pending-flow cache. +/// No template arriving later changes its octets, so a cached entry would be +/// replayed and fail forever. +#[test] +fn an_undecodable_set_is_never_cached_for_replay() { + let mut parser = NetflowParser::builder() + .with_pending_flows(PendingFlowsConfig::new(16)) + .build() + .expect("valid config"); + + // A data Set for an absent template is cached; the undecodable Set is not. + let mut body = undecodable_set(); + body.extend_from_slice(&data_set([192, 0, 2, 4])); + let result = parser.parse_bytes(&message(&body)); + assert!(result.error.is_none(), "{:?}", result.error); + + let stats = parser.ipfix_cache_info(); + assert_eq!( + stats.pending_flow_count, 1, + "only the missing-template Set may be cached" + ); + assert_eq!( + stats.metrics.pending_cached, 1, + "the undecodable Set must not be counted as cached" + ); + + // And it must not be counted as a drop either: it was never a candidate. + assert_eq!( + stats.metrics.pending_dropped, 0, + "an undecodable Set is skipped, not dropped from the cache" + ); +} + +/// Guard against over-correction: a message of entirely decodable Sets is +/// unaffected, and nothing is reported as undecodable. +#[test] +fn well_formed_messages_are_unaffected() { + let mut body = template_set(); + body.extend_from_slice(&data_set([192, 0, 2, 1])); + body.extend_from_slice(&data_set([192, 0, 2, 2])); + + let mut parser = NetflowParser::default(); + let result = parser.parse_bytes(&message(&body)); + + assert!(result.error.is_none(), "{:?}", result.error); + assert_eq!(data_record_count(&result.packets), 2); + assert!(no_template_bodies(&result.packets).is_empty()); +} + +/// Several undecodable Sets in one message are each skipped independently. +#[test] +fn consecutive_undecodable_sets_are_each_skipped() { + let mut body = template_set(); + body.extend_from_slice(&undecodable_set()); + body.extend_from_slice(&undecodable_set()); + body.extend_from_slice(&data_set([192, 0, 2, 5])); + + let mut parser = NetflowParser::default(); + let result = parser.parse_bytes(&message(&body)); + + assert_eq!(data_record_count(&result.packets), 1); + assert_eq!( + no_template_bodies(&result.packets), + vec![ + (2, NoTemplateReason::UndecodableBody), + (2, NoTemplateReason::UndecodableBody) + ] + ); +}