Move zeek-af_packet-plugin code into the main Zeek repository

This is based on commit b89a6f64123f778090d1dd6ec48e6b8e8906ea11 from
the zeek-af_packet-plugin repository.

.gitmodules

@@ -52,9 +52,6 @@
 [submodule "auxil/spicy"]
 	path = auxil/spicy
 	url = https://github.com/zeek/spicy
-[submodule "auxil/zeek-af_packet-plugin"]
-	path = auxil/zeek-af_packet-plugin
-	url = https://github.com/zeek/zeek-af_packet-plugin.git
 [submodule "auxil/libunistd"]
 	path = auxil/libunistd
 	url = https://github.com/zeek/libunistd

CMakeLists.txt

@@ -1189,18 +1189,6 @@ endif ()
 # Tell the plugin code that we're building as part of the main tree.
 set(ZEEK_PLUGIN_INTERNAL_BUILD true CACHE INTERNAL "" FORCE)
 
-set(ZEEK_HAVE_AF_PACKET no)
-if (${CMAKE_SYSTEM_NAME} MATCHES Linux)
-    if (NOT DISABLE_AF_PACKET)
-        if (NOT AF_PACKET_PLUGIN_PATH)
-            set(AF_PACKET_PLUGIN_PATH ${CMAKE_SOURCE_DIR}/auxil/zeek-af_packet-plugin)
-        endif ()
-
-        list(APPEND ZEEK_INCLUDE_PLUGINS ${AF_PACKET_PLUGIN_PATH})
-        set(ZEEK_HAVE_AF_PACKET yes)
-    endif ()
-endif ()
-
 set(ZEEK_HAVE_JAVASCRIPT no)
 if (NOT DISABLE_JAVASCRIPT)
     set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} ${PROJECT_SOURCE_DIR}/auxil/zeekjs/cmake)
@@ -1220,6 +1208,7 @@ if (NOT DISABLE_JAVASCRIPT)
     endif ()
 endif ()
 
+set(ZEEK_HAVE_AF_PACKET no CACHE INTERNAL "Zeek has AF_PACKET support")
 set(ZEEK_HAVE_JAVASCRIPT ${ZEEK_HAVE_JAVASCRIPT} CACHE INTERNAL "Zeek has JavaScript support")
 
 set(DEFAULT_ZEEKPATH_PATHS

auxil/zeek-af_packet-plugin

@@ -1 +0,0 @@
-Subproject commit b89a6f64123f778090d1dd6ec48e6b8e8906ea11

scripts/base/init-bare.zeek

@@ -5691,6 +5691,31 @@ export {
 	};
 }
 
+module AF_Packet;
+
+export {
+	## Size of the ring-buffer.
+	const buffer_size = 128 * 1024 * 1024 &redef;
+	## Size of an individual block. Needs to be a multiple of page size.
+	const block_size = 4096 * 8 &redef;
+	## Retire timeout for a single block.
+	const block_timeout = 10msec &redef;
+	## Toggle whether to use hardware timestamps.
+	const enable_hw_timestamping = F &redef;
+	## Toggle whether to use PACKET_FANOUT.
+	const enable_fanout = T &redef;
+	## Toggle defragmentation of IP packets using PACKET_FANOUT_FLAG_DEFRAG.
+	const enable_defrag = F &redef;
+	## Fanout mode.
+	const fanout_mode = FANOUT_HASH &redef;
+	## Fanout ID.
+	const fanout_id = 23 &redef;
+	## Link type (default Ethernet).
+	const link_type = 1 &redef;
+	## Checksum validation mode.
+	const checksum_validation_mode: ChecksumMode = CHECKSUM_ON &redef;
+}
+
 module DCE_RPC;
 
 export {

src/iosource/CMakeLists.txt

@@ -13,3 +13,4 @@ zeek_add_subdir_library(
     PktSrc.cc)
 
 add_subdirectory(pcap)
+add_subdirectory(af_packet)

src/iosource/af_packet/AF_Packet.cc

@@ -0,0 +1,326 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include "zeek/iosource/af_packet/AF_Packet.h"
+
+#include "zeek/iosource/af_packet/RX_Ring.h"
+#include "zeek/iosource/af_packet/af_packet.bif.h"
+
+#ifndef TP_STATUS_CSUM_VALID
+#define TP_STATUS_CSUM_VALID (1 << 7)
+#endif
+
+using namespace zeek::iosource::pktsrc;
+
+AF_PacketSource::~AF_PacketSource() { Close(); }
+
+AF_PacketSource::AF_PacketSource(const std::string& path, bool is_live) {
+    if ( ! is_live )
+        Error("AF_Packet source does not support offline input");
+
+    current_filter = -1;
+    props.path = path;
+    props.is_live = is_live;
+
+    socket_fd = -1;
+    rx_ring = nullptr;
+
+    checksum_mode = zeek::BifConst::AF_Packet::checksum_validation_mode->AsEnum();
+}
+
+void AF_PacketSource::Open() {
+    uint64_t buffer_size = zeek::BifConst::AF_Packet::buffer_size;
+    uint64_t block_size = zeek::BifConst::AF_Packet::block_size;
+    int block_timeout_msec = static_cast<int>(zeek::BifConst::AF_Packet::block_timeout * 1000.0);
+    int link_type = zeek::BifConst::AF_Packet::link_type;
+
+    bool enable_hw_timestamping = zeek::BifConst::AF_Packet::enable_hw_timestamping;
+    bool enable_fanout = zeek::BifConst::AF_Packet::enable_fanout;
+    bool enable_defrag = zeek::BifConst::AF_Packet::enable_defrag;
+
+    socket_fd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL));
+
+    if ( socket_fd < 0 ) {
+        Error(errno ? strerror(errno) : "unable to create socket");
+        return;
+    }
+
+    auto info = GetInterfaceInfo(props.path);
+
+    if ( ! info.Valid() ) {
+        Error(errno ? strerror(errno) : "unable to get interface information");
+        close(socket_fd);
+        socket_fd = -1;
+        return;
+    }
+
+    if ( ! info.IsUp() ) {
+        Error("interface is down");
+        close(socket_fd);
+        socket_fd = -1;
+        return;
+    }
+
+    // Create RX-ring
+    try {
+        rx_ring = new RX_Ring(socket_fd, buffer_size, block_size, block_timeout_msec);
+    } catch ( RX_RingException& e ) {
+        Error(errno ? strerror(errno) : "unable to create RX-ring");
+        close(socket_fd);
+        return;
+    }
+
+    // Setup interface
+    if ( ! BindInterface(info) ) {
+        Error(errno ? strerror(errno) : "unable to bind to interface");
+        close(socket_fd);
+        return;
+    }
+
+    if ( ! EnablePromiscMode(info) ) {
+        Error(errno ? strerror(errno) : "unable enter promiscuous mode");
+        close(socket_fd);
+        return;
+    }
+
+    if ( ! ConfigureFanoutGroup(enable_fanout, enable_defrag) ) {
+        Error(errno ? strerror(errno) : "failed to join fanout group");
+        close(socket_fd);
+        return;
+    }
+
+    if ( ! ConfigureHWTimestamping(enable_hw_timestamping) ) {
+        Error(errno ? strerror(errno) : "failed to configure hardware timestamping");
+        close(socket_fd);
+        return;
+    }
+
+    props.netmask = NETMASK_UNKNOWN;
+    props.selectable_fd = socket_fd;
+    props.is_live = true;
+    props.link_type = link_type;
+
+    stats.received = stats.dropped = stats.link = stats.bytes_received = 0;
+    num_discarded = 0;
+
+    Opened(props);
+}
+
+AF_PacketSource::InterfaceInfo AF_PacketSource::GetInterfaceInfo(const std::string& path) {
+    AF_PacketSource::InterfaceInfo info;
+    struct ifreq ifr;
+    int ret;
+
+    memset(&ifr, 0, sizeof(ifr));
+    snprintf(ifr.ifr_name, sizeof(ifr.ifr_name), "%s", path.c_str());
+
+    ret = ioctl(socket_fd, SIOCGIFFLAGS, &ifr);
+    if ( ret < 0 )
+        return info;
+
+    info.flags = ifr.ifr_flags;
+
+    ret = ioctl(socket_fd, SIOCGIFINDEX, &ifr);
+    if ( ret < 0 )
+        return info;
+
+    info.index = ifr.ifr_ifindex;
+
+    return info;
+}
+
+bool AF_PacketSource::BindInterface(const AF_PacketSource::InterfaceInfo& info) {
+    struct sockaddr_ll saddr_ll;
+    int ret;
+
+    memset(&saddr_ll, 0, sizeof(saddr_ll));
+    saddr_ll.sll_family = AF_PACKET;
+    saddr_ll.sll_protocol = htons(ETH_P_ALL);
+    saddr_ll.sll_ifindex = info.index;
+
+    ret = bind(socket_fd, (struct sockaddr*)&saddr_ll, sizeof(saddr_ll));
+    return (ret >= 0);
+}
+
+bool AF_PacketSource::EnablePromiscMode(const AF_PacketSource::InterfaceInfo& info) {
+    struct packet_mreq mreq;
+    int ret;
+
+    memset(&mreq, 0, sizeof(mreq));
+    mreq.mr_ifindex = info.index;
+    mreq.mr_type = PACKET_MR_PROMISC;
+
+    ret = setsockopt(socket_fd, SOL_PACKET, PACKET_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
+    return (ret >= 0);
+}
+
+bool AF_PacketSource::ConfigureFanoutGroup(bool enabled, bool defrag) {
+    if ( enabled ) {
+        uint32_t fanout_id = zeek::BifConst::AF_Packet::fanout_id;
+        uint32_t fanout_arg = ((fanout_id & 0xffff) | (GetFanoutMode(defrag) << 16));
+
+        if ( setsockopt(socket_fd, SOL_PACKET, PACKET_FANOUT, &fanout_arg, sizeof(fanout_arg)) < 0 )
+            return false;
+    }
+    return true;
+}
+
+bool AF_PacketSource::ConfigureHWTimestamping(bool enabled) {
+    if ( enabled ) {
+        struct ifreq ifr;
+        struct hwtstamp_config hwts_cfg;
+
+        memset(&hwts_cfg, 0, sizeof(hwts_cfg));
+        hwts_cfg.tx_type = HWTSTAMP_TX_OFF;
+        hwts_cfg.rx_filter = HWTSTAMP_FILTER_ALL;
+        memset(&ifr, 0, sizeof(ifr));
+        snprintf(ifr.ifr_name, sizeof(ifr.ifr_name), "%s", props.path.c_str());
+        ifr.ifr_data = &hwts_cfg;
+
+        if ( ioctl(socket_fd, SIOCSHWTSTAMP, &ifr) < 0 )
+            return false;
+
+        int opt = SOF_TIMESTAMPING_RAW_HARDWARE | SOF_TIMESTAMPING_RX_HARDWARE;
+        if ( setsockopt(socket_fd, SOL_PACKET, PACKET_TIMESTAMP, &opt, sizeof(opt)) < 0 )
+            return false;
+    }
+    return true;
+}
+
+uint32_t AF_PacketSource::GetFanoutMode(bool defrag) {
+    uint32_t fanout_mode;
+
+    switch ( zeek::BifConst::AF_Packet::fanout_mode->AsEnum() ) {
+        case BifEnum::AF_Packet::FANOUT_CPU: fanout_mode = PACKET_FANOUT_CPU; break;
+#ifdef PACKET_FANOUT_QM
+        case BifEnum::AF_Packet::FANOUT_QM: fanout_mode = PACKET_FANOUT_QM; break;
+#endif
+#ifdef PACKET_FANOUT_CBPF
+        case BifEnum::AF_Packet::FANOUT_CBPF: fanout_mode = PACKET_FANOUT_CBPF; break;
+#endif
+#ifdef PACKET_FANOUT_EBPF
+        case BifEnum::AF_Packet::FANOUT_EBPF: fanout_mode = PACKET_FANOUT_EBPF; break;
+#endif
+        default: fanout_mode = PACKET_FANOUT_HASH; break;
+    }
+
+    if ( defrag )
+        fanout_mode |= PACKET_FANOUT_FLAG_DEFRAG;
+
+    return fanout_mode;
+}
+
+void AF_PacketSource::Close() {
+    if ( socket_fd < 0 )
+        return;
+
+    delete rx_ring;
+    rx_ring = nullptr;
+
+    close(socket_fd);
+    socket_fd = -1;
+
+    Closed();
+}
+
+bool AF_PacketSource::ExtractNextPacket(zeek::Packet* pkt) {
+    if ( ! socket_fd )
+        return false;
+
+    struct tpacket3_hdr* packet = nullptr;
+    const u_char* data;
+    while ( true ) {
+        if ( ! rx_ring->GetNextPacket(&packet) )
+            return false;
+
+        current_hdr.ts.tv_sec = packet->tp_sec;
+        current_hdr.ts.tv_usec = packet->tp_nsec / 1000;
+        current_hdr.caplen = packet->tp_snaplen;
+        current_hdr.len = packet->tp_len;
+        data = (u_char*)packet + packet->tp_mac;
+
+        if ( ! ApplyBPFFilter(current_filter, &current_hdr, data) ) {
+            ++num_discarded;
+            DoneWithPacket();
+            continue;
+        }
+
+        pkt->Init(props.link_type, &current_hdr.ts, current_hdr.caplen, current_hdr.len, data);
+
+        if ( packet->tp_status & TP_STATUS_VLAN_VALID )
+            pkt->vlan = packet->hv1.tp_vlan_tci & 0x0fff;
+
+        switch ( checksum_mode ) {
+            case BifEnum::AF_Packet::CHECKSUM_OFF: {
+                // If set to off, just accept whatever checksum in the packet is correct and
+                // skip checking it here and in Zeek.
+                pkt->l4_checksummed = true;
+                break;
+            }
+            case BifEnum::AF_Packet::CHECKSUM_KERNEL: {
+                // If set to kernel, check whether the kernel thinks the checksum is valid. If it
+                // does, tell Zeek to skip checking by itself.
+                if ( ((packet->tp_status & TP_STATUS_CSUM_VALID) != 0) ||
+                     ((packet->tp_status & TP_STATUS_CSUMNOTREADY) != 0) )
+                    pkt->l4_checksummed = true;
+                else
+                    pkt->l4_checksummed = false;
+                break;
+            }
+            case BifEnum::AF_Packet::CHECKSUM_ON:
+            default: {
+                // Let Zeek handle it.
+                pkt->l4_checksummed = false;
+                break;
+            }
+        }
+
+        if ( current_hdr.len == 0 || current_hdr.caplen == 0 ) {
+            Weird("empty_af_packet_header", pkt);
+            return false;
+        }
+
+        stats.received++;
+        stats.bytes_received += current_hdr.len;
+        return true;
+    }
+
+    return false;
+}
+
+void AF_PacketSource::DoneWithPacket() { rx_ring->ReleasePacket(); }
+
+bool AF_PacketSource::PrecompileFilter(int index, const std::string& filter) {
+    return PktSrc::PrecompileBPFFilter(index, filter);
+}
+
+bool AF_PacketSource::SetFilter(int index) {
+    current_filter = index;
+    return true;
+}
+
+void AF_PacketSource::Statistics(Stats* s) {
+    if ( ! socket_fd ) {
+        s->received = s->bytes_received = s->link = s->dropped = 0;
+        return;
+    }
+
+    struct tpacket_stats_v3 tp_stats;
+    socklen_t tp_stats_len = sizeof(struct tpacket_stats_v3);
+    int ret;
+
+    ret = getsockopt(socket_fd, SOL_PACKET, PACKET_STATISTICS, &tp_stats, &tp_stats_len);
+    if ( ret < 0 ) {
+        Error(errno ? strerror(errno) : "unable to retrieve statistics");
+        s->received = s->bytes_received = s->link = s->dropped = 0;
+        return;
+    }
+
+    stats.link += tp_stats.tp_packets;
+    stats.dropped += tp_stats.tp_drops;
+
+    memcpy(s, &stats, sizeof(Stats));
+}
+
+zeek::iosource::PktSrc* AF_PacketSource::InstantiateAF_Packet(const std::string& path, bool is_live) {
+    return new AF_PacketSource(path, is_live);
+}

src/iosource/af_packet/AF_Packet.h

@@ -0,0 +1,82 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#pragma once
+
+extern "C" {
+#include <errno.h>            // errorno
+#include <linux/if.h>         // ifreq
+#include <linux/if_packet.h>  // AF_PACKET, etc.
+#include <linux/net_tstamp.h> // hwtstamp_config
+#include <linux/sockios.h>    // SIOCSHWTSTAMP
+#include <net/ethernet.h>     // ETH_P_ALL
+#include <pcap.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h> // close()
+}
+
+#include "zeek/iosource/PktSrc.h"
+#include "zeek/iosource/af_packet/RX_Ring.h"
+
+namespace zeek::iosource::pktsrc {
+
+class AF_PacketSource : public zeek::iosource::PktSrc {
+public:
+    /**
+     * Constructor.
+     *
+     * path: Name of the interface to open (the AF_Packet source doesn't
+     * support reading from files).
+     *
+     * is_live: Must be true (the AF_Packet source doesn't support offline
+     * operation).
+     */
+    AF_PacketSource(const std::string& path, bool is_live);
+
+    /**
+     * Destructor.
+     */
+    ~AF_PacketSource() override;
+
+    static PktSrc* InstantiateAF_Packet(const std::string& path, bool is_live);
+
+protected:
+    // PktSrc interface.
+    void Open() override;
+    void Close() override;
+    bool ExtractNextPacket(zeek::Packet* pkt) override;
+    void DoneWithPacket() override;
+    bool PrecompileFilter(int index, const std::string& filter) override;
+    bool SetFilter(int index) override;
+    void Statistics(Stats* stats) override;
+
+private:
+    Properties props;
+    Stats stats;
+
+    int current_filter = 0;
+    unsigned int num_discarded = 0;
+    int checksum_mode = 0;
+
+    int socket_fd = -1;
+    RX_Ring* rx_ring = nullptr;
+    struct pcap_pkthdr current_hdr = {};
+
+    struct InterfaceInfo {
+        int index = -1;
+        int flags = 0;
+
+        bool Valid() { return index >= 0; }
+        bool IsUp() { return flags & IFF_UP; }
+    };
+
+    InterfaceInfo GetInterfaceInfo(const std::string& path);
+    bool BindInterface(const InterfaceInfo& info);
+    bool EnablePromiscMode(const InterfaceInfo& info);
+    bool ConfigureFanoutGroup(bool enabled, bool defrag);
+    bool ConfigureHWTimestamping(bool enabled);
+    uint32_t GetFanoutMode(bool defrag);
+};
+
+} // namespace zeek::iosource::pktsrc

src/iosource/af_packet/CMakeLists.txt

@@ -0,0 +1,5 @@
+if (${CMAKE_SYSTEM_NAME} MATCHES Linux)
+    set(ZEEK_HAVE_AF_PACKET yes CACHE INTERNAL "")
+
+    zeek_add_plugin(Zeek AF_Packet SOURCES Plugin.cc AF_Packet.cc RX_Ring.cc BIFS af_packet.bif)
+endif ()

src/iosource/af_packet/Plugin.cc

@@ -0,0 +1,27 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include "zeek/plugin/Plugin.h"
+
+#include "zeek/iosource/Component.h"
+#include "zeek/iosource/af_packet/AF_Packet.h"
+
+namespace zeek::plugin::Zeek_AF_Packet {
+
+class Plugin : public plugin::Plugin {
+    plugin::Configuration Configure() override {
+        AddComponent(
+            new ::zeek::iosource::PktSrcComponent("AF_PacketReader", "af_packet",
+                                                  ::zeek::iosource::PktSrcComponent::LIVE,
+                                                  ::zeek::iosource::pktsrc::AF_PacketSource::InstantiateAF_Packet));
+
+        zeek::plugin::Configuration config;
+        config.name = "Zeek::AF_Packet";
+        config.description = "Packet acquisition via AF_Packet";
+        config.version.major = 4;
+        config.version.minor = 0;
+        config.version.patch = 0;
+        return config;
+    }
+} plugin;
+
+} // namespace zeek::plugin::Zeek_AF_Packet

src/iosource/af_packet/README

@@ -0,0 +1,127 @@
+
+# Zeek::AF_Packet
+
+This plugin provides native AF_Packet support for Zeek. For details about AF_Packet, see the corresponding [man page](http://man7.org/linux/man-pages/man7/packet.7.html).
+
+> **Note**:
+> Starting with Zeek version 5.2, Zeek ships with a built-in version of this plugin.
+
+## Installation
+
+Before installing the plugin, make sure your kernel supports PACKET_FANOUT[^1] and TPACKET_V3.
+
+### Package Manager
+
+The plugin is available as package for the [Zeek Package Manager](https://github.com/zeek/package-manager) and can be installed using the following command:
+
+    zkg install zeek-af_packet-plugin
+
+### Manual Install
+
+The following will compile and install the AF_Packet plugin alongside Zeek:
+
+    # ./configure && make && make install
+
+If everything built and installed correctly, you should see this:
+
+    # zeek -NN Zeek::AF_Packet
+    Zeek::AF_Packet - Packet acquisition via AF_Packet (dynamic, version 4.0.0)
+    [Packet Source] AF_PacketReader (interface prefix "af_packet"; supports live input)
+    [Type] AF_Packet::FanoutMode
+    [Type] AF_Packet::ChecksumMode
+    [Constant] AF_Packet::buffer_size
+    [Constant] AF_Packet::block_size
+    [Constant] AF_Packet::block_timeout
+    [Constant] AF_Packet::enable_hw_timestamping
+    [Constant] AF_Packet::enable_defrag
+    [Constant] AF_Packet::enable_fanout
+    [Constant] AF_Packet::fanout_mode
+    [Constant] AF_Packet::fanout_id
+    [Constant] AF_Packet::link_type
+    [Constant] AF_Packet::checksum_validation_mode
+
+## Usage
+
+Once installed, you can use AF_Packet interfaces/ports by prefixing them with `af_packet::` on the command line. For example, to use AF_Packet to monitor interface `eth0`:
+
+    # zeek -i af_packet::eth0
+
+### Permissions
+
+To use AF_Packet, running Zeek without root privileges, the Zeek processes need the CAP_NET_RAW capability. You can set it with the following command (on each sensor, after `zeekctl install`):
+
+    # setcap cap_net_raw+eip <path_to_zeek>/bin/zeek
+
+The AF_Packet plugin automatically enables promiscuous mode on the interfaces. As the plugin is using PACKET_ADD_MEMBERSHIP to enter the promiscuous mode without interfering others, the PROMISC flag is not touched. To verify that the interface entered promiscuous mode you can use `dmesg`.
+
+### Offloading
+
+Remember to disable any offloading features provided by the Network Interface Card (NIC) or Linux networking stack that interfere with Zeek. In general, Zeek expects to see network packets as they arrive on the wire. See this [blog post](https://blog.securityonion.net/2011/10/when-is-full-packet-capture-not-full.html) for more background. Toggling offloading features can be done with the `ethtool -K` command, for example:
+
+    # IFACE=eth0
+    # for offload in rx tx sg tso ufo gso gro lro; do
+    #   ethtool -K $IFACE $offload off
+    # done
+
+For more details around the involved offloads consult the [ethtool manpage](https://man7.org/linux/man-pages/man8/ethtool.8.html). In addition, `ethtool -S` can be used to gather statistics at the interface level.
+
+While all offloading should usually be disabled, the plugin supports to outsource certain tasks like checksum validation. See the [configuration section](#advanced-configuration) for further information.
+
+## Usage with `zeekctl`
+
+To use the AF_Packet plugin with `zeekctl`, the `custom` load balance method can be utilized. The following shows an exemplary configuration:
+
+    [manager]
+    type=manager
+    host=localhost
+
+    [proxy-1]
+    type=proxy
+    host=localhost
+
+    [worker-1]
+    type=worker
+    host=localhost
+    interface=af_packet::eth0
+    lb_method=custom
+    lb_procs=8
+    pin_cpus=0,1,2,3,4,5,6,7
+    # Optional parameters for per node configuration:
+    af_packet_fanout_id=23
+    af_packet_fanout_mode=AF_Packet::FANOUT_HASH
+    af_packet_buffer_size=128*1024*1024
+
+If all interfaces using `lb_method=custom` should be configured for AF_Packet, the prefix can be globally defined by adding the following line to `zeekctl.conf`:
+
+    lb_custom.InterfacePrefix=af_packet::
+
+## Advanced Configuration
+
+While the plugin aims at providing a "plug and play" user experience, it exposes several configuration options of the underlying API for customization (see [init.zeek](scripts/init.zeek) for the default values):
+
+* `buffer_size`: Set the overall buffer size allocated per socket. As the buffer is divided into blocks, this should be a multiple of the block size.
+* `block_size`: Set the size of a block in the buffer. Instead of ingesting packet by packet into the application, packets are aggregated in blocks to improve performance. The block size **must** be a multiple of the system's page size (see `getconf PAGE_SIZE`).
+* `block_timeout`: Set the timeout in milliseconds for passing a block to the application. This can be useful to reduce latency on less busy links.
+* `enable_hw_timestamping`: Enable support for hardware timestamping. Please note that this is an experimental feature.
+* `enable_defrag`: Enable defragmentation of IP packets before packets are load-balanced. This can be useful to prevent different fragments from being sent to different workers.
+* `enable_fanout`: Enable packet fanout for load-balancing across multiple workers. The load-balancing strategy is determined by the fanout mode.
+* `fanout_mode`: Set the load-balancing strategy. See [af_packet.bif](af_packet.bif) for the supported fanout modes.
+* `fanout_id`: Set the fanout ID that identifies a load-balancing group. When monitoring multiple interfaces, a separate ID has to be configured for each interface.
+* `link_type`: Set the link layer protocol.
+* `checksum_validation_mode`: Set how checksums are calculated and verified. See [af_packet.bif](af_packet.bif) for the supported validation modes.
+
+> **Note**:
+> Setting `checksum_validation_mode` will not have any effect when used with Zeek prior version 5.1.
+
+For further details on the above configuration options see the [kernel documentation](https://docs.kernel.org/networking/packet_mmap.html). Actual performance tuning is rather an art. For in-depth guidance, see the following resources, which can be transferred to Zeek:
+* [Suricata Extreme Performance Tuning guide](https://github.com/pevma/SEPTun)
+* [Suricata Extreme Performance Tuning guide - Mark II](https://github.com/pevma/SEPTun-Mark-II)
+
+## Limitations
+
+* __VLAN tagging is now supported.__ Even using AF_Packet's ``ETH_P_ALL``, the kernel removes VLAN tags from packets.
+  ~~While the tags are provided spereately, there is no efficient way to pass them to Zeek.~~ Applying knowledge about the internal data structures used by Zeek, the plugin now forwards VLAN tag control information to Zeek. Both IEEE 802.1Q and IEEE 802.1ad (QinQ) will be handled as expected.
+* Zeek workers crashing or restarting can, for a short period of time, disturb load balancing due to their packet
+  sockets being removed and later rejoining the fanout group. This may be visible in Zeek logs as gaps and/or duplicated connection entries produced by different Zeek workers.
+
+[^1]: Note that some kernel versions between 3.10 and 4.7 might exhibit a bug that prevents the required symmetric hashing. The script available at https://github.com/JustinAzoff/can-i-use-afpacket-fanout can be used to verify whether PACKET_FANOUT works as expected. This issue should have been fixed in all stable kernels by now.

src/iosource/af_packet/RX_Ring.cc

@@ -0,0 +1,97 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include "zeek/iosource/af_packet/RX_Ring.h"
+
+#include <cstring>
+#include <utility>
+
+extern "C" {
+#include <linux/if_packet.h> // AF_PACKET, etc.
+#include <sys/mman.h>        // mmap
+#include <sys/socket.h>      // socketopt consts
+#include <unistd.h>          // sysconf
+}
+
+RX_Ring::RX_Ring(int sock, size_t bufsize, size_t blocksize, int blocktimeout_msec) {
+    if ( sock < 0 )
+        throw RX_RingException("invalid socket");
+
+    // Configure socket
+    int ver = TPACKET_VERSION;
+    if ( setsockopt(sock, SOL_PACKET, PACKET_VERSION, &ver, sizeof(ver)) != 0 )
+        throw RX_RingException("unable to set TPacket version");
+
+    InitLayout(bufsize, blocksize, blocktimeout_msec);
+    if ( setsockopt(sock, SOL_PACKET, PACKET_RX_RING, (uint8_t*)&layout, sizeof(layout)) != 0 )
+        throw RX_RingException("unable to set ring layout");
+
+    // Map memory
+    size = static_cast<size_t>(layout.tp_block_size) * layout.tp_block_nr;
+    ring = (uint8_t*)mmap(nullptr, size, PROT_READ | PROT_WRITE, MAP_SHARED, sock, 0);
+    if ( ring == MAP_FAILED )
+        throw RX_RingException("unable to map ring memory");
+
+    block_num = packet_num = 0;
+    packet = nullptr;
+
+    // Init block mapping
+    blocks = new tpacket_block_desc*[layout.tp_block_nr];
+    for ( size_t i = 0; i < layout.tp_block_nr; i++ )
+        blocks[i] = (struct tpacket_block_desc*)(ring + i * layout.tp_block_size);
+}
+
+RX_Ring::~RX_Ring() {
+    ReleasePacket();
+
+    delete[] blocks;
+    munmap(ring, size);
+
+    blocks = nullptr;
+    size = 0;
+}
+
+bool RX_Ring::GetNextPacket(tpacket3_hdr** hdr) {
+    struct tpacket_hdr_v1* block_hdr = &(blocks[block_num]->hdr.bh1);
+
+    if ( (block_hdr->block_status & TP_STATUS_USER) == 0 )
+        return false;
+
+    if ( packet == nullptr ) {
+        // New block
+        packet_num = block_hdr->num_pkts;
+        if ( packet_num == 0 ) {
+            NextBlock();
+            return false;
+        }
+        packet = (struct tpacket3_hdr*)((uint8_t*)blocks[block_num] + block_hdr->offset_to_first_pkt);
+    }
+    else
+        // Continue with block
+        packet = (struct tpacket3_hdr*)((uint8_t*)packet + packet->tp_next_offset);
+
+    *hdr = packet;
+    packet_num--;
+    return true;
+}
+
+void RX_Ring::ReleasePacket() {
+    if ( packet_num == 0 )
+        NextBlock();
+}
+
+void RX_Ring::InitLayout(size_t bufsize, size_t blocksize, int blocktimeout_msec) {
+    memset(&layout, 0, sizeof(layout));
+    layout.tp_block_size = blocksize;
+    layout.tp_frame_size = TPACKET_ALIGNMENT << 7; // Seems to be irrelevant for V3
+    layout.tp_block_nr = bufsize / layout.tp_block_size;
+    layout.tp_frame_nr = (layout.tp_block_size / layout.tp_frame_size) * layout.tp_block_nr;
+    layout.tp_retire_blk_tov = blocktimeout_msec;
+}
+
+void RX_Ring::NextBlock() {
+    struct tpacket_hdr_v1* block_hdr = &(blocks[block_num]->hdr.bh1);
+
+    block_hdr->block_status = TP_STATUS_KERNEL;
+    block_num = (block_num + 1) % layout.tp_block_nr;
+    packet = nullptr;
+}

src/iosource/af_packet/RX_Ring.h

@@ -0,0 +1,44 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#pragma once
+
+extern "C" {
+#include <linux/if_packet.h> // AF_PACKET, etc.
+}
+
+#include <cstdint>
+#include <stdexcept>
+
+#define TPACKET_VERSION TPACKET_V3
+
+class RX_RingException : public std::runtime_error {
+public:
+    RX_RingException(const std::string& what_arg) : std::runtime_error(what_arg) {}
+};
+
+class RX_Ring {
+public:
+    /**
+     * Constructor
+     */
+    RX_Ring(int sock, size_t bufsize, size_t blocksize, int blocktimeout_msec);
+    ~RX_Ring();
+
+    bool GetNextPacket(tpacket3_hdr** hdr);
+    void ReleasePacket();
+
+protected:
+    void InitLayout(size_t bufsize, size_t blocksize, int blocktimeout_msec);
+    void NextBlock();
+
+private:
+    struct tpacket_req3 layout;
+    struct tpacket_block_desc** blocks;
+    struct tpacket3_hdr* packet;
+
+    unsigned int block_num;
+    unsigned int packet_num;
+
+    uint8_t* ring;
+    size_t size;
+};

src/iosource/af_packet/af_packet.bif

@@ -0,0 +1,15 @@
+
+# Options for the AF_Packet packet source.
+
+module AF_Packet;
+
+const buffer_size: count;
+const block_size: count;
+const block_timeout: interval;
+const enable_hw_timestamping: bool;
+const enable_defrag: bool;
+const enable_fanout: bool;
+const fanout_mode: FanoutMode;
+const fanout_id: count;
+const link_type: count;
+const checksum_validation_mode: ChecksumMode;

src/types.bif

@@ -247,4 +247,26 @@ enum Level %{
 	ERROR = 2,
 %}
 
+module AF_Packet;
+
+## Available fanout modes.
+enum FanoutMode %{
+	FANOUT_HASH, # PACKET_FANOUT_HASH
+	FANOUT_CPU,  # PACKET_FANOUT_CPU
+	FANOUT_QM,   # PACKET_FANOUT_QM
+	FANOUT_CBPF, # PACKET_FANOUT_CBPF
+	FANOUT_EBPF, # PACKET_FANOUT_EBPF
+%}
+
+## Available checksum validation modes.
+enum ChecksumMode %{
+	## Ignore checksums, i.e. always assume they are correct.
+	CHECKSUM_OFF,
+	## Let Zeek compute and verify checksums.
+	CHECKSUM_ON,
+	## Let the kernel handle checksum offloading.
+	## Note: Semantics may depend on the kernel and driver version.
+	CHECKSUM_KERNEL,
+%}
+
 module GLOBAL;