netflume

NetFlow v5, NetFlow v9 and IPFIX collection and parsing, as a library. Standard library only, no dependencies, Python 3.9 and up.

from netflume import Collector

with Collector(port=2055) as collector:
    for flow in collector.flows():
        print(flow.src_addr, "->", flow.dst_addr, flow.octets, "bytes")

netflume decodes the three protocols a router is likely to speak, tracks the templates v9 and IPFIX need in order to be decodable at all, and surfaces the two things an exporter says about itself that change what its numbers mean: the sampling rate, and gaps in the export sequence. It does not render, aggregate, store or alert. Those are the caller's business, and keeping them out is what makes the same package usable by a database writer, a message-bus forwarder and an offline analyser without any of them fighting the others.

Three things are optimised for, in this order when they conflict:

Performance. A busy edge router emits tens of thousands of flows a second, and a collector that falls behind loses data silently, because UDP has no retransmission. Decoding is allocation-conscious, the typed layer is opt-in rather than mandatory, and the counters needed to prove nothing was lost are maintained as part of normal operation rather than bolted on.

Versatility. The parser is separable from the socket: feed it from a capture, a queue or a test fixture. The socket layer offers a blocking iterator, a bounded poll and a raw fileno(), so it fits a thread, a loop of your own design, or a selectors reactor without adapters.

Ease of use. One import, four lines to a working collector. Field aliases that differ between exporters are resolved in one place instead of at every call site. Every diagnostic is an object you can branch on, and also a log record if that is all you want.

Contents

• Installing
• The two layers
• Collector
• Decoder and Message
• The flow shape
• Flow, the typed view
• Events
• Sampling
• Export gaps
• Counters
• Ceilings
• Logging
• Hostname resolution
• Parsing without a socket
• Sketch: logging flows to a database
• Sketch: forwarding a subset over MQTT
• Protocol support
• Tests
• Limitations
• Everything else exported
• Roadmap
• Licence

Installing

pip install netflume

Or from a checkout with pip install ., or by copying the netflume/ directory next to your code. There are no dependencies to resolve either way.

import netflume
netflume.__version__          # "0.3.0"

The two layers

Parsing turns bytes into records. No socket, no threads, no clock the caller cannot control. Feed it from a pcap, a queue, or a test fixture.

Collection binds a UDP socket and hands you what arrives.

Either works without the other. If you already have datagrams from somewhere, skip to Parsing without a socket.

Collector

Collector(port=2055, bind="0.0.0.0", decoder=None, timeout=1.0,
          rcvbuf=4*1024*1024, reuse_address=True, sock=None)

The socket is bound in the constructor, so a port already in use raises OSError there rather than at first read. Use with, or call close().

Three ways to read it

for rec, hdr in collector:            plain dicts, the cheapest form
for flow in collector.flows():        typed, built on demand
for message in collector.messages():  one per datagram
message = collector.poll(timeout=0)   one datagram, or None

__iter__ yields (record, header) rather than a bare record because a v5 or v9 record cannot be timestamped without its header. The flow start is milliseconds since the exporter booted, and turning that into a wall clock time needs the header's uptime and export time.

All three iterators are endless: a collector with nothing to report is a quiet network, not a finished job. Break out of the loop, or call stop().

Fitting into an existing event loop

Collector has a fileno(), so it goes straight into selectors or asyncio.loop.add_reader. Pair it with poll(timeout=0), which never blocks.

import selectors

sel = selectors.DefaultSelector()
sel.register(collector, selectors.EVENT_READ)

while True:
    for key, _ in sel.select(timeout=1.0):
        message = collector.poll(timeout=0)
        if message is not None:
            handle(message)

poll returns None both when nothing arrived and when what arrived would not decode. Neither is an error; read collector.decoder.take_events() if you want to know which.

Shutting down

import signal
signal.signal(signal.SIGTERM, lambda *_: collector.stop())

stop() is safe from a signal handler or another thread and returns immediately; the iterators finish within one timeout. It leaves the socket open, so a stopped collector can still be polled. close() releases it and is implied by leaving a with block.

Decoder and Message

Decoder is the parsing layer with its state attached: the template store, the sequence watch, the sampling watch. It has no socket in it.

from netflume import Decoder

decoder = Decoder()                  # track_sequence=True, track_sampling=True
message = decoder.decode(data, "10.0.0.1")

decode returns a Message, or None if the datagram was too short, carried an unsupported version, or would not parse. It never raises: a decoder that dies on one bad packet is useless on a real network. Every failure is counted and queued as a DecodeError.

A Message with no flows in it is not a failure. Template sets, option records, and data sets whose template has not arrived yet all produce one, and an exporter's first few datagrams are routinely all template. len(message) counts the flows; a message is never falsy, so if message: after a poll does not quietly discard the template-only ones.

decoder.flows(data, exporter, typed=False, now=None) is the short form for a caller who has bytes and wants only the records: it decodes and returns the flow list, or the Flow objects with typed=True, and gives an empty list rather than None when the datagram did not decode.

One Decoder serves every exporter sending to you. Templates, sequence tracking and sampling are all keyed by (exporter, observation domain) inside, because one chassis can run several domains, sample them differently and count their sequence numbers separately. It is not thread safe. Give each thread its own, or put one behind a queue.

The flow shape

A flow record is a plain dict. The keys are normalised, so v5, v9 and IPFIX produce the same key for the same idea and you write one schema, not three.

Every key is optional. Exporters send what they are configured to send, and absent is not zero: a flow with no octets is not a flow that carried nothing.

Common keys

netflume.IE is that table, {element id: (name, kind)}, and it is the whole list.

The microsecond and nanosecond timestamps are NTP

RFC 7011 §6.1.9 and §6.1.10 define dateTimeMicroseconds and dateTimeNanoseconds as 64-bit NTP timestamps: seconds since 1900 in the high word, and a fraction in units of 1/2³² in the low one. They are not counts since the UNIX epoch. dateTimeSeconds and dateTimeMilliseconds (IE 150 to 153) are plain epoch counts, so only two of the four pairs are affected.

netflume converts them at decode time, so flow_start_us holds UNIX seconds as a float, the same currency flow_timestamp deals in. The key name describes the precision the information element declares, not the unit of the value under it. Do not divide it by 1e6: reading the raw NTP word as microseconds puts the flow in the year 540,000 and makes every duration 2³²/1e6 = 4295× too long.

An all-zero timestamp field decodes to 0, meaning unset, rather than to midnight in 1900.

Keys you did not expect

Unrecognised elements are kept, not dropped:

• a standard element with no entry in the table → ie<id>, e.g. ie9999
• an enterprise-specific element → e<enterprise>.<id>, e.g. e9.33

A vendor field nobody has modelled is exactly what a forwarder might be built to ship, so nothing is thrown away to keep the shape tidy.

The header

Values

Addresses and MACs are strings; integers are ints; strings are decoded UTF-8 with trailing padding stripped. Anything the parser cannot make sense of becomes an int if it is eight bytes or fewer, and a hex string otherwise. All of it is JSON-native.

Flow, the typed view

The dict is the parser's output and stays that way. Flow is an opt-in wrapper over it, because a forwarder that wants three fields should not pay to build thirty. What it is actually for is not the annotations; it is the normalisations, each of which is otherwise open-coded at every call site:

from netflume import Flow

flow = Flow.from_record(rec, hdr, sampling_rate=1, now=None)

• flow.octets reads octets or octets_total. Same for packets. Read rec["octets"] alone and every IE 85 exporter contributes zero bytes: a wrong total rather than a missing one, which is the harder kind to notice.
• flow.src_addr falls back to the post-NAT address when there is no other.
• flow.start is derived from the record and the header, which is why from_record takes both.
• everything optional stays None, never a sentinel zero.

start is best effort and never None. IPFIX absolute timestamps win; failing those it is rebuilt from first_switched plus the header's uptime, and that reconstruction is rejected in favour of the export time if it lands more than a day away, which is what a wrapped uptime counter looks like.

Getting out to plain data

flow.as_dict()                    # every decoded field + metadata keys
flow.as_dict(include_raw=False)   # only MODELLED_FIELDS, a fixed column set

as_dict() adds _exporter, _version, _timestamp, _domain, _duration and _sampling_rate, which together are METADATA_KEYS. They are underscored so that they cannot collide with an information element name, present or future.

include_raw=False gives MODELLED_FIELDS and nothing else, which is the form a database table wants: no surprise columns when somebody plugs in a router that sends a vendor element.

Timestamps leave as unix epoch floats, deliberately. Every database and every message payload can store one without a serialiser, and started_at() is one call away for anyone who wants a datetime.

Events

Anything a console tool would have printed is an object here.

for event in decoder.take_events():
    ...

take_events() hands over what has accumulated and forgets it. Empty is the normal answer on a settled network, so it is cheap to call in a loop. It is not the answer at the start of one: the first datagram carrying each of an exporter's templates raises a TemplateLearned, so a caller that reads any event at all as trouble will be wrong about the first minute of every run. The queue holds MAX_PENDING_EVENTS and then drops the oldest, counting each drop in decoder.stats["events_dropped"]: a collector on an open UDP port receives things that are not NetFlow, and a caller that never drains must not accumulate one event per junk datagram for the life of the process. Each class has a __str__ that explains itself in a sentence, and each is also written to the log.

Only new and changed templates raise a TemplateLearned, exactly as only changed rates raise a SamplingChange. Exporters resend every template they hold every few minutes, which is what lets a collector that started late decode anything at all, and an event per resend would be an event per datagram from the exporters that prepend their templates to everything. A template evicted under MAX_TEMPLATES and later resent is new again, which is truthful: the layout was not held in between, and the data sets that arrived meanwhile are counted in deferred.

decoder.templates.take_events() is where those come from, and a caller driving parse_message rather than a Decoder reads them there instead. Decoder.decode drains the store on every datagram, including one whose body then failed to parse: a template set can be sound and the set behind it be what raised, and the layout learned from the first is still true.

Sampling

An exporter that samples reports one flow in N, and its byte and packet counts are correspondingly short. Nothing downstream can correct for that without knowing N.

decoder.sampling_rate("10.0.0.1", domain=0)   # 1000 means 1-in-1000. Never None.
flow.scaled(flow.octets)                      # the count corrected for it

Rates belong to an observation domain, not to a device, so name one. Omitting domain asks about the exporter as a whole and answers only when that is unambiguous: every domain heard from agrees, or only one has spoken. When they disagree it returns 1 rather than applying one domain's rate to another's counts.

The rate is read out of option records automatically, in every form exporters advertise it: the v5 header word, samplingInterval, samplerRandomInterval, IPFIX interval/space, and IPFIX size/population. A change raises a SamplingChange. An exporter repeating the same rate, which they do in every options record, raises one the first time and nothing after.

sampling_rate(rec) on its own, for a record you have in hand, returns None when the record says nothing about sampling and 1 when it explicitly describes an unsampled exporter. Those are different answers, and a caller holding a stale rate needs to tell them apart.

Export gaps

Every export message carries a sequence counter, and a jump in it means messages went missing on the way here, whether a saturated link or this process not keeping up. Without watching it the loss is silent: a flow that never arrives looks exactly like a flow that never happened.

message.gap              # exports missing before this message
decoder.export_gaps()    # cumulative, per stream, worst first

What the counter counts differs by version: v5 counts flow records, IPFIX counts data records, and v9 is specified to count export packets but is widely built to count records instead. Rather than trust the version, each stream is watched until one reading lands exactly on the next message, and that becomes the rule for that stream. Until one does, nothing is reported, since a wrong rule would invent a loss on every message, and a collector that cries wolf about dropped flows is worse than one that stays quiet. An exporter sending a single record per message is ambiguous forever and is watched without ever being judged.

Repeats, reordering and counter restarts are recognised and are not losses. Streams are keyed by (exporter, domain, version) and never summed together: adding a count of export messages to a count of data records gives a number that means nothing, which is why ExportGap carries its unit.

Counters

collector.stats / decoder.stats is a collections.Counter. Absent means zero.

deferred climbing at the start is normal and not a fault: v9 and IPFIX exporters resend templates periodically, often every few minutes, and data before the first one cannot be decoded by anybody.

Ceilings

Every table keyed by exporter has a ceiling. A UDP source address is whatever the sender typed, so anything keyed by one and never evicted is a memory leak that anyone able to reach the socket can pull on.

TemplateStore, SequenceWatch and SamplingWatch each take the limit as a constructor argument, so a collector facing an unusual number of exporters can raise it rather than patch the module. TemplateStore takes two, the second being max_pending.

A Decoder drains the store on every datagram, so MAX_PENDING_TEMPLATES binds nobody using one: it is there for a caller parsing without a Decoder and never draining, since an exporter alternating two layouts under one template ID raises an event every time it switches.

The defaults are far above any real deployment, since a collector sees tens of exporters, each with a handful of domains and templates. Eviction is not free when it happens. Evicting a template means the flows in its data sets are undecodable until the exporter resends it, commonly one to ten minutes later. Reading a template counts as using it, so a template still receiving data survives a flood of addresses that never send any. Watch decoder.templates.evicted, decoder.sequence.evicted and decoder.sampling.evicted; all three should stay at zero.

Logging

Everything goes to the netflume logger. The package installs a NullHandler and nothing else, so records go nowhere until you configure a handler.

import logging
logging.basicConfig(level=logging.INFO)

Hostname resolution

Not in this package. It lives in lanname, which turns an address into a name by reverse DNS, mDNS or NetBIOS, caches the answer, and never blocks the caller asking for it.

pip install lanname

from lanname import Resolver

with Resolver(mode="dns", workers=4) as resolver:
    name = resolver.lookup(flow["src_addr"])   # None until it is known

It used to be an optional module here, netflume.names, and it moved out because nothing in this package ever imported it and nothing about it was specific to flow records. Nothing here depends on it now; install it if a consumer wants names, and skip it otherwise. A database logger usually does want them, an IP address in a year-old row being much less useful than a hostname.

Two of its properties are worth knowing before it goes anywhere near a collector, and both are the reasons it is separate rather than wired in. lookup() reads a cache and returns immediately, because UDP has no backpressure and anything that blocks a receive loop drops flows silently. And its widest mode sends probes onto the LAN, which is why it does nothing at all until a mode is chosen.

Parsing without a socket

from netflume import TemplateStore, parse_message

store = TemplateStore()      # must outlive the datagram
for data, exporter in datagrams_from_a_pcap():
    hdr, flows, options = parse_message(data, exporter, store)

parse_message dispatches on the version and raises ValueError for one it does not decode, so an unsupported exporter and a truncated datagram stay distinguishable. parse_v5 and parse_v9_or_ipfix are there if you already know which you have.

The store must persist across datagrams. Exporters resend templates only periodically, and a data set arriving before its template cannot be decoded at all, which is what deferred counts.

Also exported, for a caller working with the records directly: flow_endpoints, flow_timestamp, flow_duration.

Sketch: logging flows to a database

Real sqlite3 rather than pseudocode, but the shape is what matters and record_gap and record_sampling are yours to write. Swap in whatever you actually use. MODELLED_FIELDS is the column set precisely so that a new exporter on the network does not mean a migration.

import signal
import sqlite3

from netflume import Collector, MODELLED_FIELDS
from netflume.events import ExportGap, SamplingChange

COLUMNS = ", ".join(MODELLED_FIELDS)
PLACEHOLDERS = ", ".join("?" * len(MODELLED_FIELDS))

db = sqlite3.connect("flows.db")
db.execute(f"CREATE TABLE IF NOT EXISTS flows ({COLUMNS})")

collector = Collector(port=2055)
signal.signal(signal.SIGTERM, lambda *_: collector.stop())

batch = []
with collector, db:
    for message in collector.messages():
        for flow in message.typed_flows():
            row = flow.as_dict(include_raw=False)
            batch.append([row[name] for name in MODELLED_FIELDS])

        # Batched: one commit per flow would spend the whole time in fsync,
        # and a stalled receive loop drops datagrams with no way to know.
        if len(batch) >= 500:
            db.executemany(
                f"INSERT INTO flows ({COLUMNS}) VALUES ({PLACEHOLDERS})", batch)
            db.commit()
            batch.clear()

        for event in collector.decoder.take_events():
            if isinstance(event, ExportGap):
                # Rows are missing and the table cannot show that by itself.
                # Record it, or the gap becomes a quiet lie about the traffic.
                # Store the whole stream key: gaps from different domains or
                # versions count different things and must never be summed.
                record_gap(event.exporter, event.domain, event.version,
                           event.missed, event.unit)
            elif isinstance(event, SamplingChange):
                # Every count from this domain is now 1/N of the truth. The
                # rate belongs to the domain, not to the device, so a second
                # domain would otherwise overwrite the first.
                record_sampling(event.exporter, event.domain, event.rate)

Three things worth doing that the sketch only gestures at:

• Store the sampling rate alongside the rows, not in a separate config. It changes while you are running, and a row is only interpretable with the rate that was in force when it arrived. flow.sampling_rate is already on each flow for this reason.
• Keep the writes off the receive loop once the flow rate is more than a home link. A bounded queue.Queue and a writer thread; drop on a full queue and count it, because blocking the reader loses datagrams silently, which is strictly worse than knowingly dropping a flow.
• Store raw too if you may ever want a field you did not model. A JSON column of flow.as_dict() costs space but means an exporter's vendor elements are still there when someone asks.

Sketch: forwarding a subset over MQTT

Pseudocode for the client; no MQTT library is a dependency here.

import json

from netflume import Collector

client = connect_to_broker()          # your MQTT client of choice

with Collector(port=2055) as collector:
    for rec, hdr in collector:
        # The dict form, not the typed one: three fields do not justify
        # building an object per flow, and a busy exporter sends tens of
        # thousands a second into a single-threaded loop.
        if rec.get("dst_port") not in WATCHED_PORTS:
            continue

        topic = f"netflow/{hdr['exporter']}/{rec.get('proto')}"
        client.publish(topic, json.dumps({
            "src": rec.get("src_addr"),
            "dst": rec.get("dst_addr"),
            "port": rec.get("dst_port"),
            "bytes": rec.get("octets", rec.get("octets_total")),
        }))

Notes for a real one:

• Filter before you serialise. Most flows are not interesting, and the cheapest flow is the one you never build a payload for.
• octets or octets_total, as above. If you find yourself writing that .get(..., .get(...)) more than once, use Flow and let it do it.
• Publish sampling changes on their own topic. A subscriber doing arithmetic on byte counts needs to know the multiplier changed, and it will never guess.
• Do not block the receive loop on the broker. A publish that waits for a reconnect stalls the reader, and UDP has no backpressure. Hand payloads to a bounded queue and let a publisher thread drain it.

Protocol support

Templates are keyed by (exporter address, observation domain, template ID). Different exporters reuse the same IDs for different layouts, so all three are needed.

IPFIX variable-length fields are handled in both forms: a declared length of 0xFFFF means a one-byte length prefix, or 0xFF followed by a two-byte length for values of 255 bytes or more.

Option records are decoded and returned separately from flows. They describe the exporter, not traffic; mixed into the flow list they look like flows with no addresses and inflate every count downstream. The sampling rate is read out of them automatically; interface names, application ID mappings and the rest are decoded and handed to you to use or ignore.

Set lengths, record lengths and template field counts are all bounds-checked. A malformed datagram is counted and discarded, never raised.

Tests

python -m unittest discover

273 tests, no dependencies, about a second. Several use subTest, so the number of individual checks is higher than the number of tests.

The suite is built around synthetic messages assembled byte by byte in tests/packets.py, which means the awkward cases are reachable: a template set carrying several templates, an options template whose scope length overruns its set, a data record truncated mid-field, a message that ends inside its own header. Every prefix of a full message is decoded, so anything that would only break on a fragment breaks in the suite first.

tests/test_hardening.py covers the cases that decode without raising and return something untrue, which is the failure class the fuzzer cannot see: NTP timestamps read as epoch counts, a truncated template fabricating flows, a dual-stack template erasing its own addresses, a sampling rate wiped by another observation domain, and the ceilings, which are asserted rather than assumed.

Decoder.decode promises never to raise, and that promise is attacked rather than trusted:

python tools/fuzz.py --seconds 60

It mutates valid v5, v9 and IPFIX messages with bit flips, truncations, splices and implausible length fields, then asserts nothing escapes. One decoder runs across the whole session, so template state accumulates and a later datagram meets whatever an earlier one left behind. A failure prints a hex reproducer and the seed that produced it.

Everything is synthetic. No real exporter was involved, so what the suite proves is conformance to the specifications and to its own reading of them, not agreement with any particular vendor's interpretation.

Limitations

• No sFlow. Different protocol, different wire format. It arrives on the same port often enough that unsupported_version is worth watching.
• No IPv6 transport. The socket is AF_INET. IPv6 addresses inside flow records decode fine; the exporter has to reach you over IPv4. On the roadmap.
• Single-threaded receive. One Collector, one socket, one reader. Under heavy load, move the socket read into its own thread feeding a bounded queue, and watch missed_exports, which is how you find out you needed to.
• Decoder is not thread safe. One per thread, or one behind a queue.
• Per-exporter state is capped and evicts. See Ceilings. The defaults are far above a real deployment, but a collector facing tens of thousands of distinct source addresses, or a spoofing flood, loses the least recently used templates and cannot decode their flows until the exporter resends.
• A truncated template is refused, not salvaged. A template set that ends mid-template is dropped rather than stored short, so the flows in that datagram are lost. The alternative is worse: a template missing fields cuts every later record for that ID into one real flow plus fabricated ones, and those reach the caller as genuine traffic until the next template refresh.
• No persistence, no aggregation, no alerting. By design; that is what the two sketches above are for. Template state is a different question and is on the roadmap. See below for why a restart currently costs you flows.
• Flow data is 5-tuple only. No SNI, no JA3, no DNS names, no payload. If you need those, NetFlow is the wrong data source and a mirror port is the right one.

Everything else exported

The sections above cover what most callers need. Everything below is public API too, listed here so that the export list and the documentation agree.

Names written bare are exported from the package itself, as in from netflume import service_name. Names written with a module path are exported from that module only, so from netflume.parse import NTP_EPOCH works and from netflume import NTP_EPOCH does not.

The ceiling constants are in Ceilings.

Roadmap

Planned, not promised, and ordered within each group by expected value. The three headings are the three things this package optimises for.

Performance

• Template-compiled unpacking. When every field in a template is fixed-length, which is the common case, the layout is known the moment the template is learned. Building one struct.Struct per template and unpacking a whole record in a single call replaces per-field slicing and integer conversion. This is the largest single decode win available and the reason the template store is a first-class object rather than a dictionary.
• memoryview over the datagram. Field extraction currently slices, and a slice copies. A view does not.
• Selective decoding. An optional field allow-list on the Decoder, so a forwarder that publishes five fields stops paying to decode forty. The wire offsets still have to be walked; the values do not have to be built.
• A benchmark suite, tracked in CI. Flows per second by version and by template shape, so the claims above become measurements and a regression shows up as a number rather than as a complaint.

Versatility

• Collection over IPv6. An AF_INET6 socket with IPV6_V6ONLY cleared, so exporters reaching the collector over IPv6 are decodable at all. This is about the transport; IPv6 addresses inside flow records already decode, and a template carrying both families is handled under Common keys.
• An asyncio interface. async for flow in collector for daemons already built around an event loop, without the thread-and-queue adapter every such caller currently writes.
• A persistent template store. v9 and IPFIX data records are undecodable until their template arrives, and exporters resend templates on their own schedule, commonly every one to ten minutes. A process restart therefore drops every flow until each exporter happens to resend. Saving and restoring the store removes that window, and matters more than anything else here for a long-running daemon.
• An enterprise element registry. Vendor elements decode today as e<pen>.<id> with raw bytes. Letting a caller register a name, a type and a length turns a vendor's flow direction or application ID into a first-class field.
• A capture source. Reading datagrams from a pcap or a flat file, for replaying a problem, testing a consumer against real traffic, and attaching something reproducible to a bug report.

Ease of use

• python -m netflume. A JSONL dump to stdout, enough to confirm an exporter is configured correctly and reaching you before any code is written against it. Not a console application; a smoke test.
• Wider modelling on Flow. More fields promoted from the raw dict, with the aliasing handled once, as the existing four are.
• Worked consumers. The database and MQTT sketches carried through to runnable examples, including the parts that are easy to get wrong: batching, back-pressure, and what to do when the sink is slower than the exporter.

Explicit non-goals: sFlow, which is a different protocol rather than a variant; aggregation, alerting and storage, which belong to the consumer; and any form of presentation.

Licence

MIT. See LICENSE.