fix(inputs/flow): don't drop whole NetFlow v9/IPFIX datagram on unkno…

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<uint64_t>("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<uint64_t>("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);
+}

src/inputs/flow/FlowInputStream.cpp

@@ -106,16 +106,23 @@ void FlowInputStream::_read_from_pcap_file()
                 NFSample sample;
                 sample.raw_sample = udpLayer->getLayerPayload();
                 sample.raw_sample_len = udpLayer->getLayerPayloadSize();
+                // 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<pcpp::IPv4Layer>(); IP4layer) {
+                    sample.exporter_ip = IP4layer->getSrcIPv4Address().toString();
+                } else if (auto IP6layer = netflow_pkt.getLayerOfType<pcpp::IPv6Layer>(); IP6layer) {
+                    sample.exporter_ip = IP6layer->getSrcIPv6Address().toString();
+                }
+                sample.listener_id = _name;
+                sample.exporter_port = udpLayer->getSrcPort();
                 if (process_netflow_packet(&sample)) {
-                    std::string src_ip;
-                    if (auto IP4layer = netflow_pkt.getLayerOfType<pcpp::IPv4Layer>(); IP4layer) {
-                        src_ip = IP4layer->getSrcIPv4Address().toString();
-                    } else if (auto IP6layer = netflow_pkt.getLayerOfType<pcpp::IPv6Layer>(); IP6layer) {
-                        src_ip = IP6layer->getSrcIPv6Address().toString();
-                    }
                     std::shared_lock lock(_input_mutex);
                     for (auto &proxy : _event_proxies) {
-                        static_cast<FlowInputEventProxy *>(proxy.get())->netflow_cb(src_ip, sample, rawPacket.getRawDataLen());
+                        static_cast<FlowInputEventProxy *>(proxy.get())->netflow_cb(sample.exporter_ip, sample, rawPacket.getRawDataLen());
                     }
                 } else {
                     _logger->error("invalid netflow or ipfix packet");
@@ -213,6 +220,15 @@ void FlowInputStream::_create_frame_stream_udp_socket()
             NFSample sample;
             sample.raw_sample = reinterpret_cast<uint8_t *>(event.data.get());
             sample.raw_sample_len = event.length;
+            // 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<uint16_t>(event.sender.port);
             if (process_netflow_packet(&sample)) {
                 std::shared_lock lock(_input_mutex);
                 for (auto &proxy : _event_proxies) {