GH-1620: Add event and plugin hook to track packets not processed

NEWS

@@ -40,6 +40,15 @@ New Functionality
 
 - The ``find_str`` and ``rfind_str`` bifs now support case-insensitive searches.
 
+- Added a new plugin hook for capturing packets that made it through analysis
+  without being processed called ``Plugin::HookUnprocessedPacket``. Currently
+  ARP packets or packets with a valid IP-based transport header are marked as
+  processed. This also adds an event called ``packet_not_processed`` that
+  reports the same packets.
+
+- A new command-line option ``-c`` or ``--capture-unprocessed`` will dump any
+  packets not marked as being processed, similar to the new hook and event above.
+
 Changed Functionality
 ---------------------
 

src/RunState.cc

@@ -390,8 +390,16 @@ void get_final_stats()
 		                         ? ((double)s.dropped / ((double)s.received + (double)s.dropped)) *
 		                               100.0
 		                         : 0.0;
-		reporter->Info("%" PRIu64 " packets received on interface %s, %" PRIu64 " (%.2f%%) dropped",
+
-		               s.received, ps->Path().c_str(), s.dropped, dropped_pct);
+		uint64_t not_processed = packet_mgr->GetUnprocessedCount();
+		double unprocessed_pct = not_processed > 0
+		                             ? ((double)not_processed / (double)s.received) * 100.0
+		                             : 0.0;
+
+		reporter->Info("%" PRIu64 " packets received on interface %s, %" PRIu64
+		               " (%.2f%%) dropped, %" PRIu64 " (%.2f%%) not processed",
+		               s.received, ps->Path().c_str(), s.dropped, dropped_pct, not_processed,
+		               unprocessed_pct);
 		}
 	}
 

src/event.bif

@@ -907,3 +907,11 @@ event Pcap::file_done%(path: string%);
 ##
 ## .. zeek:see:: UnknownProtocol::first_bytes_count
 event unknown_protocol%(analyzer_name: string, protocol: count, first_bytes: string%);
+
+## An event for handling packets that reached the end of processing without
+## being marked as processed. Note that this event may lead to unpredictable
+## performance spikes, particularly if a network suddenly receives a burst
+## of packets that are unprocessed.
+##
+## pkt: Data for the unprocessed packet
+event packet_not_processed%(pkt: pcap_packet%);

src/iosource/Packet.cc

@@ -68,6 +68,8 @@ void Packet::Init(int arg_link_type, pkt_timeval* arg_ts, uint32_t arg_caplen, u
 	tunnel_type = BifEnum::Tunnel::IP;
 	gre_version = -1;
 	gre_link_type = DLT_RAW;
+
+	processed = false;
 	}
 
 Packet::~Packet()
@@ -146,6 +148,33 @@ RecordValPtr Packet::ToRawPktHdrVal() const
 		return pkt_hdr;
 	}
 
+RecordValPtr Packet::ToVal(const Packet* p)
+	{
+	static auto pcap_packet = zeek::id::find_type<zeek::RecordType>("pcap_packet");
+	auto val = zeek::make_intrusive<zeek::RecordVal>(pcap_packet);
+
+	if ( p )
+		{
+		val->Assign(0, static_cast<uint32_t>(p->ts.tv_sec));
+		val->Assign(1, static_cast<uint32_t>(p->ts.tv_usec));
+		val->Assign(2, p->cap_len);
+		val->Assign(3, p->len);
+		val->Assign(4, zeek::make_intrusive<zeek::StringVal>(p->cap_len, (const char*)p->data));
+		val->Assign(5, zeek::BifType::Enum::link_encap->GetEnumVal(p->link_type));
+		}
+	else
+		{
+		val->Assign(0, 0);
+		val->Assign(1, 0);
+		val->Assign(2, 0);
+		val->Assign(3, 0);
+		val->Assign(4, zeek::val_mgr->EmptyString());
+		val->Assign(5, zeek::BifType::Enum::link_encap->GetEnumVal(BifEnum::LINK_UNKNOWN));
+		}
+
+	return val;
+	}
+
 ValPtr Packet::FmtEUI48(const u_char* mac) const
 	{
 	char buf[20];

src/iosource/Packet.h

@@ -122,6 +122,12 @@ public:
 	 */
 	RecordValPtr ToRawPktHdrVal() const;
 
+	/**
+	 * Returns a RecordVal that represents the Packet. This is used
+	 * by the get_current_packet bif.
+	 */
+	static RecordValPtr ToVal(const Packet* p);
+
 	/**
 	 * Maximal length of a layer 2 address.
 	 */
@@ -241,6 +247,14 @@ public:
 	 */
 	int gre_link_type = DLT_RAW;
 
+	/**
+	 * This flag indicates whether a packet has been processed. This can
+	 * mean different things depending on the traffic, but generally it
+	 * means that a packet has been logged in some way. We default to
+	 * false, and this can be set to true for any number of reasons.
+	 */
+	bool processed = false;
+
 private:
 	// Renders an MAC address into its ASCII representation.
 	ValPtr FmtEUI48(const u_char* mac) const;

src/packet_analysis/Manager.cc

@@ -7,6 +7,7 @@
 #include "zeek/iosource/PktDumper.h"
 #include "zeek/packet_analysis/Analyzer.h"
 #include "zeek/packet_analysis/Dispatcher.h"
+#include "zeek/plugin/Manager.h"
 #include "zeek/zeek-bif.h"
 
 using namespace zeek::packet_analysis;
@@ -106,6 +107,16 @@ void Manager::ProcessPacket(Packet* packet)
 	// Start packet analysis
 	root_analyzer->ForwardPacket(packet->cap_len, packet->data, packet, packet->link_type);
 
+	if ( ! packet->processed )
+		{
+		if ( packet_not_processed )
+			event_mgr.Enqueue(packet_not_processed, Packet::ToVal(packet));
+
+		plugin_mgr->HookUnprocessedPacket(packet);
+
+		total_not_processed++;
+		}
+
 	if ( raw_packet )
 		event_mgr.Enqueue(raw_packet, packet->ToRawPktHdrVal());
 

src/packet_analysis/Manager.h

@@ -2,6 +2,7 @@
 
 #pragma once
 
+#include "zeek/Func.h"
 #include "zeek/PacketFilter.h"
 #include "zeek/iosource/Packet.h"
 #include "zeek/packet_analysis/Component.h"
@@ -126,6 +127,12 @@ public:
 		return pkt_filter;
 		}
 
+	/**
+	 * Returns the total number of packets received that weren't considered
+	 * processed by some analyzer.
+	 */
+	uint64_t GetUnprocessedCount() const { return total_not_processed; }
+
 private:
 	/**
 	 * Instantiates a new analyzer instance.
@@ -163,6 +170,8 @@ private:
 	uint64_t unknown_sampling_rate = 0;
 	double unknown_sampling_duration = 0;
 	uint64_t unknown_first_bytes_count = 0;
+
+	uint64_t total_not_processed = 0;
 	};
 
 	} // namespace packet_analysis

src/packet_analysis/protocol/arp/ARP.cc

@@ -101,6 +101,11 @@ bool ARPAnalyzer::AnalyzePacket(size_t len, const uint8_t* data, Packet* packet)
 		return false;
 		}
 
+	// ARP packets are considered processed if we get to this point. There may be issues
+	// with the processing of them, but they're actually an ARP packet and anything else
+	// will be reported via events.
+	packet->processed = true;
+
 	// Check the address description fields.
 	switch ( ntohs(ah->ar_hrd) )
 		{

src/packet_analysis/protocol/ip/IPBasedAnalyzer.cc

@@ -63,6 +63,10 @@ bool IPBasedAnalyzer::AnalyzePacket(size_t len, const uint8_t* data, Packet* pkt
 	if ( ! conn )
 		return false;
 
+	// If we successfuly made a connection for this packet that means it'll eventually
+	// get logged, which means we can mark this packet as having been processed.
+	pkt->processed = true;
+
 	bool is_orig = (tuple.src_addr == conn->OrigAddr()) && (tuple.src_port == conn->OrigPort());
 
 	conn->CheckFlowLabel(is_orig, ip_hdr->FlowLabel());

src/plugin/Manager.cc

@@ -1012,6 +1012,29 @@ bool Manager::HookReporter(const std::string& prefix, const EventHandlerPtr even
 	return result;
 	}
 
+void Manager::HookUnprocessedPacket(const Packet* packet) const
+	{
+	HookArgumentList args;
+
+	if ( HavePluginForHook(META_HOOK_PRE) )
+		{
+		args.emplace_back(HookArgument{packet});
+		MetaHookPre(HOOK_UNPROCESSED_PACKET, args);
+		}
+
+	hook_list* l = hooks[HOOK_UNPROCESSED_PACKET];
+
+	if ( l )
+		for ( hook_list::iterator i = l->begin(); i != l->end(); ++i )
+			{
+			Plugin* p = (*i).second;
+			p->HookUnprocessedPacket(packet);
+			}
+
+	if ( HavePluginForHook(META_HOOK_POST) )
+		MetaHookPost(HOOK_UNPROCESSED_PACKET, args, HookArgument());
+	}
+
 void Manager::MetaHookPre(HookType hook, const HookArgumentList& args) const
 	{
 	if ( hook_list* l = hooks[HOOK_CALL_FUNCTION] )

src/plugin/Manager.h

@@ -419,6 +419,15 @@ public:
 	                  const zeek::detail::Location* location2, bool time,
 	                  const std::string& message);
 
+	/**
+	 * Hook for packets that are considered unprocessed by an Analyzer. This
+	 * typically means that a packet has not had a log entry written for it by
+	 * the time analysis finishes.
+	 *
+	 * @param packet The data for an unprocessed packet
+	 */
+	void HookUnprocessedPacket(const Packet* packet) const;
+
 	/**
 	 * Internal method that registers a freshly instantiated plugin with
 	 * the manager.

src/plugin/Plugin.cc

@@ -31,6 +31,7 @@ const char* hook_name(HookType h)
 		"SetupAnalyzerTree",
 		"LogInit",
 		"LogWrite",
+		"UnprocessedPacket",
 		// MetaHooks
 		"MetaHookPre",
 		"MetaHookPost",
@@ -244,7 +245,11 @@ void HookArgument::Describe(ODesc* d) const
 				d->Add("<no content>");
 
 			d->Add(")");
+			break;
 			}
+
+		case PACKET:
+			d->Add("<packet>");
 			break;
 		}
 	}
@@ -432,6 +437,8 @@ bool Plugin::HookReporter(const std::string& prefix, const EventHandlerPtr event
 	return true;
 	}
 
+void Plugin::HookUnprocessedPacket(const Packet* packet) { }
+
 void Plugin::MetaHookPre(HookType hook, const HookArgumentList& args) { }
 
 void Plugin::MetaHookPost(HookType hook, const HookArgumentList& args, HookArgument result) { }

src/plugin/Plugin.h

@@ -30,6 +30,7 @@ class ODesc;
 class Event;
 class Func;
 class Obj;
+class Packet;
 
 template <class T> class IntrusivePtr;
 using ValPtr = IntrusivePtr<Val>;
@@ -68,6 +69,7 @@ enum HookType
 	HOOK_LOG_INIT, //< Activates Plugin::HookLogInit
 	HOOK_LOG_WRITE, //< Activates Plugin::HookLogWrite
 	HOOK_REPORTER, //< Activates Plugin::HookReporter
+	HOOK_UNPROCESSED_PACKET, //<Activates Plugin::HookUnprocessedPacket
 
 	// Meta hooks.
 	META_HOOK_PRE, //< Activates Plugin::MetaHookPre().
@@ -208,7 +210,8 @@ public:
 		THREAD_FIELDS,
 		LOCATION,
 		ARG_LIST,
-		INPUT_FILE
+		INPUT_FILE,
+		PACKET
 		};
 
 	/**
@@ -370,7 +373,17 @@ public:
 		}
 
 	/**
-	 * Returns the value for a boolen argument. The argument's type must
+	 * Returns the value for a zeek::Packet* argument. The argument's type must
+	 * Constructor with a zeek::Packet* argument.
+	 */
+	explicit HookArgument(const Packet* packet)
+		{
+		type = PACKET;
+		arg.packet = packet;
+		}
+
+	/**
+	 * Returns the value for a boolean argument. The argument's type must
 	 * match accordingly.
 	 */
 	bool AsBool() const
@@ -509,7 +522,7 @@ public:
 		}
 
 	/**
-	 * Returns the value for a vod pointer argument. The argument's type
+	 * Returns the value for a void pointer argument. The argument's type
 	 * must match accordingly.
 	 */
 	const void* AsVoidPtr() const
@@ -518,6 +531,16 @@ public:
 		return arg.voidp;
 		}
 
+	/**
+	 * Returns the value for a Packet pointer argument. The argument's type
+	 * must match accordingly.
+	 */
+	const Packet* AsPacket() const
+		{
+		assert(type == PACKET);
+		return arg.packet;
+		}
+
 	/**
 	 * Returns the argument's type.
 	 */
@@ -546,6 +569,7 @@ private:
 		const void* voidp;
 		const logging::WriterBackend::WriterInfo* winfo;
 		const detail::Location* loc;
+		const Packet* packet;
 		} arg;
 
 	// Outside union because these have dtors.
@@ -1039,6 +1063,15 @@ protected:
 	                          const zeek::detail::Location* location2, bool time,
 	                          const std::string& message);
 
+	/**
+	 * Hook for packets that are considered unprocessed by an Analyzer. This
+	 * typically means that a packet has not had a log entry written for it by
+	 * the time analysis finishes.
+	 *
+	 * @param packet The data for an unprocessed packet
+	 */
+	virtual void HookUnprocessedPacket(const Packet* packet);
+
 	// Meta hooks.
 	virtual void MetaHookPre(HookType hook, const HookArgumentList& args);
 	virtual void MetaHookPost(HookType hook, const HookArgumentList& args, HookArgument result);

src/zeek.bif

@@ -3509,30 +3509,13 @@ function dump_current_packet%(file_name: string%) : bool
 ## .. zeek:see:: dump_current_packet dump_packet
 function get_current_packet%(%) : pcap_packet
 	%{
-	static auto pcap_packet = zeek::id::find_type<zeek::RecordType>("pcap_packet");
+	const Packet* p = nullptr;
-	const Packet* p;
-	auto pkt = zeek::make_intrusive<zeek::RecordVal>(pcap_packet);
 	zeek::iosource::PktSrc* pkt_src = zeek::run_state::detail::current_packet_source();
 
 	if ( ! pkt_src || ! pkt_src->GetCurrentPacket(&p) )
-		{
+		p = nullptr;
-		pkt->Assign(0, 0);
-		pkt->Assign(1, 0);
-		pkt->Assign(2, 0);
-		pkt->Assign(3, 0);
-		pkt->Assign(4, zeek::val_mgr->EmptyString());
-		pkt->Assign(5, zeek::BifType::Enum::link_encap->GetEnumVal(BifEnum::LINK_UNKNOWN));
-		return pkt;
-		}
 
-	pkt->Assign(0, static_cast<uint32_t>(p->ts.tv_sec));
+	return Packet::ToVal(p);
-	pkt->Assign(1, static_cast<uint32_t>(p->ts.tv_usec));
-	pkt->Assign(2, p->cap_len);
-	pkt->Assign(3, p->len);
-	pkt->Assign(4, zeek::make_intrusive<zeek::StringVal>(p->cap_len, (const char*)p->data));
-	pkt->Assign(5, zeek::BifType::Enum::link_encap->GetEnumVal(p->link_type));
-
-	return pkt;
 	%}
 
 ## Function to get the raw headers of the currently processed packet.

testing/btest/plugins/hooks-plugin/src/Plugin.cc

@@ -26,6 +26,7 @@ zeek::plugin::Configuration Plugin::Configure()
 	EnableHook(zeek::plugin::HOOK_SETUP_ANALYZER_TREE);
 	EnableHook(zeek::plugin::HOOK_LOG_INIT);
 	EnableHook(zeek::plugin::HOOK_LOG_WRITE);
+	EnableHook(zeek::plugin::HOOK_UNPROCESSED_PACKET);
 
 	zeek::plugin::Configuration config;
 	config.name = "Demo::Hooks";
@@ -272,3 +273,18 @@ bool Plugin::HookLogWrite(const std::string& writer, const std::string& filter,
 	fprintf(stderr, "%.6f %-15s %s %s\n", zeek::run_state::network_time, "| HookLogWrite", info.path, d.Description());
 	return true;
 	}
+
+void Plugin::HookUnprocessedPacket(const zeek::Packet* packet)
+	{
+	zeek::ODesc d;
+	d.Add("[");
+	d.Add("ts=");
+	d.Add(packet->time);
+	d.Add(" len=");
+	d.Add(packet->len);
+	d.Add(" header=");
+	d.AddN(reinterpret_cast<const char*>(packet->data), 10);
+	d.Add("]");
+
+	fprintf(stderr, "%.6f %-23s %s\n", zeek::run_state::network_time, "| HookUnprocessedPacket", d.Description());
+	}

testing/btest/plugins/hooks-plugin/src/Plugin.h

@@ -24,6 +24,7 @@ protected:
 	                  int num_fields, const zeek::threading::Field* const* fields,
 	                  zeek::threading::Value** vals) override;
 	void HookSetupAnalyzerTree(zeek::Connection *conn) override;
+	void HookUnprocessedPacket(const zeek::Packet* packet) override;
 	void MetaHookPre(zeek::plugin::HookType hook, const zeek::plugin::HookArgumentList& args) override;
 	void MetaHookPost(zeek::plugin::HookType hook, const zeek::plugin::HookArgumentList& args,
 	                  zeek::plugin::HookArgument result) override;