netflow-capture

Modules

A reusable NixOS module that turns nfdump's nfpcapd into a declarative, per-interface NetFlow/IPFIX capture service. You describe a set of named listeners; the module renders one hardened nfpcapd-<name> systemd unit per interface, auto-creates the on-disk storage layout, and manages the capture-output user.

What problem it solves

nfpcapd records live traffic off a NIC and writes rotating NetFlow/IPFIX flow files (and, optionally, raw pcap) that you can later query with nfdump. Wiring it up by hand means: one long argv per interface, a system user for the output, pre-created directories (it will not make them for you), and a systemd unit that opens a raw socket but doesn't leave itself running as root longer than needed.

This module collapses all of that into an option tree. Add a listener attribute and you get a fully-formed, restart-on-failure capture service.

The traps it encodes

Two things about nfpcapd are easy to get wrong, and the module bakes in the right answer:

1. It runs as User=root, on purpose, even though it's handed -u/-g. Opening a raw AF_PACKET capture socket needs root. nfpcapd opens the socket first, then drops privilege to the configured user itself. If you "fix" the unit to run directly as the unprivileged user, the socket open fails with EPERM. The system user only ever owns the on-disk output — never the running process at startup.

2. nfpcapd will not create missing output directories. If the -w path (or the raw-pcap -p path) doesn't exist, the daemon exits immediately. The module therefore emits systemd.tmpfiles rules to pre-create the base storage dir, each listener's per-interface subdirectory, and any pcap sidecar dir, owned by the capture user.

Everything else — worker threads, socket buffer, node cache, snap length, active and inactive flow-expiration windows, and the file-rotation window — is a per-listener option so you can tune each interface independently.

Usage

Import default.nix as a NixOS module, then:

{
  services.nfpcapd = {
    enable = true;

    # Where flow files land. One subdirectory per listener is created underneath.
    storageDir = "/var/log/netflow";

    listeners = {
      # Attribute name is the listener name AND the default interface + subdir name.
      eth0 = {
        enable = true;
        subdirectory = "wired";
      };

      wlo1 = {
        enable = true;
        workerThreads = 4;      # bump when compression is enabled at high levels
        socketBufferMB = 64;    # raise on high-throughput links to avoid drops
        rotateTime = 60;        # rotate flow files every 60s
      };

      # Capture only DNS, and also keep raw pcap on the side.
      dns = {
        enable = true;
        interface = "eth0";
        subdirectory = "dns";
        capturePcapDirectory = "/var/log/nfpcapd/pcap";
        additionalOptions = "'port 53 and proto udp'";
      };
    };
  };
}

Query the captured flows afterwards with nfdump -R /var/log/netflow/wired.

Options

Top-level (services.nfpcapd):

Option	Default	Meaning
enable	false	Master switch. Asserts at least one listener is configured.
storageDir	/var/log/netflow	Base output dir; one subdir per listener underneath.
user / group	nfpcapd	System user/group that owns the capture output.
listeners	{}	Attrset of named listeners (see below).
globalAdditionalOptions	""	Extra argv appended to every listener.

Per listener (services.nfpcapd.listeners.<name>):

Option	Default	Meaning
enable	false	Enable this listener's unit.
interface	listener name	NIC to capture from (-i).
subdirectory	listener name	Output subdir under storageDir.
capturePcapDirectory	null	If set, also write raw pcap here (-p).
workerThreads	2	Worker threads (-W). Keep ≤ logical core count.
socketBufferMB	20	Capture socket buffer in MB (-b).
nodeCacheSize	524288	Flow node cache size in bytes (-B).
snaplen	1522	Snapshot length (-s).
activeExpirationSeconds	300	Active flow expiry (-e first field).
inactiveExpirationSeconds	60	Inactive flow expiry (-e second field).
rotateTime	300	File rotation window, seconds or an nfdump time expr (-t).
verboseMode	false	Echo captured data to stdout (-E); debugging only.
additionalOptions	""	Extra argv for this listener (BPF filter, compression, etc.).

Caveats

• Requires the nfdump package (pulled from pkgs at build time).
• The service opens raw capture sockets — it needs to run on the host with the NIC, not inside a network namespace that can't see the traffic.
• High packet rates want a larger socketBufferMB (and possibly nodeCacheSize) to avoid kernel-side drops; short rotateTime values create many small files.
• RestartPreventExitStatus = 0 means a clean exit (status 0) is treated as intentional and the unit is not restarted; any failure restarts after 5s.

Source

modules/netflow-capture/default.nix

# netflow-capture — declarative per-interface NetFlow/IPFIX capture with nfdump's nfpcapd.
#
# Turns `nfpcapd` (from the `nfdump` package) into a NixOS service that models an
# arbitrary set of named packet-capture listeners as a submodule option tree. Each
# listener becomes its own `nfpcapd-<name>` systemd unit with independent worker
# threads, socket-buffer size, flow-expiration windows, rotation window, and an
# optional raw-pcap sidecar. Storage directories are auto-created via tmpfiles.
#
# Import it and set `services.nfpcapd.enable = true;` plus at least one listener.
#
# The two traps that make this non-obvious are documented inline below:
#   1. The unit runs as User=root even though nfpcapd is handed -u/-g: root is
#      required to open the raw capture socket, then nfpcapd drops privilege to
#      the unprivileged user itself.
#   2. nfpcapd will NOT create missing output paths, so every listener's output
#      subdir must be pre-created by tmpfiles or the daemon exits immediately.
{
  lib,
  config,
  pkgs,
  ...
}:
let
  inherit (lib)
    mkOption
    types
    mkIf
    mkEnableOption
    ;
  cfg = config.services.nfpcapd;

  interfaceOpts = _: {
    options = {
      enable = mkEnableOption "Enable this listener";
      workerThreads = mkOption {
        description = "Number of worker threads to spawn. Helps when compression is enabled at high levels. Should not be higher than the number of logical cores of the listening machine.";
        default = 2;
        type = types.int;
      };

      socketBufferMB = mkOption {
        description = "set socket buffer size in MB (default 20MB)";
        default = 20;
        type = types.int;
      };

      nodeCacheSize = mkOption {
        description = "set node cache size in bytes (default 524288)";
        default = 524288;
        type = types.int;
      };

      interface = mkOption {
        description = "Interface to listen to. Defaults to the name of the listener.";
        default = null;
        type = types.nullOr types.str;
      };

      subdirectory = mkOption {
        description = "Define a subdirectory structure for captured files. Defaults to the listener's name";
        example = ''
          nfpcapd.subdirectory = "wired";
        '';
        default = null;
        type = types.nullOr types.str;
      };

      capturePcapDirectory = mkOption {
        description = "If set, also record raw pcap files to this target directory.";
        example = ''
          nfpcapd.capturePcapDirectory = "/var/log/nfpcapd/pcap";
        '';
        default = null;
        type = types.nullOr types.str;
      };

      snaplen = mkOption {
        description = "set the snapshot length (default 1522)";
        default = 1522;
        type = types.int;
      };

      activeExpirationSeconds = mkOption {
        description = "Set the active flow expire time in seconds (default 300)";
        default = 300;
        type = types.int;
      };

      inactiveExpirationSeconds = mkOption {
        description = "Set the inactive flow expire time in seconds (default 60)";
        default = 60;
        type = types.int;
      };

      rotateTime = mkOption {
        description = "Time window (seconds, or an nfdump time expression) to rotate pcap/nfcapd files.";
        default = 300;
        type = types.either types.int types.str;
      };

      verboseMode = lib.mkEnableOption "Whether to output capture data to stdout. Only use for debugging.";
      additionalOptions = mkOption {
        description = "Additional command line options to be passed to this nfpcapd listener";
        default = "";
        example = ''
          nfpcapd.additionalOptions = "-o fat,payload -z=lzo " + (
            if shouldOnlyCaptureDNSUDPPackets then "'port 53 and proto udp'" else ""
          );
        '';
        type = types.str;
      };
    };
  };
in
{
  options = {
    services.nfpcapd = {
      enable = mkOption {
        description = "Enable the capture of NetFlow data with nfpcapd";
        default = false;
        example = ''
          services.nfpcapd = {
            enable = true;
            listeners = {
              "wlo1" = {
                enable = true;
                subdirectory = "wireless";
              };
            };
          };
        '';
        type = types.bool;
      };
      storageDir = mkOption {
        description = "Folder to save capture data to. One subdirectory per listener is created underneath it.";
        default = "/var/log/netflow";
        type = types.str;
      };
      user = mkOption {
        description = "User that owns the on-disk capture output. nfpcapd drops to this user after opening the raw socket.";
        default = "nfpcapd";
        type = types.str;
      };
      group = mkOption {
        description = "Group for the on-disk capture output.";
        default = "nfpcapd";
        type = types.str;
      };

      listeners = mkOption {
        description = "Configuration of interfaces to listen to. One nfpcapd-<name> systemd unit is created per attribute.";
        default = { };
        example = ''
          eth0 = {
            enable = true;
            subdirectory = "wired";
          };
          wlo1 = {
            enable = true;
          };
        '';
        type = with types; attrsOf (submodule interfaceOpts);
      };
      globalAdditionalOptions = mkOption {
        description = "Additional command line options to be passed to all nfpcapd listeners";
        default = "";
        example = ''
          nfpcapd.globalAdditionalOptions = "-z=lzo " + (
            if shouldOnlyCaptureDNSUDPPackets then "'port 53 and proto udp'" else ""
          );
        '';
        type = types.str;
      };
    };
  };
  config = mkIf cfg.enable {
    assertions = [
      {
        assertion = builtins.length (builtins.attrNames cfg.listeners) > 0;
        message = "At least one network interface must be configured in services.nfpcapd.listeners.";
      }
    ];

    # nfpcapd refuses to create missing output directories, so pre-create the base
    # storage dir, every listener's per-interface subdir, and any raw-pcap sidecar dir.
    systemd.tmpfiles.rules = [
      "d ${cfg.storageDir} 0750 ${cfg.user} ${cfg.group}"
    ]
    ++ (
      with builtins;
      (filter (_: _ != null) (
        (attrValues (
          mapAttrs (
            _: interfaceCfg:
            if interfaceCfg.capturePcapDirectory != null then
              "d ${interfaceCfg.capturePcapDirectory} 0750 ${cfg.user} ${cfg.group}"
            else
              null
          ) cfg.listeners
        ))
        ++ (attrValues (
          mapAttrs (
            interfaceName: interfaceCfg:
            "d ${cfg.storageDir}/${
              if interfaceCfg.subdirectory != null then interfaceCfg.subdirectory else interfaceName
            } 0750 ${cfg.user} ${cfg.group}"
          ) cfg.listeners
        ))
      ))
    );
    users.users.${cfg.user} = {
      isSystemUser = true;
      inherit (cfg) group;
    };
    users.groups.${cfg.group} = { };

    systemd.services = builtins.listToAttrs (
      builtins.attrValues (
        builtins.mapAttrs (
          listener: interfaceCfg:
          let
            interface = if interfaceCfg.interface != null then interfaceCfg.interface else listener;
          in
          {
            name = "nfpcapd-${listener}";
            value = {
              inherit (interfaceCfg) enable;
              description = "IPFIX/NetFlow capture daemon (nfpcapd)";
              after = [ "network.target" ];
              wantedBy = [ "multi-user.target" ];
              serviceConfig = {
                ExecStart = ''
                  ${pkgs.nfdump}/bin/nfpcapd \
                    -u ${cfg.user} \
                    -g ${cfg.group} \
                    -i ${interface} \
                    -b ${toString interfaceCfg.socketBufferMB}MB \
                    -B ${toString interfaceCfg.nodeCacheSize} \
                    -w ${cfg.storageDir}/${
                      if interfaceCfg.subdirectory != null then interfaceCfg.subdirectory else listener
                    } \
                    -W ${toString interfaceCfg.workerThreads} \
                    -s ${toString interfaceCfg.snaplen} \
                    -e ${toString interfaceCfg.activeExpirationSeconds},${toString interfaceCfg.inactiveExpirationSeconds} \
                    -t ${
                      if lib.isInt interfaceCfg.rotateTime then
                        (toString interfaceCfg.rotateTime)
                      else
                        interfaceCfg.rotateTime
                    } \
                    ${
                      if interfaceCfg.capturePcapDirectory != null then "-p " + interfaceCfg.capturePcapDirectory else ""
                    } ${
                      if interfaceCfg.verboseMode then "-E" else ""
                    } ${interfaceCfg.additionalOptions} ${cfg.globalAdditionalOptions}'';
                Type = "simple";
                # Root is required to open the raw capture socket; nfpcapd then
                # drops to -u/-g itself. Do NOT set User=${cfg.user} here or the
                # socket open fails with EPERM.
                User = "root";
                Group = "root";
                ProtectHome = true;
                PrivateTmp = true;
                ProtectKernelModules = true;
                ProtectKernelTunables = true;
                Restart = "on-failure";
                RestartPreventExitStatus = 0;
                RestartSec = 5;
              };
            };
          }
        ) cfg.listeners
      )
    );
  };
}