Metadata-Version: 2.4
Name: flowtap
Version: 0.1.0
Summary: NetFlow v5, NetFlow v9 and IPFIX collection and parsing, standard library only
Author: mjaksn
License-Expression: MIT
Project-URL: Homepage, https://github.com/mjaksn/flowtap
Project-URL: Source, https://github.com/mjaksn/flowtap
Project-URL: Issues, https://github.com/mjaksn/flowtap/issues
Keywords: netflow,ipfix,netflow9,flow,collector,network
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: System Administrators
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Networking :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# flowtap

[![CI](https://github.com/mjaksn/flowtap/actions/workflows/ci.yml/badge.svg)](https://github.com/mjaksn/flowtap/actions/workflows/ci.yml)
[![Release](https://github.com/mjaksn/flowtap/actions/workflows/release.yml/badge.svg)](https://github.com/mjaksn/flowtap/actions/workflows/release.yml)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)

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

```python
from flowtap import Collector

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

`flowtap` 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 — 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](#installing)
- [The two layers](#the-two-layers)
- [Collector](#collector)
- [Decoder and Message](#decoder-and-message)
- [The flow shape](#the-flow-shape)
- [Flow, the typed view](#flow-the-typed-view)
- [Events](#events)
- [Sampling](#sampling)
- [Export gaps](#export-gaps)
- [Counters](#counters)
- [Logging](#logging)
- [Hostname resolution](#hostname-resolution)
- [Parsing without a socket](#parsing-without-a-socket)
- [Sketch: logging flows to a database](#sketch-logging-flows-to-a-database)
- [Sketch: forwarding a subset over MQTT](#sketch-forwarding-a-subset-over-mqtt)
- [Protocol support](#protocol-support)
- [Tests](#tests)
- [Limitations](#limitations)
- [Roadmap](#roadmap)

---

## Installing

```bash
pip install .
```

Or copy the `flowtap/` directory next to your code. There is nothing to build
and nothing to fetch.

```python
import flowtap
flowtap.__version__          # "0.1.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](#parsing-without-a-socket).

---

## Collector

```python
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()`.

| | |
| --- | --- |
| `port`, `bind` | where to listen. Name an interface if only one faces the exporters. |
| `decoder` | an existing [`Decoder`](#decoder-and-message), to share template state or turn tracking off. |
| `timeout` | how long the blocking reads inside iteration wait before coming up for air. Bounds how quickly `stop()` is noticed; it is not a deadline on receiving. |
| `rcvbuf` | kernel receive buffer to request. None leaves the system default. |
| `sock` | an already-bound socket to use instead of making one. |

### Three ways to read it

```python
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.

```python
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

```python
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 `poll`ed. `close()` releases it and
is implied by leaving a `with` block.

| method | |
| --- | --- |
| `poll(timeout=0)` | one `Message` or None. `timeout=None` waits indefinitely. |
| `messages(timeout=None)` | endless iterator of `Message` |
| `flows(now=None)` | endless iterator of `Flow` |
| `stop()` | end the iterators; safe from any thread |
| `close()` | release the socket; idempotent |
| `address` | the `(host, port)` actually bound, which matters when `port=0` |
| `stats` | the decoder's [counters](#counters) |
| `fileno()` | the socket descriptor |

---

## 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.

```python
from flowtap 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`](#events).

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.

| `Message` | |
| --- | --- |
| `.header` | the message header dict |
| `.flows` | list of flow record dicts |
| `.options` | list of option record dicts — what the exporter says about itself |
| `.gap` | exports the sequence counter says never arrived before this message |
| `.sampling_rate` | the 1-in-N rate in force for this message's exporter and observation domain |
| `.exporter`, `.version`, `.sequence` | read off the header |
| `.typed_flows(now=None)` | the flows as [`Flow`](#flow-the-typed-view) objects |

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

| key | from | meaning |
| --- | --- | --- |
| `src_addr`, `dst_addr` | IE 8/12, 27/28 | address strings. **IPv4 and IPv6 share these keys** — one field carries either. A template carrying both families zero-fills the unused one, and the zeros never displace the populated family. |
| `src_port`, `dst_port` | IE 7, 11 | int |
| `proto` | IE 4 | IP protocol number |
| `octets`, `packets` | IE 1, 2 | this flow's counts |
| `octets_total`, `packets_total` | IE 85, 86 | **the same idea, different elements.** Exporters send one pair or the other. |
| `tcp_flags` | IE 6 | flags seen over the flow's life, as an int |
| `first_switched`, `last_switched` | IE 22, 21 | ms since the exporter booted (v5, v9) |
| `flow_start_ms`, `flow_end_ms` | IE 152, 153 | absolute epoch ms (IPFIX). Also `_s`, `_us`, `_ns` variants |
| `in_if`, `out_if` | IE 10, 14 | interface indices |
| `src_mask`, `dst_mask` | IE 9, 13 | prefix lengths |
| `src_as`, `dst_as` | IE 16, 17 | AS numbers |
| `next_hop` | IE 15, 62 | address string |
| `post_nat_src_addr`, `post_nat_dst_addr` | IE 225, 226 | **may be the only addresses present** |
| `tos`, `vlan`, `src_mac`, `dst_mac`, `if_name`, … | | see `flowtap/ie.py` for the full table |

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

### 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

| key | |
| --- | --- |
| `version` | 5, 9 or 10 |
| `exporter` | source address of the device, as a string |
| `domain` | observation domain; the engine ID for v5 |
| `sequence` | the export sequence counter |
| `unix_secs` | export time, epoch seconds |
| `sys_uptime` | ms since the exporter booted; None for IPFIX |

### 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 — 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:

```python
from flowtap 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.

| attribute | |
| --- | --- |
| `raw` | the parser's dict, not a copy. Treat as read-only. |
| `exporter`, `version`, `domain` | always present |
| `start` | flow start, unix epoch float. Always present — see below |
| `duration` | seconds, or None |
| `src_addr`, `dst_addr`, `src_port`, `dst_port`, `proto` | or None |
| `octets`, `packets`, `tcp_flags`, `in_if`, `out_if` | or None |
| `sampling_rate` | 1-in-N, or 1 |

| property / method | |
| --- | --- |
| `endpoints` | `(src, dst)` |
| `end` | `start + duration`, or None |
| `src_kind`, `dst_kind` | `"private"`, `"public"`, `"multicast"`, `"special"`, `"unknown"` |
| `is_external` | one end local, one not |
| `proto_name` | `"TCP"`, `"UDP"`, … or None |
| `service` | well known name for the **destination** port, or None |
| `flags` | `"...AP..."`, `""` if the exporter sent none |
| `started_at()` | `start` as an aware UTC datetime |
| `scaled(value)` | value corrected for the sampling rate; None in, None out |
| `get(key, default)` | any field of the record, modelled or not |
| `as_dict(include_raw=True)` | flat, JSON-safe dict |

`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 if it lands more than a day away — which is what a
wrapped uptime counter looks like — in favour of the export time.

### Getting out to plain data

```python
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` — `METADATA_KEYS`. They are underscored so 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 the console version would have printed is an object here.

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

| class | fields | |
| --- | --- | --- |
| `SamplingChange` | `exporter`, `domain`, `rate`, `previous` | a domain stated, or restated, how much it is leaving out |
| `ExportGap` | `exporter`, `domain`, `version`, `missed`, `unit` | exports that never arrived |
| `DecodeError` | `exporter`, `reason`, `detail` | a datagram that would not decode. `reason` is `"short"`, `"unsupported"` or `"malformed"` |

`take_events()` hands over what has accumulated and forgets it. Empty is the
normal answer on a healthy network, so it is cheap to call in a loop. 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.

---

## 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.

```python
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 — 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.

```python
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** — 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.

| key | |
| --- | --- |
| `packets` | datagrams received |
| `bytes_rx` | bytes received |
| `flows` | flow records decoded |
| `option_records` | option records decoded |
| `templates_new` | templates learned or changed |
| `deferred` | data sets dropped because their template had not arrived |
| `malformed` | datagrams too short, or with a truncated header |
| `unsupported_version` | not v5, v9 or IPFIX — usually sFlow on the wrong port |
| `parse_errors` | datagrams whose body would not parse |
| `missed_exports` | total gap, all streams |
| `v5_msgs`, `v9_msgs`, `v10_msgs` | messages per version |

`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.

---

## Logging

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

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

| logger | what |
| --- | --- |
| `flowtap.collector` | INFO once at bind |
| `flowtap.sampling` | WARNING when an exporter starts sampling, INFO when it stops |
| `flowtap.sequence` | WARNING on the first gap for an exporter |
| `flowtap.decoder` | DEBUG per undecodable datagram |
| `flowtap.names` | WARNING for an unreadable hosts file, DEBUG for failed lookups |

---

## Hostname resolution

**Optional, off by default, and not imported unless you ask for it.**

```python
from flowtap.names import Resolver

with Resolver(mode="dns", workers=4) as resolver:
    name = resolver.lookup("192.168.1.10")     # None until it is known
```

| mode | |
| --- | --- |
| `"off"` | the default. No lookups, no threads, no traffic. |
| `"dns"` | reverse DNS only. Still a query per address, to the resolver you already use. |
| `"all"` | reverse DNS, then mDNS to 224.0.0.251, then a NetBIOS status query to the host itself. **This sends probes onto the LAN.** |

`lookup()` only ever reads a TTL cache and returns immediately; misses are
queued for background workers. That is not an optimisation. UDP has no
backpressure, so anything that blocks a receive loop drops flows silently. The
first sighting of an address returns None and the name appears on a later one —
absence is "not known yet", not "has no name".

Public addresses are skipped unless `resolve_public=True`: a busy link produces
thousands, most resolve to something uninformative, and each is a query
somebody else can see. `hosts_files=[path]` loads static mappings, which answer
even in `"off"` mode.

It was kept because a database logger genuinely wants names — an IP address in
a year-old row is much less useful than a hostname — and because it is
self-contained. It was made opt-in because its widest mode does active network
probing, and a library that did that unasked, inside a daemon somebody
installed to log flows, would be doing something the caller never sanctioned on
a network they may not own.

---

## Parsing without a socket

```python
from flowtap 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 — that is what `deferred` counts.

Also exported, for a caller working at that level: `decode_value`,
`read_template_fields`, `record_min_length`, `parse_data_record`,
`flow_endpoints`, `flow_timestamp`, `flow_duration`.

---

## Sketch: logging flows to a database

Real `sqlite3`, so it runs, but the shape is what matters — 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.

```python
import signal
import sqlite3

from flowtap import Collector, MODELLED_FIELDS
from flowtap.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.
                record_gap(event.exporter, event.missed, event.unit)
            elif isinstance(event, SamplingChange):
                # Every count from this exporter is now 1/N of the truth.
                record_sampling(event.exporter, 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.

```python
import json

from flowtap 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

| version | header | templates | notes |
| --- | --- | --- | --- |
| NetFlow v5 | 24 bytes | none, fixed 48-byte records | fully decoded on the first packet |
| NetFlow v9 | 20 bytes | FlowSet 0 (data), 1 (options) | |
| IPFIX / v10 | 16 bytes | Set 2 (data), 3 (options) | enterprise fields and variable-length encoding |

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

```bash
python -m unittest discover
```

247 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.

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

```bash
python tools/fuzz.py --seconds 60
```

It mutates valid v5, v9 and IPFIX messages — bit flips, truncations, splices,
implausible length fields — and 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.
- **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.

---

## 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 — 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

- **Dual-stack collection.** An `AF_INET6` socket with `IPV6_V6ONLY` cleared,
  so exporters reaching the collector over IPv6 are decodable at all.
- **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 flowtap`.** 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 below 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](LICENSE).
