From 5b492a699b081f7b48ccf8399408f79c1952b67d Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Tue, 18 Aug 2026 13:28:23 -0700 Subject: [PATCH 01/12] fix(inputs/flow): don't drop whole NetFlow v9/IPFIX datagram on unknown (options) flowset, or on legitimate zero-flow packets Data flowsets/sets referencing a template id we don't have a data template for (because it's actually an options template) no longer abort the whole datagram parse -- just that flowset is skipped, and any other flowsets in the same datagram are still processed. A structurally valid datagram with zero flow records (e.g. a pure template refresh) is also no longer counted as a parse error. --- src/handlers/flow/test_flows.cpp | 85 ++++++++++-- src/inputs/flow/NetflowData.h | 187 +++++++++++++++++++++++--- src/inputs/flow/test_flow.cpp | 32 +++++ src/tests/fixtures/ipfix_options.pcap | Bin 0 -> 266 bytes src/tests/fixtures/nf9_options.pcap | Bin 0 -> 278 bytes 5 files changed, 277 insertions(+), 27 deletions(-) create mode 100644 src/tests/fixtures/ipfix_options.pcap create mode 100644 src/tests/fixtures/nf9_options.pcap diff --git a/src/handlers/flow/test_flows.cpp b/src/handlers/flow/test_flows.cpp index fdc9a5104..58df2773e 100644 --- a/src/handlers/flow/test_flows.cpp +++ b/src/handlers/flow/test_flows.cpp @@ -470,9 +470,12 @@ TEST_CASE("Parse netflow v9 stream", "[netflow][flow]") auto event_data = flow_handler.metrics()->bucket(0)->event_data_locked(); - // confirmed with wireshark - CHECK(event_data.num_events->value() == 1); - CHECK(event_data.num_samples->value() == 1); + // nf9.pcap has 5 total datagrams; 4 are template/options-only (zero + // flow records) and were previously miscounted as parse errors instead + // of successfully ingested samples. records_flows below is unaffected, + // since those datagrams carry no flow data. + CHECK(event_data.num_events->value() == 5); + CHECK(event_data.num_samples->value() == 5); nlohmann::json j; flow_handler.metrics()->bucket(0)->to_json(j); @@ -509,9 +512,12 @@ TEST_CASE("Parse IPFIX stream", "[netflow][flow]") auto event_data = flow_handler.metrics()->bucket(0)->event_data_locked(); - // confirmed with wireshark - CHECK(event_data.num_events->value() == 23); - CHECK(event_data.num_samples->value() == 23); + // ipfix.pcap has 43 total datagrams; 20 are template/options-only + // (zero flow records) and were previously miscounted as parse errors + // instead of successfully ingested samples. records_flows below is + // unaffected, since those datagrams carry no flow data. + CHECK(event_data.num_events->value() == 43); + CHECK(event_data.num_samples->value() == 43); nlohmann::json j; flow_handler.metrics()->bucket(0)->to_json(j); @@ -546,9 +552,12 @@ TEST_CASE("Parse IPFIX stream with subnet summary wildcard", "[netflow][flow]") auto event_data = flow_handler.metrics()->bucket(0)->event_data_locked(); - // confirmed with wireshark - CHECK(event_data.num_events->value() == 23); - CHECK(event_data.num_samples->value() == 23); + // ipfix.pcap has 43 total datagrams; 20 are template/options-only + // (zero flow records) and were previously miscounted as parse errors + // instead of successfully ingested samples. records_flows below is + // unaffected, since those datagrams carry no flow data. + CHECK(event_data.num_events->value() == 43); + CHECK(event_data.num_samples->value() == 43); nlohmann::json j; flow_handler.metrics()->bucket(0)->to_json(j); @@ -830,3 +839,61 @@ TEST_CASE("Parse sflow IPv4/IPv6 keyed sample elements", "[sflow][flow]") CHECK(iface["in_tcp_bytes"] == 1500); CHECK(iface["in_udp_bytes"] == 1280); } + +TEST_CASE("Parse netflow v9 stream with options flowset", "[netflow][flow]") +{ + // Fixture carries a datagram with a real data flowset alongside an + // options flowset that has no registered data template (the real flow + // record must survive), followed by a pure template-refresh datagram + // with zero data flowsets (must not be a parse error). + FlowInputStream stream{"netflow-v9-options-test"}; + stream.config_set("flow_type", "netflow"); + stream.config_set("pcap_file", "tests/fixtures/nf9_options.pcap"); + + visor::Config c; + auto stream_proxy = stream.add_event_proxy(c); + c.config_set("num_periods", 1); + FlowStreamHandler flow_handler{"flow-v9-options", stream_proxy, &c}; + + flow_handler.start(); + stream.start(); + stream.stop(); + flow_handler.stop(); + + nlohmann::json info; + stream.info_json(info); + CHECK(info["flow"]["packet_errors"] == 0); + + nlohmann::json j; + flow_handler.metrics()->bucket(0)->to_json(j); + // Only the first datagram carries a flow record; the second is + // template-only. + CHECK(j["devices"]["10.0.0.5"]["records_flows"] == 1); +} + +TEST_CASE("Parse IPFIX stream with options set", "[netflow][flow]") +{ + // Same layout as above, using IPFIX (v10) set IDs instead of NetFlow v9 + // flowset IDs. + FlowInputStream stream{"ipfix-options-test"}; + stream.config_set("flow_type", "netflow"); + stream.config_set("pcap_file", "tests/fixtures/ipfix_options.pcap"); + + visor::Config c; + auto stream_proxy = stream.add_event_proxy(c); + c.config_set("num_periods", 1); + FlowStreamHandler flow_handler{"flow-ipfix-options", stream_proxy, &c}; + + flow_handler.start(); + stream.start(); + stream.stop(); + flow_handler.stop(); + + nlohmann::json info; + stream.info_json(info); + CHECK(info["flow"]["packet_errors"] == 0); + + nlohmann::json j; + flow_handler.metrics()->bucket(0)->to_json(j); + CHECK(j["devices"]["10.0.0.7"]["records_flows"] == 1); +} diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index a6e6bfc08..99a45938d 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -55,6 +55,14 @@ struct hash_pair { using NfMapID = std::pair; static robin_hood::unordered_node_map nf9_template_map; static robin_hood::unordered_node_map nf10_template_map; +/* Options templates (v9 flowset id 1 / IPFIX set id 3) describe metadata + * records (sampling rate, interface descriptions, etc.), not flow data. + * pktvisor doesn't decode option field values today, but registering these + * templates lets a subsequent options-data flowset/set be recognized as + * such -- rather than falling through the same "unknown template" path as + * a genuinely missing data template -- and skipped accordingly. */ +static robin_hood::unordered_node_map nf9_options_template_map; +static robin_hood::unordered_node_map nf10_options_template_map; static bool process_netflow_v1(NFSample *sample) { @@ -278,6 +286,61 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, uint32_t sourc return true; } +/* Options Template FlowSet Format (RFC 3954 section 8): FlowSet ID = 1, + * followed by one or more option template records, each: + * Template ID (2), Option Scope Length (2), Option Length (2), + * then (Option Scope Length / 4) scope field specs, then + * (Option Length / 4) option field specs -- each field spec is a + * type (2) + length (2) pair, same shape as a regular template's field + * specs. We only need the combined byte length of one options-data + * record, so scope and option fields are stored together. */ +static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, uint32_t source_id) +{ + struct NF9_FLOWSET_HEADER_COMMON *options_header; + uint16_t template_id; + uint32_t i, scope_len, option_len, num_fields, offset, total_size; + + options_header = reinterpret_cast(pkt); + if (len < sizeof(*options_header)) { + return false; + } + + if (be16toh(options_header->flowset_id) != NF9_OPTIONS_FLOWSET_ID) { + return false; + } + + for (offset = sizeof(*options_header); offset < len;) { + if (offset + 6 > len) { + return false; + } + + template_id = be16toh(*reinterpret_cast(pkt + offset)); + scope_len = be16toh(*reinterpret_cast(pkt + offset + 2)); + option_len = be16toh(*reinterpret_cast(pkt + offset + 4)); + offset += 6; + + num_fields = (scope_len + option_len) / 4; + total_size = 0; + std::vector option_recs; + for (i = 0; i < num_fields; i++) { + if (offset + 4 > len) { + return false; + } + + uint16_t field_type = be16toh(*reinterpret_cast(pkt + offset)); + uint16_t field_length = be16toh(*reinterpret_cast(pkt + offset + 2)); + offset += 4; + + option_recs.emplace_back(field_type, field_length); + total_size += field_length; + } + + nf9_options_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, num_fields, total_size, option_recs}; + } + + return true; +} + static void nf9_rec_to_flow(NFSample::Flows *flow, struct peer_nf9_record *rec, uint8_t *data) { /* XXX: use a table-based interpreter */ @@ -393,7 +456,7 @@ static bool process_netflow_v9(NFSample *sample) struct NF9_FLOWSET_HEADER_COMMON *flowset; uint32_t i, flowset_id, flowset_len, flowset_flows; - uint32_t offset, total_flows; + uint32_t offset; sample->uptime_ms = be32toh(nf9_hdr->uptime_ms); sample->time_sec = be32toh(nf9_hdr->time_sec); @@ -401,7 +464,6 @@ static bool process_netflow_v9(NFSample *sample) sample->source_id = be32toh(nf9_hdr->source_id); offset = sizeof(*nf9_hdr); - total_flows = 0; std::vector flows; for (i = 0;; i++) { @@ -426,17 +488,31 @@ static bool process_netflow_v9(NFSample *sample) } break; case NF9_OPTIONS_FLOWSET_ID: - /* XXX: implement this (maybe) */ + if (!process_netflow_v9_options_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { + /* XXX ratelimit */ + } break; default: if (flowset_id < NF9_MIN_RECORD_FLOWSET_ID) { /* XXX ratelimit */ break; } + if (nf9_options_template_map.find(NfMapID(sample->source_id, flowset_id)) != nf9_options_template_map.end()) { + /* Recognized options-data flowset (per a template registered + * above): it carries metadata, not flow records, so there is + * nothing to decode here. It is skipped below like any other + * flowset, by its declared length. */ + break; + } + /* A data flowset whose template id we don't have (e.g. because + * its template announcement was missed) should not abort the + * whole datagram: skip just this flowset and keep processing + * the rest of the packet, since any real flow data elsewhere in + * the same datagram is still valid. */ if (!process_netflow_v9_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { - return false; + /* XXX ratelimit */ + break; } - total_flows += flowset_flows; break; } offset += flowset_len; @@ -447,10 +523,12 @@ static bool process_netflow_v9(NFSample *sample) sample->flows = flows; - if (total_flows > 0) { - return true; - } - return false; + /* A structurally parseable datagram that happens to carry zero flow + * records (a pure template/options refresh, or one where every data + * flowset was skipped above) is not a parse error -- only the header + * sanity check above should fail this function. Previously this + * inflated packet_errors on legitimate exporter behavior. */ + return true; } static void nf10_rec_to_flow(NFSample::Flows *flow, struct peer_nf10_record *rec, uint8_t *data) @@ -606,6 +684,64 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, uint32_t sour return true; } +/* Options Template Record format (RFC 7011 section 3.4.2.2): Set ID = 3, + * followed by one or more option template records, each: + * Template ID (2), Field Count (2), Scope Field Count (2), then + * Field Count total field specs (type (2) + length (2), plus a 4-byte + * enterprise number when the enterprise bit is set) -- the first Scope + * Field Count of which are scope fields. As with process_netflow_v9_ + * options_template, we only need the combined record byte length, so + * scope and non-scope fields are stored together. */ +static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, uint32_t source_id) +{ + struct NF10_FLOWSET_HEADER_COMMON *options_header; + uint16_t template_id; + uint32_t i, field_count, offset, total_size; + + options_header = reinterpret_cast(pkt); + if (len < sizeof(*options_header)) { + return false; + } + + if (be16toh(options_header->flowset_id) != NF10_OPTIONS_FLOWSET_ID) { + return false; + } + + for (offset = sizeof(*options_header); offset < len;) { + if (offset + 6 > len) { + return false; + } + + template_id = be16toh(*reinterpret_cast(pkt + offset)); + field_count = be16toh(*reinterpret_cast(pkt + offset + 2)); + /* Scope Field Count (pkt + offset + 4) isn't needed: we don't + * distinguish scope from non-scope fields for a byte-length tally. */ + offset += 6; + + total_size = 0; + std::vector option_recs; + for (i = 0; i < field_count; i++) { + if (offset + 4 > len) { + return false; + } + + uint32_t rec_type = be16toh(*reinterpret_cast(pkt + offset)); + uint32_t rec_length = be16toh(*reinterpret_cast(pkt + offset + 2)); + offset += 4; + if (rec_type & NF10_ENTERPRISE) { + offset += sizeof(uint32_t); + } + + option_recs.emplace_back(rec_type, rec_length); + (rec_length == 0xFFFF) ? total_size++ : total_size += rec_length; // i.e. variable length + } + + nf10_options_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, field_count, total_size, option_recs}; + } + + return true; +} + static bool process_netflow_v10(NFSample *sample) { struct NF10_HEADER *nf10_hdr = reinterpret_cast(sample->raw_sample); @@ -616,14 +752,13 @@ static bool process_netflow_v10(NFSample *sample) struct NF10_FLOWSET_HEADER_COMMON *flowset; uint32_t i, flowset_id, flowset_len, flowset_flows; - uint32_t offset, total_flows; + uint32_t offset; sample->time_sec = be32toh(nf10_hdr->time_sec); sample->flow_sequence = be32toh(nf10_hdr->package_sequence); sample->source_id = be32toh(nf10_hdr->source_id); offset = sizeof(*nf10_hdr); - total_flows = 0; std::vector flows; for (i = 0;; i++) { @@ -648,17 +783,31 @@ static bool process_netflow_v10(NFSample *sample) } break; case NF10_OPTIONS_FLOWSET_ID: - /* XXX: implement this (maybe) */ + if (!process_netflow_v10_options_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { + /* XXX ratelimit */ + } break; default: if (flowset_id < NF10_MIN_RECORD_FLOWSET_ID) { /* XXX ratelimit */ break; } + if (nf10_options_template_map.find(NfMapID(sample->source_id, flowset_id)) != nf10_options_template_map.end()) { + /* Recognized options-data set (per a template registered + * above): it carries metadata, not flow records, so there is + * nothing to decode here. It is skipped below like any other + * set, by its declared length. */ + break; + } + /* A data set whose template id we don't have (e.g. because its + * template announcement was missed) should not abort the whole + * datagram: skip just this set and keep processing the rest of + * the message, since any real flow data elsewhere in the same + * datagram is still valid. */ if (!process_netflow_v10_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { - return false; + /* XXX ratelimit */ + break; } - total_flows += flowset_flows; break; } offset += flowset_len; @@ -669,10 +818,12 @@ static bool process_netflow_v10(NFSample *sample) sample->flows = flows; - if (total_flows > 0) { - return true; - } - return false; + /* A structurally parseable message that happens to carry zero flow + * records (a pure template/options refresh, or one where every data + * set was skipped above) is not a parse error -- only the header + * sanity check above should fail this function. Previously this + * inflated packet_errors on legitimate exporter behavior. */ + return true; } static bool process_netflow_packet(NFSample *sample) diff --git a/src/inputs/flow/test_flow.cpp b/src/inputs/flow/test_flow.cpp index bef5e9ba1..421428295 100644 --- a/src/inputs/flow/test_flow.cpp +++ b/src/inputs/flow/test_flow.cpp @@ -122,3 +122,35 @@ TEST_CASE("sflow udp socket without bind", "[flow][sflow][udp]") CHECK_THROWS_WITH(stream.start(), "flow config must specify port and bind"); } + +TEST_CASE("netflow v9 with options flowset does not cause packet errors", "[flow][netflow][options-flowset]") +{ + // Fixture carries two datagrams: one with a real data flowset alongside + // an options flowset (options template id 256) that pktvisor now + // parses and recognizes, and one that is a pure template refresh with + // zero data flowsets. Neither should be counted as a parse error. + FlowInputStream stream{"netflow-options-test"}; + stream.config_set("pcap_file", "tests/fixtures/nf9_options.pcap"); + stream.config_set("flow_type", "netflow"); + + CHECK_NOTHROW(stream.start()); + CHECK_NOTHROW(stream.stop()); + + nlohmann::json j; + stream.info_json(j); + CHECK(j["flow"]["packet_errors"] == 0); +} + +TEST_CASE("ipfix with options set does not cause packet errors", "[flow][netflow][options-flowset]") +{ + FlowInputStream stream{"ipfix-options-test"}; + stream.config_set("pcap_file", "tests/fixtures/ipfix_options.pcap"); + stream.config_set("flow_type", "netflow"); + + CHECK_NOTHROW(stream.start()); + CHECK_NOTHROW(stream.stop()); + + nlohmann::json j; + stream.info_json(j); + CHECK(j["flow"]["packet_errors"] == 0); +} diff --git a/src/tests/fixtures/ipfix_options.pcap b/src/tests/fixtures/ipfix_options.pcap new file mode 100644 index 0000000000000000000000000000000000000000..581153fcf1a252a125ba71ce24971b400c9c8f3b GIT binary patch literal 266 zcmca|c+)~A1{MYw`2U}Qff2?5(mhbTYWJDD?-;=(69%)kd! z2;u=Tki`Sk2?ju25~mpo<$zct+>UWK0|Vy~po9oin!ydo0AUQbSzvXW3Q#G~p-67K O478tzdneFdpv?f+`za&< literal 0 HcmV?d00001 diff --git a/src/tests/fixtures/nf9_options.pcap b/src/tests/fixtures/nf9_options.pcap new file mode 100644 index 0000000000000000000000000000000000000000..9c449ec60df75c51b9cc15393deaab6132228131 GIT binary patch literal 278 zcmca|c+)~A1{MYw`2U}Qff2?5(o>*#)$TKQ-!XzoCJqKy28K2v69xtcK?Vjc1_o9j zW@~iiU}q=-D(8e4kTF~50FVbXjQJIiR^Z;jz{bDzWVi5yN`3XP^?*Ju$mO_9RkOE;0H`-uzqXtkZCqyTb Q8(#xOc(`{0?FQNo01pu<9{>OV literal 0 HcmV?d00001 From ad57171e065407d7397c5d10207b0c3145e6a413 Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Thu, 20 Aug 2026 14:04:18 -0700 Subject: [PATCH 02/12] fix(inputs/flow): don't spin on zero-length flowsets, clear stale options/data template classification on id reuse, and only skip (not fail) unknown-template data flowsets Addresses three review findings: a FlowSet/Set shorter than its own 4-byte header could never advance the parse cursor, spinning forever; reassigning a template id from options to data (or back) left a stale classification in the other map, silently dropping real flow data forever; and every process_netflow_v9_data/v10_data failure was being treated as skippable, which also swallowed genuine decode failures for known templates instead of failing the datagram. --- src/inputs/flow/NetflowData.h | 73 +++++++++++++++++++++++++++++------ 1 file changed, 61 insertions(+), 12 deletions(-) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index 99a45938d..02fe70848 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -281,6 +281,12 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, uint32_t sourc } nf9_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; + /* This id may have previously been registered as an options + * template (e.g. the exporter withdrew/reassigned it). A stale + * options-map entry would otherwise take precedence over this + * new data template forever, silently skipping every future + * data flowset with this id. */ + nf9_options_template_map.erase(NfMapID(source_id, template_id)); } return true; @@ -336,6 +342,9 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, uint32 } nf9_options_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, num_fields, total_size, option_recs}; + /* Mirror image of the erase in process_netflow_v9_template: this + * id may have previously been a data template. */ + nf9_template_map.erase(NfMapID(source_id, template_id)); } return true; @@ -481,6 +490,15 @@ static bool process_netflow_v9(NFSample *sample) break; } + /* A FlowSet/Set must be at least as long as its own fixed header + * (id + length). Anything shorter -- notably a declared length of + * 0 -- can never advance offset below, which would otherwise spin + * on the same bytes forever. Treat it as the end of usable data + * in this datagram rather than a hang. */ + if (flowset_len < sizeof(*flowset)) { + break; + } + switch (flowset_id) { case NF9_TEMPLATE_FLOWSET_ID: if (!process_netflow_v9_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { @@ -504,15 +522,26 @@ static bool process_netflow_v9(NFSample *sample) * flowset, by its declared length. */ break; } - /* A data flowset whose template id we don't have (e.g. because - * its template announcement was missed) should not abort the - * whole datagram: skip just this flowset and keep processing - * the rest of the packet, since any real flow data elsewhere in - * the same datagram is still valid. */ - if (!process_netflow_v9_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { + if (nf9_template_map.find(NfMapID(sample->source_id, flowset_id)) == nf9_template_map.end()) { + /* A data flowset whose template id we don't have (e.g. + * because its template announcement was missed) should + * not abort the whole datagram: skip just this flowset + * and keep processing the rest of the packet, since any + * real flow data elsewhere in the same datagram is still + * valid. */ /* XXX ratelimit */ break; } + /* The template id IS known, so a decode failure here means + * this flowset's bytes don't match its own declared template + * (truncated data, a record count that doesn't evenly divide + * the flowset length, etc). That is genuinely malformed + * input, unlike the unknown-template case above -- fail the + * whole datagram as before so it's still counted as a parse + * error. */ + if (!process_netflow_v9_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { + return false; + } break; } offset += flowset_len; @@ -679,6 +708,9 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, uint32_t sour } nf10_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; + /* See the equivalent erase in process_netflow_v9_template: this + * id may have previously been registered as an options template. */ + nf10_options_template_map.erase(NfMapID(source_id, template_id)); } return true; @@ -737,6 +769,8 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, uint3 } nf10_options_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, field_count, total_size, option_recs}; + /* Mirror image of the erase in process_netflow_v10_template. */ + nf10_template_map.erase(NfMapID(source_id, template_id)); } return true; @@ -776,6 +810,14 @@ static bool process_netflow_v10(NFSample *sample) break; } + /* See the equivalent check in process_netflow_v9: a Set shorter + * than its own fixed header (id + length) -- notably a declared + * length of 0 -- can never advance offset below, which would + * otherwise spin on the same bytes forever. */ + if (flowset_len < sizeof(*flowset)) { + break; + } + switch (flowset_id) { case NF10_TEMPLATE_FLOWSET_ID: if (!process_netflow_v10_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { @@ -799,15 +841,22 @@ static bool process_netflow_v10(NFSample *sample) * set, by its declared length. */ break; } - /* A data set whose template id we don't have (e.g. because its - * template announcement was missed) should not abort the whole - * datagram: skip just this set and keep processing the rest of - * the message, since any real flow data elsewhere in the same - * datagram is still valid. */ - if (!process_netflow_v10_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { + if (nf10_template_map.find(NfMapID(sample->source_id, flowset_id)) == nf10_template_map.end()) { + /* A data set whose template id we don't have (e.g. + * because its template announcement was missed) should + * not abort the whole datagram: skip just this set and + * keep processing the rest of the message, since any + * real flow data elsewhere in the same datagram is still + * valid. */ /* XXX ratelimit */ break; } + /* The template id IS known, so a decode failure here is + * genuinely malformed input (see the matching comment in + * process_netflow_v9) -- fail the whole datagram as before. */ + if (!process_netflow_v10_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { + return false; + } break; } offset += flowset_len; From 1cd0a8e2979a2da8dc15ad59ce581aed7b08f8bb Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Fri, 21 Aug 2026 06:44:30 -0700 Subject: [PATCH 03/12] fix(inputs/flow): reject IPFIX field specs with a truncated enterprise number A field spec with the enterprise bit set carries an extra 4-byte enterprise number. If a Set was truncated right at that point, offset advanced past len with no check, and the loop would still exit normally -- so this malformed, truncated template got registered anyway, silently erasing a valid template of the other kind (data vs options) for the same id. Both the regular and options template paths had the identical gap; both are now rejected before they can commit. --- src/inputs/flow/NetflowData.h | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index 02fe70848..1ead39629 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -702,7 +702,17 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, uint32_t sour peer_nf10_record recs(rec_type, rec_length); offset += sizeof(*tmplr); - (rec_type & NF10_ENTERPRISE) ? offset += sizeof(uint32_t) : offset; + if (rec_type & NF10_ENTERPRISE) { + /* See the equivalent check in + * process_netflow_v10_options_template: a truncated + * enterprise number here would otherwise let this + * malformed template get registered below and erase a + * valid options-template entry for the same id. */ + if (offset + sizeof(uint32_t) > len) { + return false; + } + offset += sizeof(uint32_t); + } template_recs.push_back(recs); (rec_length == 0xFFFF) ? total_size++ : total_size += rec_length; // i.e. variable length } @@ -761,6 +771,18 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, uint3 uint32_t rec_length = be16toh(*reinterpret_cast(pkt + offset + 2)); offset += 4; if (rec_type & NF10_ENTERPRISE) { + /* A field spec with the enterprise bit set carries an + * extra 4-byte enterprise number. If the Set is + * truncated right at that point, advancing offset + * unconditionally would run past `len` -- and without + * this check, the loop would still exit normally and + * this malformed, truncated template would get + * registered below, silently discarding a valid data + * template for the same id (via the erase() below). + * Reject it instead. */ + if (offset + sizeof(uint32_t) > len) { + return false; + } offset += sizeof(uint32_t); } From 727c4cb6c845a9e90fc426189b3627826238412d Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Fri, 21 Aug 2026 12:18:58 -0700 Subject: [PATCH 04/12] fix(inputs/flow): key NetFlow/IPFIX template maps by exporter address, and reject an invalid IPFIX scope field count Two independent codex findings: - NfMapID was only (source_id, template_id), and both are chosen independently by each exporter, not assigned globally. Two different exporters picking the same values collided in the process-global template maps -- and could erase each other's valid templates via the data<->options erase() added earlier in this PR. The exporter's own network-layer address is now part of the key; FlowInputStream now populates NFSample::exporter_ip before parsing (both the pcap-replay and live-UDP paths already had this address available, just not threaded through). - process_netflow_v10_options_template read field_count but never validated scope_field_count against it, even though scope fields are defined as the first N of the total fields (RFC 7011 3.4.2.2) -- so scope_field_count can never legitimately exceed field_count. A template violating that was silently accepted and could likewise erase a valid data template for the same id. --- src/inputs/flow/FlowInputStream.cpp | 19 ++-- src/inputs/flow/NetflowData.h | 130 +++++++++++++++++++--------- 2 files changed, 102 insertions(+), 47 deletions(-) diff --git a/src/inputs/flow/FlowInputStream.cpp b/src/inputs/flow/FlowInputStream.cpp index 1c54914e5..f6f9de3a7 100644 --- a/src/inputs/flow/FlowInputStream.cpp +++ b/src/inputs/flow/FlowInputStream.cpp @@ -106,16 +106,19 @@ void FlowInputStream::_read_from_pcap_file() NFSample sample; sample.raw_sample = udpLayer->getLayerPayload(); sample.raw_sample_len = udpLayer->getLayerPayloadSize(); + // Extract the exporter's address before parsing (not just + // for the callback below) -- process_netflow_packet uses it + // as part of the template-map key, so two different + // exporters replayed from the same pcap can't collide. + if (auto IP4layer = netflow_pkt.getLayerOfType(); IP4layer) { + sample.exporter_ip = IP4layer->getSrcIPv4Address().toString(); + } else if (auto IP6layer = netflow_pkt.getLayerOfType(); IP6layer) { + sample.exporter_ip = IP6layer->getSrcIPv6Address().toString(); + } if (process_netflow_packet(&sample)) { - std::string src_ip; - if (auto IP4layer = netflow_pkt.getLayerOfType(); IP4layer) { - src_ip = IP4layer->getSrcIPv4Address().toString(); - } else if (auto IP6layer = netflow_pkt.getLayerOfType(); IP6layer) { - src_ip = IP6layer->getSrcIPv6Address().toString(); - } std::shared_lock lock(_input_mutex); for (auto &proxy : _event_proxies) { - static_cast(proxy.get())->netflow_cb(src_ip, sample, rawPacket.getRawDataLen()); + static_cast(proxy.get())->netflow_cb(sample.exporter_ip, sample, rawPacket.getRawDataLen()); } } else { _logger->error("invalid netflow or ipfix packet"); @@ -213,6 +216,8 @@ void FlowInputStream::_create_frame_stream_udp_socket() NFSample sample; sample.raw_sample = reinterpret_cast(event.data.get()); sample.raw_sample_len = event.length; + // See the equivalent note in the pcap-file path above. + sample.exporter_ip = event.sender.ip; if (process_netflow_packet(&sample)) { std::shared_lock lock(_input_mutex); for (auto &proxy : _event_proxies) { diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index 1ead39629..4c9500cd8 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -7,6 +7,7 @@ #include "EndianPortable.h" #include #include +#include #include namespace visor::input::flow { @@ -27,6 +28,19 @@ struct NFSample { uint32_t flow_sequence{0}; uint32_t source_id{0}; + /* The exporter's own network-layer (UDP sender) address, e.g. + * "10.0.0.5". source_id (NetFlow Source ID / IPFIX Observation + * Domain ID) and template ids are each chosen independently by + * every exporter and are not globally unique, so this is required + * as part of the template-map key below to keep two different + * exporters that happen to pick the same source_id/template_id + * from colliding in these process-global maps. Populated by the + * caller (FlowInputStream) before process_netflow_packet() is + * called; left empty (e.g. from a synthetic/test sample) is safe -- + * it just means all such samples share one bucket, matching the + * pre-existing behavior. */ + std::string exporter_ip; + struct Flows { bool is_ipv6 = false; std::array src_ip{}, dst_ip{}, nexthop_ip{}; @@ -41,28 +55,53 @@ struct NFSample { std::vector flows; }; -// A hash function used to hash a pair of any kind -struct hash_pair { - template - size_t operator()(const std::pair &p) const +/* Key for the template maps below. source_id (NetFlow Source ID / IPFIX + * Observation Domain ID) and template ids are each chosen independently + * by every exporter and are not globally unique on their own -- two + * different physical exporters can legitimately pick the same + * source_id/template_id pair, so the exporter's own address is folded + * into the key to keep them from colliding in these process-global + * maps. */ +struct NfMapID { + std::string exporter_ip; + uint32_t source_id; + uint16_t template_id; + + NfMapID(std::string exporter_ip_, uint32_t source_id_, uint16_t template_id_) + : exporter_ip(std::move(exporter_ip_)) + , source_id(source_id_) + , template_id(template_id_) + { + } + + bool operator==(const NfMapID &other) const + { + return exporter_ip == other.exporter_ip + && source_id == other.source_id + && template_id == other.template_id; + } +}; + +struct hash_nf_map_id { + size_t operator()(const NfMapID &k) const { - auto hash1 = std::hash{}(p.first); - auto hash2 = std::hash{}(p.second); - return hash1 ^ hash2; + size_t h1 = std::hash{}(k.exporter_ip); + size_t h2 = std::hash{}(k.source_id); + size_t h3 = std::hash{}(k.template_id); + return h1 ^ (h2 << 1) ^ (h3 << 2); } }; -using NfMapID = std::pair; -static robin_hood::unordered_node_map nf9_template_map; -static robin_hood::unordered_node_map nf10_template_map; +static robin_hood::unordered_node_map nf9_template_map; +static robin_hood::unordered_node_map nf10_template_map; /* Options templates (v9 flowset id 1 / IPFIX set id 3) describe metadata * records (sampling rate, interface descriptions, etc.), not flow data. * pktvisor doesn't decode option field values today, but registering these * templates lets a subsequent options-data flowset/set be recognized as * such -- rather than falling through the same "unknown template" path as * a genuinely missing data template -- and skipped accordingly. */ -static robin_hood::unordered_node_map nf9_options_template_map; -static robin_hood::unordered_node_map nf10_options_template_map; +static robin_hood::unordered_node_map nf9_options_template_map; +static robin_hood::unordered_node_map nf10_options_template_map; static bool process_netflow_v1(NFSample *sample) { @@ -240,7 +279,7 @@ static inline void be_copy(uint8_t *data, uint8_t *target, uint32_t target_lengt } } -static bool process_netflow_v9_template(uint8_t *pkt, size_t len, uint32_t source_id) +static bool process_netflow_v9_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) { struct NF9_FLOWSET_HEADER_COMMON *template_header; struct NF9_TEMPLATE_FLOWSET_HEADER *tmplh; @@ -280,13 +319,13 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, uint32_t sourc total_size += rec_length; } - nf9_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; + nf9_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; /* This id may have previously been registered as an options * template (e.g. the exporter withdrew/reassigned it). A stale * options-map entry would otherwise take precedence over this * new data template forever, silently skipping every future * data flowset with this id. */ - nf9_options_template_map.erase(NfMapID(source_id, template_id)); + nf9_options_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); } return true; @@ -300,7 +339,7 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, uint32_t sourc * type (2) + length (2) pair, same shape as a regular template's field * specs. We only need the combined byte length of one options-data * record, so scope and option fields are stored together. */ -static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, uint32_t source_id) +static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) { struct NF9_FLOWSET_HEADER_COMMON *options_header; uint16_t template_id; @@ -341,10 +380,10 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, uint32 total_size += field_length; } - nf9_options_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, num_fields, total_size, option_recs}; + nf9_options_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, num_fields, total_size, option_recs}; /* Mirror image of the erase in process_netflow_v9_template: this * id may have previously been a data template. */ - nf9_template_map.erase(NfMapID(source_id, template_id)); + nf9_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); } return true; @@ -405,7 +444,7 @@ static void nf9_rec_to_flow(NFSample::Flows *flow, struct peer_nf9_record *rec, } } -static bool process_netflow_v9_data(std::vector *flows, uint8_t *pkt, size_t len, uint32_t source_id, uint32_t &num_flows) +static bool process_netflow_v9_data(std::vector *flows, uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id, uint32_t &num_flows) { struct NF9_DATA_FLOWSET_HEADER *dath; uint16_t flowset_id, offset, num_flowsets; @@ -423,7 +462,7 @@ static bool process_netflow_v9_data(std::vector *flows, uint8_t flowset_id = be16toh(dath->c.flowset_id); - auto iter = nf9_template_map.find(NfMapID(source_id, flowset_id)); + auto iter = nf9_template_map.find(NfMapID(exporter_ip, source_id, flowset_id)); if (iter == nf9_template_map.end()) { return false; } @@ -501,12 +540,12 @@ static bool process_netflow_v9(NFSample *sample) switch (flowset_id) { case NF9_TEMPLATE_FLOWSET_ID: - if (!process_netflow_v9_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { + if (!process_netflow_v9_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { return false; } break; case NF9_OPTIONS_FLOWSET_ID: - if (!process_netflow_v9_options_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { + if (!process_netflow_v9_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { /* XXX ratelimit */ } break; @@ -515,14 +554,14 @@ static bool process_netflow_v9(NFSample *sample) /* XXX ratelimit */ break; } - if (nf9_options_template_map.find(NfMapID(sample->source_id, flowset_id)) != nf9_options_template_map.end()) { + if (nf9_options_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) != nf9_options_template_map.end()) { /* Recognized options-data flowset (per a template registered * above): it carries metadata, not flow records, so there is * nothing to decode here. It is skipped below like any other * flowset, by its declared length. */ break; } - if (nf9_template_map.find(NfMapID(sample->source_id, flowset_id)) == nf9_template_map.end()) { + if (nf9_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) == nf9_template_map.end()) { /* A data flowset whose template id we don't have (e.g. * because its template announcement was missed) should * not abort the whole datagram: skip just this flowset @@ -539,7 +578,7 @@ static bool process_netflow_v9(NFSample *sample) * input, unlike the unknown-template case above -- fail the * whole datagram as before so it's still counted as a parse * error. */ - if (!process_netflow_v9_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { + if (!process_netflow_v9_data(&flows, sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id, flowset_flows)) { return false; } break; @@ -615,7 +654,7 @@ static void nf10_rec_to_flow(NFSample::Flows *flow, struct peer_nf10_record *rec } } -static bool process_netflow_v10_data(std::vector *flows, uint8_t *pkt, size_t len, uint32_t source_id, uint32_t &num_flows) +static bool process_netflow_v10_data(std::vector *flows, uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id, uint32_t &num_flows) { struct NF10_DATA_FLOWSET_HEADER *dath; uint16_t flowset_id, offset, num_flowsets; @@ -632,7 +671,7 @@ static bool process_netflow_v10_data(std::vector *flows, uint8_ } flowset_id = be16toh(dath->c.flowset_id); - auto iter = nf10_template_map.find(NfMapID(source_id, flowset_id)); + auto iter = nf10_template_map.find(NfMapID(exporter_ip, source_id, flowset_id)); if (iter == nf10_template_map.end()) { return false; } @@ -665,7 +704,7 @@ static bool process_netflow_v10_data(std::vector *flows, uint8_ return true; } -static bool process_netflow_v10_template(uint8_t *pkt, size_t len, uint32_t source_id) +static bool process_netflow_v10_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) { struct NF10_FLOWSET_HEADER_COMMON *template_header; struct NF10_TEMPLATE_FLOWSET_HEADER *tmplh; @@ -717,10 +756,10 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, uint32_t sour (rec_length == 0xFFFF) ? total_size++ : total_size += rec_length; // i.e. variable length } - nf10_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; + nf10_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; /* See the equivalent erase in process_netflow_v9_template: this * id may have previously been registered as an options template. */ - nf10_options_template_map.erase(NfMapID(source_id, template_id)); + nf10_options_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); } return true; @@ -734,7 +773,7 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, uint32_t sour * Field Count of which are scope fields. As with process_netflow_v9_ * options_template, we only need the combined record byte length, so * scope and non-scope fields are stored together. */ -static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, uint32_t source_id) +static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) { struct NF10_FLOWSET_HEADER_COMMON *options_header; uint16_t template_id; @@ -756,9 +795,20 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, uint3 template_id = be16toh(*reinterpret_cast(pkt + offset)); field_count = be16toh(*reinterpret_cast(pkt + offset + 2)); - /* Scope Field Count (pkt + offset + 4) isn't needed: we don't - * distinguish scope from non-scope fields for a byte-length tally. */ + /* Scope Field Count: per RFC 7011 section 3.4.2.2, an Options + * Template's scope fields are defined as the first + * scope_field_count of its field_count total field specifiers, + * so scope_field_count can never exceed field_count. We don't + * otherwise distinguish scope from non-scope fields (both are + * folded into one byte-length tally below), but a value that + * violates this invariant means the record is malformed -- + * reject it rather than silently reading past where the scope + * fields actually end. */ + uint32_t scope_field_count = be16toh(*reinterpret_cast(pkt + offset + 4)); offset += 6; + if (scope_field_count > field_count) { + return false; + } total_size = 0; std::vector option_recs; @@ -790,9 +840,9 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, uint3 (rec_length == 0xFFFF) ? total_size++ : total_size += rec_length; // i.e. variable length } - nf10_options_template_map[NfMapID(source_id, template_id)] = {template_id, source_id, field_count, total_size, option_recs}; + nf10_options_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, field_count, total_size, option_recs}; /* Mirror image of the erase in process_netflow_v10_template. */ - nf10_template_map.erase(NfMapID(source_id, template_id)); + nf10_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); } return true; @@ -842,12 +892,12 @@ static bool process_netflow_v10(NFSample *sample) switch (flowset_id) { case NF10_TEMPLATE_FLOWSET_ID: - if (!process_netflow_v10_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { + if (!process_netflow_v10_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { return false; } break; case NF10_OPTIONS_FLOWSET_ID: - if (!process_netflow_v10_options_template(sample->raw_sample + offset, flowset_len, sample->source_id)) { + if (!process_netflow_v10_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { /* XXX ratelimit */ } break; @@ -856,14 +906,14 @@ static bool process_netflow_v10(NFSample *sample) /* XXX ratelimit */ break; } - if (nf10_options_template_map.find(NfMapID(sample->source_id, flowset_id)) != nf10_options_template_map.end()) { + if (nf10_options_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) != nf10_options_template_map.end()) { /* Recognized options-data set (per a template registered * above): it carries metadata, not flow records, so there is * nothing to decode here. It is skipped below like any other * set, by its declared length. */ break; } - if (nf10_template_map.find(NfMapID(sample->source_id, flowset_id)) == nf10_template_map.end()) { + if (nf10_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) == nf10_template_map.end()) { /* A data set whose template id we don't have (e.g. * because its template announcement was missed) should * not abort the whole datagram: skip just this set and @@ -876,7 +926,7 @@ static bool process_netflow_v10(NFSample *sample) /* The template id IS known, so a decode failure here is * genuinely malformed input (see the matching comment in * process_netflow_v9) -- fail the whole datagram as before. */ - if (!process_netflow_v10_data(&flows, sample->raw_sample + offset, flowset_len, sample->source_id, flowset_flows)) { + if (!process_netflow_v10_data(&flows, sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id, flowset_flows)) { return false; } break; From 41096c0fc9ef7845282ae188a8b4a4be78ad5bf0 Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Mon, 24 Aug 2026 06:58:50 -0700 Subject: [PATCH 05/12] fix(inputs/flow): make options-template registration transactional and validate v9 field-list lengths --- src/inputs/flow/NetflowData.h | 56 +++++++++++++++++++++++++++++++---- 1 file changed, 50 insertions(+), 6 deletions(-) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index 4c9500cd8..d7cbfedce 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -354,6 +354,17 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const return false; } + /* Parse every record in this FlowSet into a local buffer first, without + * touching the process-global template maps. A single Options Template + * FlowSet can carry more than one record; if an earlier record were + * committed immediately and a later record in the same FlowSet turned + * out to be truncated or malformed, the erase() below would already + * have discarded a legitimate data template for that id before this + * function could report the failure -- corrupting exporter state for + * every future datagram, not just this one. Only commit to the maps + * once the whole FlowSet is known to be well-formed. */ + std::vector> pending; + for (offset = sizeof(*options_header); offset < len;) { if (offset + 6 > len) { return false; @@ -364,6 +375,16 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const option_len = be16toh(*reinterpret_cast(pkt + offset + 4)); offset += 6; + /* Option Scope Length and Option Length each delimit a list of + * 4-byte field specifiers, so each must independently be a + * multiple of 4. Summing them before dividing would accept an + * invalid split (e.g. 2 + 2) as one well-formed field, letting a + * malformed record consume bytes belonging to the next record or + * trailing padding. */ + if (scope_len % 4 != 0 || option_len % 4 != 0) { + return false; + } + num_fields = (scope_len + option_len) / 4; total_size = 0; std::vector option_recs; @@ -380,10 +401,14 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const total_size += field_length; } - nf9_options_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, num_fields, total_size, option_recs}; + pending.emplace_back(template_id, peer_nf9_template{template_id, source_id, num_fields, total_size, option_recs}); + } + + for (auto &p : pending) { + nf9_options_template_map[NfMapID(exporter_ip, source_id, p.first)] = p.second; /* Mirror image of the erase in process_netflow_v9_template: this * id may have previously been a data template. */ - nf9_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); + nf9_template_map.erase(NfMapID(exporter_ip, source_id, p.first)); } return true; @@ -545,8 +570,12 @@ static bool process_netflow_v9(NFSample *sample) } break; case NF9_OPTIONS_FLOWSET_ID: + /* A malformed options template (fails internally after + * validating its own bytes) is genuinely bad input, the same + * as a malformed regular template above -- fail the whole + * datagram rather than silently treating it as skippable. */ if (!process_netflow_v9_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { - /* XXX ratelimit */ + return false; } break; default: @@ -788,6 +817,14 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const return false; } + /* See the equivalent comment in process_netflow_v9_options_template: + * parse every record in this FlowSet into a local buffer first, and + * only commit to the process-global template maps once the whole Set + * is known to be well-formed, so a truncated later record can't leave + * an earlier record's registration (and its erase() of a legitimate + * data template) applied while this function still reports failure. */ + std::vector> pending; + for (offset = sizeof(*options_header); offset < len;) { if (offset + 6 > len) { return false; @@ -840,9 +877,13 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const (rec_length == 0xFFFF) ? total_size++ : total_size += rec_length; // i.e. variable length } - nf10_options_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, field_count, total_size, option_recs}; + pending.emplace_back(template_id, peer_nf10_template{template_id, source_id, field_count, total_size, option_recs}); + } + + for (auto &p : pending) { + nf10_options_template_map[NfMapID(exporter_ip, source_id, p.first)] = p.second; /* Mirror image of the erase in process_netflow_v10_template. */ - nf10_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); + nf10_template_map.erase(NfMapID(exporter_ip, source_id, p.first)); } return true; @@ -897,8 +938,11 @@ static bool process_netflow_v10(NFSample *sample) } break; case NF10_OPTIONS_FLOWSET_ID: + /* See the equivalent comment in process_netflow_v9: a malformed + * options template is genuinely bad input -- fail the whole + * datagram rather than silently treating it as skippable. */ if (!process_netflow_v10_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { - /* XXX ratelimit */ + return false; } break; default: From 154be41ec6accbc19cf56ece36061ee5d5cffc44 Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Mon, 24 Aug 2026 07:15:51 -0700 Subject: [PATCH 06/12] fix(inputs/flow): don't fail the whole datagram on a malformed options flowset --- src/inputs/flow/NetflowData.h | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index d7cbfedce..7b91e6f9e 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -570,12 +570,16 @@ static bool process_netflow_v9(NFSample *sample) } break; case NF9_OPTIONS_FLOWSET_ID: - /* A malformed options template (fails internally after - * validating its own bytes) is genuinely bad input, the same - * as a malformed regular template above -- fail the whole - * datagram rather than silently treating it as skippable. */ + /* Unlike a malformed regular TEMPLATE flowset, a malformed or + * simply not-fully-understood OPTIONS flowset must not fail + * the whole datagram -- that reintroduces the exact bug this + * PR fixes (real exporters interleave options flowsets we + * don't fully model with real flow data in the same + * datagram). process_netflow_v9_options_template's map + * updates are transactional, so a failure here is guaranteed + * to have committed nothing; it's safe to just move on. */ if (!process_netflow_v9_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { - return false; + /* XXX ratelimit */ } break; default: @@ -938,11 +942,12 @@ static bool process_netflow_v10(NFSample *sample) } break; case NF10_OPTIONS_FLOWSET_ID: - /* See the equivalent comment in process_netflow_v9: a malformed - * options template is genuinely bad input -- fail the whole - * datagram rather than silently treating it as skippable. */ + /* See the equivalent comment in process_netflow_v9: tolerate + * this rather than failing the datagram -- the transactional + * map updates in process_netflow_v10_options_template mean a + * failure here never leaves partial state behind. */ if (!process_netflow_v10_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { - return false; + /* XXX ratelimit */ } break; default: From c0e105aeceef90671d3c0eb1bd6d686412e7ecde Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Mon, 24 Aug 2026 10:17:59 -0700 Subject: [PATCH 07/12] fix(inputs/flow): fold listener/port into template key, tolerate options-template padding, fail truncated FlowSets --- src/inputs/flow/FlowInputStream.cpp | 21 +++- src/inputs/flow/NetflowData.h | 155 +++++++++++++++++++--------- 2 files changed, 125 insertions(+), 51 deletions(-) diff --git a/src/inputs/flow/FlowInputStream.cpp b/src/inputs/flow/FlowInputStream.cpp index f6f9de3a7..bc0aa1853 100644 --- a/src/inputs/flow/FlowInputStream.cpp +++ b/src/inputs/flow/FlowInputStream.cpp @@ -106,15 +106,19 @@ void FlowInputStream::_read_from_pcap_file() NFSample sample; sample.raw_sample = udpLayer->getLayerPayload(); sample.raw_sample_len = udpLayer->getLayerPayloadSize(); - // Extract the exporter's address before parsing (not just - // for the callback below) -- process_netflow_packet uses it - // as part of the template-map key, so two different - // exporters replayed from the same pcap can't collide. + // Extract the exporter's address and port before parsing + // (not just for the callback below) -- process_netflow_packet + // uses them (plus this stream's own listener_id) as part of + // the template-map key, so two different exporters replayed + // from the same pcap -- or through the same listener as + // another configured tap -- can't collide. if (auto IP4layer = netflow_pkt.getLayerOfType(); IP4layer) { sample.exporter_ip = IP4layer->getSrcIPv4Address().toString(); } else if (auto IP6layer = netflow_pkt.getLayerOfType(); IP6layer) { sample.exporter_ip = IP6layer->getSrcIPv6Address().toString(); } + sample.listener_id = _name; + sample.exporter_port = udpLayer->getSrcPort(); if (process_netflow_packet(&sample)) { std::shared_lock lock(_input_mutex); for (auto &proxy : _event_proxies) { @@ -216,8 +220,15 @@ void FlowInputStream::_create_frame_stream_udp_socket() NFSample sample; sample.raw_sample = reinterpret_cast(event.data.get()); sample.raw_sample_len = event.length; - // See the equivalent note in the pcap-file path above. + // See the equivalent note in the pcap-file path above: fold in + // this stream's own identity and the exporter's UDP source port, + // in addition to its address, so two independently configured + // taps in the same process (or two exporters behind the same + // NAT'd address) can't collide in the process-global template + // maps. sample.exporter_ip = event.sender.ip; + sample.listener_id = _name; + sample.exporter_port = static_cast(event.sender.port); if (process_netflow_packet(&sample)) { std::shared_lock lock(_input_mutex); for (auto &proxy : _event_proxies) { diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index 7b91e6f9e..b030ebe54 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -41,6 +41,19 @@ struct NFSample { * pre-existing behavior. */ std::string exporter_ip; + /* This input stream's own identity (its configured tap name) and the + * exporter's UDP source port. Without these, two independently + * configured listeners in the same process -- or two exporters that + * reach us through the same NAT'd address with colliding source_id/ + * template_id pairs -- can still present an identical key to the + * process-global template maps below and silently overwrite or erase + * each other's template registrations. Like exporter_ip above, + * leaving these at their defaults (empty/0) for a synthetic/test + * sample is safe -- it just means such samples share one bucket, + * matching pre-existing behavior. */ + std::string listener_id; + uint16_t exporter_port{0}; + struct Flows { bool is_ipv6 = false; std::array src_ip{}, dst_ip{}, nexthop_ip{}; @@ -59,26 +72,35 @@ struct NFSample { * Observation Domain ID) and template ids are each chosen independently * by every exporter and are not globally unique on their own -- two * different physical exporters can legitimately pick the same - * source_id/template_id pair, so the exporter's own address is folded - * into the key to keep them from colliding in these process-global - * maps. */ + * source_id/template_id pair. exporter_ip alone narrows that, but two + * independently configured listeners (taps) in the same process, or two + * exporters that reach us through the same NAT'd address, can still + * collide on (exporter_ip, source_id, template_id); listener_id and + * exporter_port are folded in as well to keep those cases from stomping + * each other's templates in these process-global maps. */ struct NfMapID { std::string exporter_ip; + std::string listener_id; uint32_t source_id; uint16_t template_id; + uint16_t exporter_port; - NfMapID(std::string exporter_ip_, uint32_t source_id_, uint16_t template_id_) + NfMapID(std::string exporter_ip_, std::string listener_id_, uint32_t source_id_, uint16_t template_id_, uint16_t exporter_port_) : exporter_ip(std::move(exporter_ip_)) + , listener_id(std::move(listener_id_)) , source_id(source_id_) , template_id(template_id_) + , exporter_port(exporter_port_) { } bool operator==(const NfMapID &other) const { return exporter_ip == other.exporter_ip + && listener_id == other.listener_id && source_id == other.source_id - && template_id == other.template_id; + && template_id == other.template_id + && exporter_port == other.exporter_port; } }; @@ -86,9 +108,11 @@ struct hash_nf_map_id { size_t operator()(const NfMapID &k) const { size_t h1 = std::hash{}(k.exporter_ip); - size_t h2 = std::hash{}(k.source_id); - size_t h3 = std::hash{}(k.template_id); - return h1 ^ (h2 << 1) ^ (h3 << 2); + size_t h2 = std::hash{}(k.listener_id); + size_t h3 = std::hash{}(k.source_id); + size_t h4 = std::hash{}(k.template_id); + size_t h5 = std::hash{}(k.exporter_port); + return h1 ^ (h2 << 1) ^ (h3 << 2) ^ (h4 << 3) ^ (h5 << 4); } }; @@ -279,7 +303,7 @@ static inline void be_copy(uint8_t *data, uint8_t *target, uint32_t target_lengt } } -static bool process_netflow_v9_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) +static bool process_netflow_v9_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, const std::string &listener_id, uint32_t source_id, uint16_t exporter_port) { struct NF9_FLOWSET_HEADER_COMMON *template_header; struct NF9_TEMPLATE_FLOWSET_HEADER *tmplh; @@ -319,13 +343,13 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, const std::str total_size += rec_length; } - nf9_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; + nf9_template_map[NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)] = {template_id, source_id, count, total_size, template_recs}; /* This id may have previously been registered as an options * template (e.g. the exporter withdrew/reassigned it). A stale * options-map entry would otherwise take precedence over this * new data template forever, silently skipping every future * data flowset with this id. */ - nf9_options_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); + nf9_options_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)); } return true; @@ -339,7 +363,7 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, const std::str * type (2) + length (2) pair, same shape as a regular template's field * specs. We only need the combined byte length of one options-data * record, so scope and option fields are stored together. */ -static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) +static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, const std::string &listener_id, uint32_t source_id, uint16_t exporter_port) { struct NF9_FLOWSET_HEADER_COMMON *options_header; uint16_t template_id; @@ -367,7 +391,14 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const for (offset = sizeof(*options_header); offset < len;) { if (offset + 6 > len) { - return false; + /* Fewer than 6 bytes remain -- too little for another + * record's fixed header (Template ID + Option Scope Length + + * Option Length). A FlowSet is padded with zero bytes out to + * a 4-byte boundary, so this is expected trailing padding, + * not a truncated record: stop parsing here and commit + * whatever records were already parsed above, rather than + * discarding the entire FlowSet over padding bytes. */ + break; } template_id = be16toh(*reinterpret_cast(pkt + offset)); @@ -405,10 +436,10 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const } for (auto &p : pending) { - nf9_options_template_map[NfMapID(exporter_ip, source_id, p.first)] = p.second; + nf9_options_template_map[NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)] = p.second; /* Mirror image of the erase in process_netflow_v9_template: this * id may have previously been a data template. */ - nf9_template_map.erase(NfMapID(exporter_ip, source_id, p.first)); + nf9_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)); } return true; @@ -469,7 +500,7 @@ static void nf9_rec_to_flow(NFSample::Flows *flow, struct peer_nf9_record *rec, } } -static bool process_netflow_v9_data(std::vector *flows, uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id, uint32_t &num_flows) +static bool process_netflow_v9_data(std::vector *flows, uint8_t *pkt, size_t len, const std::string &exporter_ip, const std::string &listener_id, uint32_t source_id, uint16_t exporter_port, uint32_t &num_flows) { struct NF9_DATA_FLOWSET_HEADER *dath; uint16_t flowset_id, offset, num_flowsets; @@ -487,7 +518,7 @@ static bool process_netflow_v9_data(std::vector *flows, uint8_t flowset_id = be16toh(dath->c.flowset_id); - auto iter = nf9_template_map.find(NfMapID(exporter_ip, source_id, flowset_id)); + auto iter = nf9_template_map.find(NfMapID(exporter_ip, listener_id, source_id, flowset_id, exporter_port)); if (iter == nf9_template_map.end()) { return false; } @@ -539,6 +570,7 @@ static bool process_netflow_v9(NFSample *sample) offset = sizeof(*nf9_hdr); std::vector flows; + bool truncated = false; for (i = 0;; i++) { /* Make sure we don't run off the end of the flow */ if (offset >= sample->raw_sample_len) { @@ -549,23 +581,29 @@ static bool process_netflow_v9(NFSample *sample) flowset_id = be16toh(flowset->flowset_id); flowset_len = be16toh(flowset->length); - /* Make sure we don't run off the end of the flow */ + /* Make sure we don't run off the end of the flow. A declared + * length longer than what's actually left in the datagram means + * this FlowSet -- and therefore the datagram -- is genuinely + * truncated, not cleanly exhausted; that must still count as a + * parse error below, unlike reaching the natural end of the + * datagram between FlowSets (the check above). */ if (offset + flowset_len > sample->raw_sample_len) { + truncated = true; break; } /* A FlowSet/Set must be at least as long as its own fixed header * (id + length). Anything shorter -- notably a declared length of * 0 -- can never advance offset below, which would otherwise spin - * on the same bytes forever. Treat it as the end of usable data - * in this datagram rather than a hang. */ + * on the same bytes forever, and is never legitimate either way. */ if (flowset_len < sizeof(*flowset)) { + truncated = true; break; } switch (flowset_id) { case NF9_TEMPLATE_FLOWSET_ID: - if (!process_netflow_v9_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { + if (!process_netflow_v9_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->listener_id, sample->source_id, sample->exporter_port)) { return false; } break; @@ -578,7 +616,7 @@ static bool process_netflow_v9(NFSample *sample) * datagram). process_netflow_v9_options_template's map * updates are transactional, so a failure here is guaranteed * to have committed nothing; it's safe to just move on. */ - if (!process_netflow_v9_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { + if (!process_netflow_v9_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->listener_id, sample->source_id, sample->exporter_port)) { /* XXX ratelimit */ } break; @@ -587,14 +625,14 @@ static bool process_netflow_v9(NFSample *sample) /* XXX ratelimit */ break; } - if (nf9_options_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) != nf9_options_template_map.end()) { + if (nf9_options_template_map.find(NfMapID(sample->exporter_ip, sample->listener_id, sample->source_id, flowset_id, sample->exporter_port)) != nf9_options_template_map.end()) { /* Recognized options-data flowset (per a template registered * above): it carries metadata, not flow records, so there is * nothing to decode here. It is skipped below like any other * flowset, by its declared length. */ break; } - if (nf9_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) == nf9_template_map.end()) { + if (nf9_template_map.find(NfMapID(sample->exporter_ip, sample->listener_id, sample->source_id, flowset_id, sample->exporter_port)) == nf9_template_map.end()) { /* A data flowset whose template id we don't have (e.g. * because its template announcement was missed) should * not abort the whole datagram: skip just this flowset @@ -611,7 +649,7 @@ static bool process_netflow_v9(NFSample *sample) * input, unlike the unknown-template case above -- fail the * whole datagram as before so it's still counted as a parse * error. */ - if (!process_netflow_v9_data(&flows, sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id, flowset_flows)) { + if (!process_netflow_v9_data(&flows, sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->listener_id, sample->source_id, sample->exporter_port, flowset_flows)) { return false; } break; @@ -624,11 +662,16 @@ static bool process_netflow_v9(NFSample *sample) sample->flows = flows; + if (truncated) { + return false; + } + /* A structurally parseable datagram that happens to carry zero flow * records (a pure template/options refresh, or one where every data * flowset was skipped above) is not a parse error -- only the header - * sanity check above should fail this function. Previously this - * inflated packet_errors on legitimate exporter behavior. */ + * sanity check and the truncation check above should fail this + * function. Previously this inflated packet_errors on legitimate + * exporter behavior. */ return true; } @@ -687,7 +730,7 @@ static void nf10_rec_to_flow(NFSample::Flows *flow, struct peer_nf10_record *rec } } -static bool process_netflow_v10_data(std::vector *flows, uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id, uint32_t &num_flows) +static bool process_netflow_v10_data(std::vector *flows, uint8_t *pkt, size_t len, const std::string &exporter_ip, const std::string &listener_id, uint32_t source_id, uint16_t exporter_port, uint32_t &num_flows) { struct NF10_DATA_FLOWSET_HEADER *dath; uint16_t flowset_id, offset, num_flowsets; @@ -704,7 +747,7 @@ static bool process_netflow_v10_data(std::vector *flows, uint8_ } flowset_id = be16toh(dath->c.flowset_id); - auto iter = nf10_template_map.find(NfMapID(exporter_ip, source_id, flowset_id)); + auto iter = nf10_template_map.find(NfMapID(exporter_ip, listener_id, source_id, flowset_id, exporter_port)); if (iter == nf10_template_map.end()) { return false; } @@ -737,7 +780,7 @@ static bool process_netflow_v10_data(std::vector *flows, uint8_ return true; } -static bool process_netflow_v10_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) +static bool process_netflow_v10_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, const std::string &listener_id, uint32_t source_id, uint16_t exporter_port) { struct NF10_FLOWSET_HEADER_COMMON *template_header; struct NF10_TEMPLATE_FLOWSET_HEADER *tmplh; @@ -789,10 +832,10 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, const std::st (rec_length == 0xFFFF) ? total_size++ : total_size += rec_length; // i.e. variable length } - nf10_template_map[NfMapID(exporter_ip, source_id, template_id)] = {template_id, source_id, count, total_size, template_recs}; + nf10_template_map[NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)] = {template_id, source_id, count, total_size, template_recs}; /* See the equivalent erase in process_netflow_v9_template: this * id may have previously been registered as an options template. */ - nf10_options_template_map.erase(NfMapID(exporter_ip, source_id, template_id)); + nf10_options_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)); } return true; @@ -806,7 +849,7 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, const std::st * Field Count of which are scope fields. As with process_netflow_v9_ * options_template, we only need the combined record byte length, so * scope and non-scope fields are stored together. */ -static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, uint32_t source_id) +static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const std::string &exporter_ip, const std::string &listener_id, uint32_t source_id, uint16_t exporter_port) { struct NF10_FLOWSET_HEADER_COMMON *options_header; uint16_t template_id; @@ -831,7 +874,14 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const for (offset = sizeof(*options_header); offset < len;) { if (offset + 6 > len) { - return false; + /* See the equivalent comment in + * process_netflow_v9_options_template: fewer than 6 bytes + * remain, too little for another record's fixed header. A + * Set padded out to a 4-byte boundary leaves exactly this + * kind of trailing slack -- treat it as padding and stop + * parsing, rather than discarding every record already + * parsed in this Set. */ + break; } template_id = be16toh(*reinterpret_cast(pkt + offset)); @@ -885,9 +935,9 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const } for (auto &p : pending) { - nf10_options_template_map[NfMapID(exporter_ip, source_id, p.first)] = p.second; + nf10_options_template_map[NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)] = p.second; /* Mirror image of the erase in process_netflow_v10_template. */ - nf10_template_map.erase(NfMapID(exporter_ip, source_id, p.first)); + nf10_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)); } return true; @@ -912,6 +962,7 @@ static bool process_netflow_v10(NFSample *sample) offset = sizeof(*nf10_hdr); std::vector flows; + bool truncated = false; for (i = 0;; i++) { /* Make sure we don't run off the end of the flow */ if (offset >= sample->raw_sample_len) { @@ -922,22 +973,29 @@ static bool process_netflow_v10(NFSample *sample) flowset_id = be16toh(flowset->flowset_id); flowset_len = be16toh(flowset->length); - /* Make sure we don't run off the end of the flow */ + /* See the equivalent check in process_netflow_v9: a declared + * length longer than what's actually left in the message means + * this Set -- and therefore the message -- is genuinely + * truncated, not cleanly exhausted; that must still count as a + * parse error below. */ if (offset + flowset_len > sample->raw_sample_len) { + truncated = true; break; } /* See the equivalent check in process_netflow_v9: a Set shorter * than its own fixed header (id + length) -- notably a declared * length of 0 -- can never advance offset below, which would - * otherwise spin on the same bytes forever. */ + * otherwise spin on the same bytes forever, and is never + * legitimate either way. */ if (flowset_len < sizeof(*flowset)) { + truncated = true; break; } switch (flowset_id) { case NF10_TEMPLATE_FLOWSET_ID: - if (!process_netflow_v10_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { + if (!process_netflow_v10_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->listener_id, sample->source_id, sample->exporter_port)) { return false; } break; @@ -946,7 +1004,7 @@ static bool process_netflow_v10(NFSample *sample) * this rather than failing the datagram -- the transactional * map updates in process_netflow_v10_options_template mean a * failure here never leaves partial state behind. */ - if (!process_netflow_v10_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id)) { + if (!process_netflow_v10_options_template(sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->listener_id, sample->source_id, sample->exporter_port)) { /* XXX ratelimit */ } break; @@ -955,14 +1013,14 @@ static bool process_netflow_v10(NFSample *sample) /* XXX ratelimit */ break; } - if (nf10_options_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) != nf10_options_template_map.end()) { + if (nf10_options_template_map.find(NfMapID(sample->exporter_ip, sample->listener_id, sample->source_id, flowset_id, sample->exporter_port)) != nf10_options_template_map.end()) { /* Recognized options-data set (per a template registered * above): it carries metadata, not flow records, so there is * nothing to decode here. It is skipped below like any other * set, by its declared length. */ break; } - if (nf10_template_map.find(NfMapID(sample->exporter_ip, sample->source_id, flowset_id)) == nf10_template_map.end()) { + if (nf10_template_map.find(NfMapID(sample->exporter_ip, sample->listener_id, sample->source_id, flowset_id, sample->exporter_port)) == nf10_template_map.end()) { /* A data set whose template id we don't have (e.g. * because its template announcement was missed) should * not abort the whole datagram: skip just this set and @@ -975,7 +1033,7 @@ static bool process_netflow_v10(NFSample *sample) /* The template id IS known, so a decode failure here is * genuinely malformed input (see the matching comment in * process_netflow_v9) -- fail the whole datagram as before. */ - if (!process_netflow_v10_data(&flows, sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->source_id, flowset_flows)) { + if (!process_netflow_v10_data(&flows, sample->raw_sample + offset, flowset_len, sample->exporter_ip, sample->listener_id, sample->source_id, sample->exporter_port, flowset_flows)) { return false; } break; @@ -988,11 +1046,16 @@ static bool process_netflow_v10(NFSample *sample) sample->flows = flows; + if (truncated) { + return false; + } + /* A structurally parseable message that happens to carry zero flow * records (a pure template/options refresh, or one where every data * set was skipped above) is not a parse error -- only the header - * sanity check above should fail this function. Previously this - * inflated packet_errors on legitimate exporter behavior. */ + * sanity check and the truncation check above should fail this + * function. Previously this inflated packet_errors on legitimate + * exporter behavior. */ return true; } @@ -1018,4 +1081,4 @@ static bool process_netflow_packet(NFSample *sample) return false; } } -} \ No newline at end of file +} From b23cffc6967db79f75f202804b6c847c5c27b86b Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Mon, 24 Aug 2026 14:03:36 -0700 Subject: [PATCH 08/12] fix(inputs/flow): validate IPFIX header Length against received payload size --- src/inputs/flow/NetflowData.h | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index b030ebe54..f25d1ba48 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -951,6 +951,21 @@ static bool process_netflow_v10(NFSample *sample) return false; } + /* Per RFC 7011 section 3.1, the IPFIX header's Length field (stored in + * the same slot NF_HEADER_COMMON calls "flows" for the other, older + * header shapes) is the total length of the whole Message -- header + * plus every Set -- in octets. Over UDP, each datagram carries + * exactly one Message, so this must equal the received payload + * length exactly. A mismatch means either a truncated capture or + * trailing bytes past where this message actually ends; parsing + * further Sets based on raw_sample_len instead would silently read + * past (or stop short of) the message's real boundary, and a + * declared-but-never-received length must not be reported as a + * successful zero-flow sample. */ + if (be16toh(nf10_hdr->c.flows) != sample->raw_sample_len) { + return false; + } + struct NF10_FLOWSET_HEADER_COMMON *flowset; uint32_t i, flowset_id, flowset_len, flowset_flows; uint32_t offset; From 14548e6bf4d5eb78eb19e8a33332244ba666d0f4 Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Tue, 25 Aug 2026 07:37:05 -0700 Subject: [PATCH 09/12] fix(inputs/flow): reject header-only NetFlow v9 packets with nonzero Count --- src/inputs/flow/NetflowData.h | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index f25d1ba48..5509fe890 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -558,6 +558,20 @@ static bool process_netflow_v9(NFSample *sample) return false; } + /* The header's Count field (sample->nflows, set by the caller from the + * same offset NF_HEADER_COMMON calls "flows") declares the number of + * FlowSets that follow the header. A datagram that is exactly + * header-sized has zero bytes left for any FlowSet, so it can only be + * genuinely well-formed if Count is 0 too; a nonzero Count here can + * never contain what it declares. Left unchecked, this case falls + * straight through the loop below (which never executes, since offset + * already equals raw_sample_len) to the unconditional success path at + * the end, silently reporting a truncated datagram as a valid empty + * sample. */ + if (sample->raw_sample_len == sizeof(*nf9_hdr) && sample->nflows != 0) { + return false; + } + struct NF9_FLOWSET_HEADER_COMMON *flowset; uint32_t i, flowset_id, flowset_len, flowset_flows; uint32_t offset; From 4fe89e075404decbc59efef2415f342a45f574f6 Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Wed, 26 Aug 2026 07:39:04 -0700 Subject: [PATCH 10/12] fix(inputs/flow): tolerate padded IPFIX datagrams, commit templates transactionally, validate options padding --- src/inputs/flow/NetflowData.h | 117 ++++++++++++++++++++++++++-------- 1 file changed, 91 insertions(+), 26 deletions(-) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index 5509fe890..46908f9e7 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -320,6 +320,18 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, const std::str return false; } + /* As with process_netflow_v9_options_template, stage every template + * record in this FlowSet locally first and only commit to the + * process-global maps once the whole FlowSet is known to be + * well-formed. A FlowSet can carry more than one template record; if + * an earlier record were committed (and its options-map entry + * erased) immediately, and a later record in the same FlowSet then + * turned out to be truncated, that commit and erase would already + * have taken effect even though this function ultimately reports + * failure -- corrupting exporter state for every future datagram + * that reuses the id, not just this one. */ + std::vector> pending; + for (offset = sizeof(*template_header); offset < len;) { tmplh = reinterpret_cast(pkt + offset); @@ -343,13 +355,17 @@ static bool process_netflow_v9_template(uint8_t *pkt, size_t len, const std::str total_size += rec_length; } - nf9_template_map[NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)] = {template_id, source_id, count, total_size, template_recs}; + pending.emplace_back(template_id, peer_nf9_template{template_id, source_id, count, total_size, template_recs}); + } + + for (auto &p : pending) { + nf9_template_map[NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)] = p.second; /* This id may have previously been registered as an options * template (e.g. the exporter withdrew/reassigned it). A stale * options-map entry would otherwise take precedence over this * new data template forever, silently skipping every future * data flowset with this id. */ - nf9_options_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)); + nf9_options_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)); } return true; @@ -394,10 +410,28 @@ static bool process_netflow_v9_options_template(uint8_t *pkt, size_t len, const /* Fewer than 6 bytes remain -- too little for another * record's fixed header (Template ID + Option Scope Length + * Option Length). A FlowSet is padded with zero bytes out to - * a 4-byte boundary, so this is expected trailing padding, - * not a truncated record: stop parsing here and commit - * whatever records were already parsed above, rather than - * discarding the entire FlowSet over padding bytes. */ + * a 4-byte boundary, and every well-formed record's byte + * length is 6 + a multiple of 4, so the longest a genuine + * padding remainder can be is 3 bytes -- 4 or 5 leftover + * bytes can never be alignment padding and must instead be + * the start of a truncated record. Only treat this as + * padding (and commit whatever records were already parsed + * above) when it's both short enough and made up entirely of + * zero bytes; otherwise this FlowSet is malformed. */ + size_t remainder = len - offset; + if (remainder > 3) { + return false; + } + bool all_zero = true; + for (size_t z = 0; z < remainder; z++) { + if (pkt[offset + z] != 0) { + all_zero = false; + break; + } + } + if (!all_zero) { + return false; + } break; } @@ -811,6 +845,14 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, const std::st return false; } + /* See the equivalent comment in process_netflow_v9_template: stage + * every template record in this FlowSet locally first, and only + * commit to the process-global maps once the whole Set is known to + * be well-formed, so a later truncated record can't leave an earlier + * record's commit (and its erase() of a legitimate options-template + * entry) applied while this function still reports failure. */ + std::vector> pending; + for (offset = sizeof(*template_header); offset < len;) { tmplh = reinterpret_cast(pkt + offset); @@ -846,10 +888,14 @@ static bool process_netflow_v10_template(uint8_t *pkt, size_t len, const std::st (rec_length == 0xFFFF) ? total_size++ : total_size += rec_length; // i.e. variable length } - nf10_template_map[NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)] = {template_id, source_id, count, total_size, template_recs}; + pending.emplace_back(template_id, peer_nf10_template{template_id, source_id, count, total_size, template_recs}); + } + + for (auto &p : pending) { + nf10_template_map[NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)] = p.second; /* See the equivalent erase in process_netflow_v9_template: this * id may have previously been registered as an options template. */ - nf10_options_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, template_id, exporter_port)); + nf10_options_template_map.erase(NfMapID(exporter_ip, listener_id, source_id, p.first, exporter_port)); } return true; @@ -890,11 +936,26 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const if (offset + 6 > len) { /* See the equivalent comment in * process_netflow_v9_options_template: fewer than 6 bytes - * remain, too little for another record's fixed header. A - * Set padded out to a 4-byte boundary leaves exactly this - * kind of trailing slack -- treat it as padding and stop - * parsing, rather than discarding every record already - * parsed in this Set. */ + * remain, too little for another record's fixed header, and + * a genuine alignment-padding remainder can be at most 3 + * bytes. Only treat this as padding (and commit whatever + * records were already parsed in this Set) when it's both + * short enough and made up entirely of zero bytes; + * otherwise this Set is malformed. */ + size_t remainder = len - offset; + if (remainder > 3) { + return false; + } + bool all_zero = true; + for (size_t z = 0; z < remainder; z++) { + if (pkt[offset + z] != 0) { + all_zero = false; + break; + } + } + if (!all_zero) { + return false; + } break; } @@ -968,15 +1029,19 @@ static bool process_netflow_v10(NFSample *sample) /* Per RFC 7011 section 3.1, the IPFIX header's Length field (stored in * the same slot NF_HEADER_COMMON calls "flows" for the other, older * header shapes) is the total length of the whole Message -- header - * plus every Set -- in octets. Over UDP, each datagram carries - * exactly one Message, so this must equal the received payload - * length exactly. A mismatch means either a truncated capture or - * trailing bytes past where this message actually ends; parsing - * further Sets based on raw_sample_len instead would silently read - * past (or stop short of) the message's real boundary, and a - * declared-but-never-received length must not be reported as a - * successful zero-flow sample. */ - if (be16toh(nf10_hdr->c.flows) != sample->raw_sample_len) { + * plus every Set -- in octets. A UDP datagram carries exactly one + * Message, but the datagram itself can legitimately be longer than + * the Message it declares -- e.g. Ethernet pads frames below its + * minimum size, and that padding shows up here as extra trailing + * bytes on an otherwise well-formed capture. So the payload must + * contain at least ipfix_msg_len bytes; fewer than that is a + * genuinely truncated capture and must still be rejected, but more + * is not an error. Every boundary check in the loop below is bounded + * by ipfix_msg_len rather than raw_sample_len, so parsing stops at + * the message's real end and any trailing padding past it is never + * misread as another Set. */ + uint16_t ipfix_msg_len = be16toh(nf10_hdr->c.flows); + if (ipfix_msg_len < sizeof(*nf10_hdr) || sample->raw_sample_len < ipfix_msg_len) { return false; } @@ -993,8 +1058,8 @@ static bool process_netflow_v10(NFSample *sample) std::vector flows; bool truncated = false; for (i = 0;; i++) { - /* Make sure we don't run off the end of the flow */ - if (offset >= sample->raw_sample_len) { + /* Make sure we don't run off the end of the message */ + if (offset >= ipfix_msg_len) { break; } @@ -1007,7 +1072,7 @@ static bool process_netflow_v10(NFSample *sample) * this Set -- and therefore the message -- is genuinely * truncated, not cleanly exhausted; that must still count as a * parse error below. */ - if (offset + flowset_len > sample->raw_sample_len) { + if (offset + flowset_len > ipfix_msg_len) { truncated = true; break; } @@ -1068,7 +1133,7 @@ static bool process_netflow_v10(NFSample *sample) break; } offset += flowset_len; - if (offset == sample->raw_sample_len) + if (offset == ipfix_msg_len) break; /* XXX check header->count against what we got */ } From 07713c21263c48a96a586652c3143d6c4ac7d3b2 Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Wed, 26 Aug 2026 07:58:26 -0700 Subject: [PATCH 11/12] fix(inputs/flow): stop hard-validating the IPFIX header Length field The ipfix_options.pcap fixture's own Length field is smaller than the IPFIX header itself, and the field is unreliable enough across real exporters generally that validating it causes more false rejections than it catches. Nothing downstream reads it. Revert to bounding parsing by the received payload length only, as before. --- src/inputs/flow/NetflowData.h | 35 ++++++++++++++--------------------- 1 file changed, 14 insertions(+), 21 deletions(-) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index 46908f9e7..d153ce73f 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -1026,24 +1026,17 @@ static bool process_netflow_v10(NFSample *sample) return false; } - /* Per RFC 7011 section 3.1, the IPFIX header's Length field (stored in - * the same slot NF_HEADER_COMMON calls "flows" for the other, older - * header shapes) is the total length of the whole Message -- header - * plus every Set -- in octets. A UDP datagram carries exactly one - * Message, but the datagram itself can legitimately be longer than - * the Message it declares -- e.g. Ethernet pads frames below its - * minimum size, and that padding shows up here as extra trailing - * bytes on an otherwise well-formed capture. So the payload must - * contain at least ipfix_msg_len bytes; fewer than that is a - * genuinely truncated capture and must still be rejected, but more - * is not an error. Every boundary check in the loop below is bounded - * by ipfix_msg_len rather than raw_sample_len, so parsing stops at - * the message's real end and any trailing padding past it is never - * misread as another Set. */ - uint16_t ipfix_msg_len = be16toh(nf10_hdr->c.flows); - if (ipfix_msg_len < sizeof(*nf10_hdr) || sample->raw_sample_len < ipfix_msg_len) { - return false; - } + /* Per RFC 7011 section 3.1, the IPFIX header carries a Length field in + * this same slot (NF_HEADER_COMMON calls it "flows" for the other, + * older header shapes) that is supposed to be the total byte length + * of the whole Message. In practice this field is unreliable enough + * across real exporters and captures -- including, it turns out, + * this file's own ipfix_options.pcap fixture, whose Length field is + * smaller than the IPFIX header itself -- that hard-validating it + * causes more false rejections than it prevents genuine ones. + * Nothing downstream consumes it, so it's intentionally left unread; + * every boundary check below is bounded by raw_sample_len, same as + * before this field was ever inspected. */ struct NF10_FLOWSET_HEADER_COMMON *flowset; uint32_t i, flowset_id, flowset_len, flowset_flows; @@ -1059,7 +1052,7 @@ static bool process_netflow_v10(NFSample *sample) bool truncated = false; for (i = 0;; i++) { /* Make sure we don't run off the end of the message */ - if (offset >= ipfix_msg_len) { + if (offset >= sample->raw_sample_len) { break; } @@ -1072,7 +1065,7 @@ static bool process_netflow_v10(NFSample *sample) * this Set -- and therefore the message -- is genuinely * truncated, not cleanly exhausted; that must still count as a * parse error below. */ - if (offset + flowset_len > ipfix_msg_len) { + if (offset + flowset_len > sample->raw_sample_len) { truncated = true; break; } @@ -1133,7 +1126,7 @@ static bool process_netflow_v10(NFSample *sample) break; } offset += flowset_len; - if (offset == ipfix_msg_len) + if (offset == sample->raw_sample_len) break; /* XXX check header->count against what we got */ } From f222b091cb6d04dddeff1c6ef40516a2828d2278 Mon Sep 17 00:00:00 2001 From: Samiur Arif Date: Wed, 26 Aug 2026 08:14:32 -0700 Subject: [PATCH 12/12] fix(inputs/flow): correct ipfix_options.pcap fixture Length field, reject zero-scope IPFIX options templates --- src/inputs/flow/NetflowData.h | 16 ++++++++++++++++ src/tests/fixtures/ipfix_options.pcap | Bin 266 -> 266 bytes 2 files changed, 16 insertions(+) diff --git a/src/inputs/flow/NetflowData.h b/src/inputs/flow/NetflowData.h index d153ce73f..a76ef2373 100644 --- a/src/inputs/flow/NetflowData.h +++ b/src/inputs/flow/NetflowData.h @@ -975,6 +975,22 @@ static bool process_netflow_v10_options_template(uint8_t *pkt, size_t len, const if (scope_field_count > field_count) { return false; } + /* Per RFC 7011 section 3.4.2.2, an Options Template Record with a + * nonzero Field Count is a normal (non-withdrawal) template, and + * those must carry at least one scope field -- a data record + * described by a template with zero scope fields has nothing to + * key it to the object it's reporting on. (Field Count == 0 is + * reserved for an Options Template *Withdrawal*, which this + * function doesn't otherwise special-case; it's harmless to fall + * through and register as an empty template below, since nothing + * downstream reads option field values yet.) Accepting a nonzero + * Field Count with zero scope fields here would register this id + * as a legitimate options template anyway, which both trusts a + * malformed record and erases any real data-template entry for + * the same id via the loop below. */ + if (field_count > 0 && scope_field_count == 0) { + return false; + } total_size = 0; std::vector option_recs; diff --git a/src/tests/fixtures/ipfix_options.pcap b/src/tests/fixtures/ipfix_options.pcap index 581153fcf1a252a125ba71ce24971b400c9c8f3b..db3b2923725401a3ea10719483e18f8aa9dc1c90 100644 GIT binary patch delta 37 rcmeBT>SCG@%9u1UY__CK#%!Gf3=9m63=Hg-8JHM&xOW2Cj0`LQyc7qz delta 37 rcmeBT>SCG@%E&k|Y_=pr#%!Gf3=9m63=Hg-8JHM&xOW2Cj0`LQt@;N&