netflower

High-performance network flow extractor for edge devices.

netflower extracts bidirectional network flows — either from a live interface or from .pcap / .pcapng files — and produces 82 features compatible with the CICFlowMeter feature set.

Features

• Live capture — capture flows in real time from a network interface; flows are emitted only when complete (TCP FIN/RST or idle timeout), never cut by an arbitrary boundary
• PCAP flows to CSV — convert .pcap / .pcapng files to flow-based CSV, with optional parallel processing
• Edge-optimized — dpkt for packet parsing (~10–25x faster than Scapy on ARM), Welford's online algorithm for O(1) memory per flow, batch-buffered CSV output
• No extra pip dependencies for live capture — libpcap is accessed via ctypes

Installation

pip install netflower

Live capture requires libpcap on the system:

# Linux
sudo apt install libpcap-dev

# macOS
brew install libpcap

Usage

Live capture

Captures packets from a network interface and emits each completed flow via callback.

from netflower import capture_live

def on_flow(flow: dict):
    print(
        f"{flow['src_ip']}:{flow['src_port']} -> "
        f"{flow['dst_ip']}:{flow['dst_port']} | "
        f"duration={flow['flow_duration']:.3f}s  "
        f"pkts={flow['tot_fwd_pkts'] + flow['tot_bwd_pkts']}"
    )

handle = capture_live("eth0", on_flow=on_flow)
handle.start()

# ... rest of your program ...

handle.stop()

As a context manager:

import time
with capture_live("eth0", on_flow=on_flow) as handle:
    handle.start()
    time.sleep(60)

Save the raw packets of each completed flow to individual .pcap files:

handle = capture_live(
    "eth0",
    on_flow=on_flow,
    save_pcap=True,
    pcap_dir="/tmp/flows",
)
handle.start()

Note: live capture requires root or CAP_NET_RAW privilege.

PCAP flows to CSV

Converts a .pcap or .pcapng file into a flow-based CSV.

from netflower import convert_pcap_to_csv

# Single-process
n = convert_pcap_to_csv("capture.pcap", "flows.csv")
print(f"Extracted {n} flows")

# Parallel -- use all available CPUs
n = convert_pcap_to_csv("capture.pcap", "flows.csv", n_jobs=-1)

API

capture_live(interface, on_flow, **kwargs) -> CaptureHandle

CaptureHandle exposes .start(), .stop(), and context manager support.

convert_pcap_to_csv(input_path, output_path, **kwargs) -> int

Returns the number of flow rows written.

Architecture

Both APIs share the same internal pipeline:

Packet source
    |
    v
_parser.py          parse Ethernet/IP/TCP/UDP frame
                    discard non-IPv4-TCP/UDP packets
    |
    v
FlowSession         hash table of active flows
                    route each packet to the correct Flow
                    (forward or backward direction)
    |
    v
Flow                accumulate statistics incrementally -- O(1) per packet,
                    O(1) memory (Welford's online algorithm, no lists stored)
    |
    v  on timeout / FIN / RST / flush_all
Writer              emit the 82-feature dict

Live capture internals

capture_live wraps libpcap via ctypes (no compiled extension required):

1. pcap_open_live opens the interface in promiscuous mode (snaplen=65535, to_ms=1000).
2. A PcapHandler_cb C-callable is registered and pcap_loop(-1, ...) is launched in a daemon thread.
3. Each callback invocation reads raw bytes with ctypes.string_at(pkt_data, hdr.caplen) and forwards the packet to FlowSession.
4. Garbage collection runs every 1 000 packets (session.gc(ts)) to evict flows that exceeded flow_timeout.
5. handle.stop() calls pcap_breakloop (thread-safe), waits for the thread (timeout 5 s), flushes remaining flows, and closes the handle.

When save_pcap=True, a _PcapSavingWriter buffers raw (PktHdr, bytes) per flow key and writes a dedicated .pcap file via pcap_dump_open / pcap_dump / pcap_dump_close at the moment each flow is emitted.

PCAP converter internals

Single-process mode (n_jobs=1): packets are read sequentially via dpkt, parsed, and fed directly into a FlowSession -> CsvWriter chain.

Parallel mode (n_jobs > 1):

main process
  reads packets sequentially
  routes each packet by hash(bidirectional_key) % n_jobs
       |              |              |
       v              v              v
  worker 0       worker 1  ...  worker N-1
  FlowSession    FlowSession    FlowSession
  temp_0.csv     temp_1.csv     temp_N.csv
       |              |              |
       +——————————————+——————————————+
                      |
                 _merge_csvs
                      |
                 flows.csv

The routing hash is bidirectionally deterministic — min(a,b), max(a,b), proto — so forward and backward packets of the same flow always land on the same worker. GC messages are broadcast to all workers every gc_interval packets.

Key constants

Flow features

Both APIs produce the same 82 features per flow:

Supported formats

• pcap -- standard libpcap format
• pcapng -- next-generation capture format

Only IPv4 TCP and UDP flows are extracted; other protocols are silently skipped.

License

MIT -- see LICENSE.