# netring examples
All ~45 examples are organized by topic. Example *names* (what you
pass to `cargo run --example <name>`) are stable across the
reorganization — only the file paths inside this directory changed.
Most examples need `CAP_NET_RAW` on the test binary. Use `just
setcap` once to grant capabilities, then run examples as your
normal user.
---
## basic/ — sync capture, TX, bridge
The lowest-level synchronous API. Start here if you want to
understand the building blocks before reaching for the async
wrappers.
| `capture` | Plain `Capture::open` → batch loop |
| `inject` | Plain `Injector::open` → `send` |
| `tx_replay` | `AsyncInjector::send_stream` + `TxPacer` rate pacing + egress `tx_timestamps` (Phase D) |
| `bridge` | Forward packets between two interfaces sync |
| `batch_processing` | Walking a `PacketBatch` with sequence-gap detection |
| `low_latency` | `LowLatency` ring profile + busy-poll |
| `stats_monitor` | Live `Capture::stats()` polling loop |
| `channel_consumer` | `ChannelCapture` (runtime-agnostic, no tokio) |
| `pcap_write` | Sync capture → `CaptureWriter` pcap file |
| `dpi` | Simple deep-packet-inspection demo |
## async_basics/ — async wrappers, no flow features
The async API surface. Use these as templates for the typical
`AsyncCapture::open(iface)?` recipe.
| `async_capture` | `AsyncCapture::readable().await` loop |
| `async_inject` | `AsyncInjector::send().await` with backpressure |
| `async_bridge` | Async bidirectional bridge |
| `async_stream` | `into_stream()` + `futures::StreamExt` |
| `async_streamext` | Stream combinator usage (`map`, `filter`, …) |
| `async_pipeline` | tokio `mpsc` pipeline with worker pool |
| `async_signal` | Ctrl-C graceful shutdown |
| `async_lo_dedup` | `Dedup::loopback()` for `lo` capture |
| `async_stats_monitor` | `StreamCapture::capture_stats()` (plan 20) |
| `async_metrics` | `metrics` crate counters |
## filter/ — BPF and eBPF filtering
| `bpf_filter` | Typed `BpfFilter::builder()` end-to-end |
| `ebpf_filter` | `aya`-loaded eBPF socket filter |
| `async_filter` | `AsyncCapture::open_with_filter` + runtime swap |
## scaling/ — fanout and multi-source
| `fanout` | Sync `Capture` joining a `PACKET_FANOUT` group |
| `async_fanout_workers` | `AsyncMultiCapture::open_workers` |
| `async_multi_interface` | `AsyncMultiCapture::open(&["lo", "eth0"])` |
See [`docs/scaling.md`](../docs/scaling.md) for the fanout decision
matrix and anti-patterns.
## xdp/ — AF_XDP kernel bypass
| `xdp_send` | TX-only AF_XDP via `XdpMode::Tx` |
| `async_xdp` | Default `AsyncXdpSocket` recipe |
| `async_xdp_busy_poll` | Busy-poll trio for latency |
| `async_xdp_self_loaded` | `with_default_program()` — no external XDP loader; `PROMISC=1` for promiscuous capture (issue #4) |
| `async_xdp_custom_program` | `with_program(prog)` for a caller-loaded XDP program |
| `xdp_multiqueue` | full-NIC capture via **`XdpCapture`** — one socket per RX queue + promiscuous (issues #4/#6) |
| `xdp_sharded` | **`XdpShardedRunner`** — one `Monitor` per RX queue (worker-per-core, busy-poll), the line-rate tier (issue #6) |
## flow/ — flow tracking (no L7 parsing)
| `async_flow_keys` | Bare `FiveTuple` extraction |
| `async_flow_summary` | Lifecycle events + summary stats |
| `async_flow_history` | Zeek-style `HistoryString` per flow |
| `async_flow_channel` | `cap.flow_stream` over an mpsc channel |
| `async_flow_filter` | Combining `with_dedup` + `flow_stream` |
| `async_flow_idle_per_key` | `with_idle_timeout_fn` for protocol-aware timeouts |
| `async_flow_conversations` | `Conversation<K>` aggregate |
| `async_flow_with_tap` | `with_pcap_tap` — record while you track |
## l7/ — session + datagram parsing (HTTP, DNS, custom)
The L7 surface. `multi_protocol_monitor` demuxes ICMP/TCP/UDP at
the flow level; `http_session` and `dns_lookups` add real protocol
parsing. For the "watch this interface for everything" recipe, use
the declarative `Monitor::builder()` API (see `monitor/` below).
| `async_on_tick` | Custom `DatagramParser::on_tick` emitting heartbeats | `tokio,flow,parse` |
| `multi_protocol_monitor` | One `flow_stream`, demux per-L4 (ICMP / TCP / UDP) | `tokio,flow,parse` |
| `http_session` | TCP/80,8080 → `HttpParser` → request/response events | `tokio,http` |
| `dns_lookups` | UDP/53 → `DnsUdpParser::with_correlation()` → query/response/RTT/unanswered | `tokio,dns` |
## pcap/ — offline replay
| `async_pcap_replay` | `AsyncPcapSource::open(...).flow_events(ext)` for flow-level replay |
| `async_pcap_sessions` | `AsyncPcapSource::open(...).sessions(ext, parser)` one-liner |
## monitor/ — declarative Monitor API (0.21+)
The `Monitor::builder()` API. Each example targets a single
aspect of the surface; compose them as needed for your own
code. All examples use plain `#[tokio::main]` (multi-thread
runtime) — `Monitor` is `Send` since 0.21.
**Start here for 0.25:** `monitor_subscriptions` — the typed subscription
engine (`packet()` / `flow::<P>()` / `session::<P>()` + `.expr("…")`), the
headline API with kernel filter pushdown.
| `monitor_subscriptions` | **the 0.25 subscription engine** — `packet()` / `flow::<P>()` / `session::<P>()` tiers + `.expr("…")` runtime filters + kernel pushdown |
| `monitor_basic` | `Monitor::builder()` + `.on::<FlowStarted<Tcp>>(...)` + StdoutSink |
| `monitor_detector_macro` | `detector!` macro for 3 stateless detectors (SshAttempt / HttpRequest / DnsQuery) |
| `monitor_layered_sinks` | `MinSeverity` + `DedupeAnomalies` + `RateLimitAnomalies` + `Tee::factory` over `StdoutSink` |
| `monitor_async_handler` | `on_async::<E>(...)` with `Arc<Pool>` capture for simulated I/O |
| `monitor_stream_consumer` | `with_broadcast::<Http>()` + `subscribe::<Http>()` → `EventStream` consumer |
| `monitor_pcap_replay` | `pcap_source(path) + pcap_speed_factor(f) + replay()` offline pipeline |
| `monitor_sharded_runner` | `ShardedRunner::new(iface, FanoutMode::Cpu, group, N, build)` per-CPU sharding |
| `monitor_eve_to_filebeat` | `EveSink` (feature `eve-sink`) writing Suricata-format EVE JSON for Filebeat ingest |
| `monitor_tracing_json` | structured JSON logging of anomalies + telemetry via `tracing-subscriber` (`TracingSink` + `on_capture_stats`) |
| `monitor_flow_export` | `export_flows` (NDJSON/IPFIX `FlowRecord`s) + `export_active_timeout` interim records for long flows (W1c) |
| `monitor_resilience` | `BackendErrorPolicy::Reopen` + `catch_handler_panics` — survive a flapping source + a panicking handler (W1e) |
| `monitor_eve_tls` | Suricata EVE `event_type:"tls"` records via `on_fingerprint` + `EveTlsSink` (W1d) |
| `monitor_metrics_export` | `MetricsSink` (feature `metrics`) Prometheus counter facade |
| `monitor_overload` | **Backpressure/overload detection** (#54): `OverloadDetector` driven from `on_capture_stats` turns `CaptureTelemetry::drop_rate` into a debounced `Normal`↔`Emergency` signal (Suricata-style hysteresis — no flapping); react to shedding however you like |
| `monitor_port_scan` | `pattern_detector!` over `PortScanDetector` (TRW scoring) |
| `monitor_beacon_detector` | `pattern_detector!` over `BeaconDetector` (period variance) |
| `monitor_rita_beacon` | `pattern_detector!` over `RitaBeaconDetector` — robust beacon scoring (Bowley skew + MADM, RITA v5); survives jitter/outliers the CV detector misses |
| `monitor_yara` | **YARA-X payload scanning** (feature `yara`, HEAVY — pulls wasmtime): `.yara(rules)` + `.on_yara_match(|key, m| …)` scanning each flow's accumulated payload at flow end, so signatures span segment boundaries |
| `monitor_dga_query` | `pattern_detector!` over `DgaScorer` (bigram entropy on DNS) |
| `monitor_file_hash_dfir` | `Sha256Sink + FileType` (feature `file-hash`) DFIR file hashing |
| `monitor_ech_adoption` | ECH downgrade detection via `EchOutcome` |
| `monitor_net_diagnostic` | **3 signals in one Monitor** (0.22 high-level API): unified ICMP errors via `on_icmp_error` (flow-joined), TCP resets via `on_tcp_reset`, per-app bandwidth via `on_bandwidth` + typed `BandwidthReport`. The 0.22 headline — 306 LoC of hand-rolled classifiers/HashMap/tick collapsed to ~70. |
| `monitor_multi_thread_default` | plain `#[tokio::main]` (multi-thread) — `Monitor` is `Send` (0.21) and **the run-loop future is `Send + 'static` (0.23)**, demonstrated via `tokio::spawn(monitor.run_for(..))` |
| `monitor_report_stream` | `report_to(period, build, sink)` + `JsonReportSink` (0.22 §3) shipping a typed `BandwidthSnapshot` report — the third output stream beside anomalies and broadcast |
| `monitor_label_table` | `LabelTable::new().set(...)` + `MonitorBuilder::label_table` — custom well-known port → app-label map feeding `on_bandwidth` |
| `monitor_arp_watch` | **L2 ARP** (feature `arp`): raw ARP feed via `on_arp` + spoof/binding-change anomalies via `on_arp_anomaly` (`SpoofSuspected`/`BindingChanged`), with `arp_allow`/`arp_warmup` |
| `monitor_ndp_watch` | **IPv6 Neighbor Discovery** (feature `ndp`): the ARP sibling for IPv6 — raw NDP feed via `on_ndp` + spoof/binding-change anomalies via `on_ndp_anomaly` (`SpoofSuspected` on unsolicited-override NA, `BindingChanged`), with `ndp_allow`/`ndp_warmup` |
| `monitor_lateral_movement` | **Active Directory / lateral movement** (feature `ad-protocols`): `.protocol::<Smb\|Kerberos\|Ldap\|Rdp>()` + `.on::<P>()` emitting anomalies for admin-share/DCSync (SMB), Kerberoasting (Kerberos), GetUserSPNs (LDAP), and RDP target usernames |
| `monitor_asset_discovery` | **Passive asset discovery** (feature `asset-protocols`): `.protocol::<Dhcp\|Ssdp\|Nbns>()` + `.on::<P>()` surfacing DHCP hostname/fingerprint, SSDP/UPnP firmware banners, and NetBIOS names — device facts leaked onto the LAN with no active probing |
| `monitor_ssh_hassh` | **SSH visibility + HASSH** (feature `ssh`): `.protocol::<Ssh>()` + `.on::<Ssh>()` surfacing the SSH version banner and the HASSH / HASSHServer handshake fingerprints (the SSH analogue of JA3/JA4) |
| `monitor_quic_sni` | **QUIC SNI visibility** (feature `quic`): `.protocol::<Quic>()` + `.on::<Quic>()` recovering the destination hostname + ALPN from the passive-readable QUIC Initial (HTTP/3), no decryption keys |
| `monitor_ioc` | **Threat-intel IOC matching** (features `dns,tls,http`): `.ioc(IocSet)` matching flow IPs, DNS queries, TLS SNI/JA4, and HTTP Host against a blocklist of bad IPs / subdomain-aware domains / JA4s, emitting `ioc_match` anomalies |
| `monitor_ioc_reload` | **Hot-reload the IOC blocklist** (#53): `Monitor::reload_handle()` + `ReloadHandle::set_ioc(..)` swap the live threat-intel set from a control task (feed refresh / SIGHUP) without dropping packets — lock-free `arc-swap` RCU |
| `monitor_p0f` | **Passive TCP/OS fingerprinting** (feature `p0f`): `.on_p0f()` emitting a p0f-3 signature per TCP SYN/SYN-ACK, identifying client + server OS from stack defaults without payload inspection |
| `monitor_flow_risk` | **nDPI-style flow risk** (features `tls,http`): `.flow_risk()` flagging `obsolete_tls` (negotiated SSLv3/TLS1.0/1.1) and `cleartext_http_credentials` (`Authorization: Basic` over plaintext) as `flow_risk` anomalies |
| `monitor_sigma_rules` | **Sigma rule evaluation** (features `sigma,dns,tls,http`): `.sigma(SigmaRuleSet)` evaluating vendor-neutral Sigma rules (bucketed by `logsource.category`) against DNS/HTTP/TLS records, emitting `sigma_match` anomalies |
| `monitor_sigma_reload` | **Hot-reload the Sigma rule set** (#53): `Monitor::reload_handle()` + `ReloadHandle::set_sigma(..)` swap the live rule set from a control task (SIEM content push / SIGHUP) without dropping packets — lock-free `arc-swap` RCU (same-category content refreshes hot-swap fully) |
| `monitor_ml_features` | **CICFlowMeter ML features** (feature `ml-features`): `.on_ml_features(|f| …)` emitting flowscope `CicFlowFeatures` (totals + IAT + active/idle) per flow at flow end, for offline ML pipelines |
| `monitor_nprint` | **nPrint per-flow matrix** (feature `nprint`): `.nprint(cfg)` + `.on_nprint(|key, m| …)` emitting flowscope's `NPrintMatrix` (one ternary header-bit row per packet) per flow at flow end, dumped as CSV for model-agnostic ML |
| `monitor_infra_protocols` | **Tier-2 UDP infra protocols** (feature `infra-protocols`): `.protocol::<Ntp/Snmp/Tftp/Radius>()` + `.on::<P>()` surfacing NTP/SNMP/TFTP/RADIUS passive metadata |
| `monitor_app_protocols` | **Tier-2 app/OT/VPN protocols** (features `ftp,smtp,modbus,dnp3,stun,wireguard`): `.protocol::<P>()` + `.on::<P>()` surfacing FTP/SMTP/Modbus/DNP3/STUN/WireGuard passive metadata |
| `monitor_ocsf` | **OCSF Detection Findings** (feature `ocsf-sink`): `.sink(OcsfSink::stdout())` writing each anomaly as an OCSF 2004 NDJSON finding (AWS Security Lake / Splunk OCSF), paired here with IOC matching |
| `monitor_l2_discovery` | **L2 neighbor discovery** (features `lldp,cdp`): `.on_lldp()` + `.on_cdp()` surfacing LLDP (IEEE 802.1AB) and CDP (Cisco) device announcements — chassis/device id, port, system name, platform — the infrastructure half of an asset inventory |
| `monitor_asset_inventory` | **Passive asset inventory** (features `asset,arp,ndp,lldp,cdp`): `.asset_inventory(cap)` + `.on_asset()` building a MAC-keyed device record fed by ARP/NDP/LLDP/CDP; fires on new-or-changed assets |
| `monitor_traffic_aggregation` | **Traffic aggregation** (#121, features `flow,dns,tls`): `.on_aggregate(period, …)` — rolling top talkers, host-pair traffic matrix, top DNS names / TLS SNI |
| `monitor_red_metrics` | **RED metrics** (#122, features `flow,dns`): `.on_red(period, …)` — request Rate / Error rate / Duration quantiles per protocol (DNS by rcode+timeouts, flows by end reason) |
| `monitor_owner_bandwidth` | **Owner-attributed bandwidth** (#130, feature `flow`): `.with_flow_attribution(hook)` + `.on_owner_bandwidth(period, …)` — throughput by tenant/container/subscriber with an unattributed bucket |
| `monitor_dns_monitor` | **DNS monitoring** (#120, features `flow,dns`): `.on_dns(…)` structured exchanges + `.name_map()`/`.on_name(…)` passive IP→name learning |
| `monitor_encrypted_dns` | **Encrypted-DNS visibility** (#133, features `tls,quic`): `.on_encrypted_dns(…)` flagging DoH/DoT/DoQ sessions and their resolver |
| `monitor_ip_defrag` | **IP-fragment reassembly** (#134, features `flow,dns`): `.reassemble_ip_fragments()` reassembles IPv4 fragments before parsing — defeats fragmentation-based evasion |
| `monitor_netns_capture` | **Network-namespace capture** (#126): `NetNs::from_name(..)` + `CaptureBuilder::netns(&ns)` to capture on a container's interface (needs `CAP_SYS_ADMIN`) |
| `monitor_triggered_pcap` | **Rotating + triggered pcap** (#125, feature `pcap`): `RotatingPcapWriter` + `TriggeredPcapWriter` pre-trigger ring — capture the lead-up to an event |
| `monitor_session_bandwidth` | **Per-session bandwidth** (features `tokio,flow`): rolling top-N active 5-tuples via `BandwidthByKey<FlowKey>` + cumulative bytes/avg-throughput from `FlowEnded.stats` |
| `monitor_process_bandwidth` | **Per-process bandwidth** (features `tokio,flow`): a `/proc/net` socket→pid resolver feeding `.with_flow_attribution(hook)` + `.on_owner_bandwidth(…)` — local-host, best-effort (nethogs-style) |
All take an `<iface>` argument (default `lo`) and an optional
`<seconds>` deadline. Pair with `synthetic_traffic` for
self-demoable runs without root.
```sh
cargo run --features "monitor-quickstart" --example monitor_basic -- lo 10
cargo run --features "monitor-quickstart" --example monitor_stream_consumer -- lo 10
cargo run --features "monitor-quickstart" --example monitor_sharded_runner -- lo 4 10
```
See [`docs/MIGRATING_0.21_TO_0.22.md`](../docs/MIGRATING_0.21_TO_0.22.md)
for the 0.21 → 0.22 transition guide (typed protocol roles, flat
`FlowPacket`, the operations toolkit `on_bandwidth` / `on_icmp_error` /
`on_tcp_reset`, the report stream, cross-shard merge, and the legacy
0.19 API removal).
## anomaly/ — multi-protocol anomaly correlators
Real-life detectors built on `netring::correlate`'s primitives
(`TimeBucketedCounter`, `KeyIndexed`) over raw `AsyncCapture` streams.
> **0.22:** the legacy `AnomalyMonitor` / `AnomalyRule` harness was
> removed. The detector examples that used it are gone; their
> equivalents live under `monitor/` on the declarative
> `Monitor::builder()` + `detector!` / `pattern_detector!` API
> (`monitor/port_scan`, `monitor/beacon_detector`, `monitor/dga_query`,
> `monitor/net_diagnostic`).
| `dns_query_burst` | Per-source-IP DNS rate anomaly via `TimeBucketedCounter`. Raw primitives, no harness. | `tokio,dns` |
| `dns_resolved_no_connection` | Cross-protocol: DNS resolved but no TCP/UDP follows within timeout. Uses `KeyIndexed::drain_expired` to surface unfulfilled resolutions. Raw primitives. | `tokio,dns` |
## util/ — demo helpers
| **`synthetic_traffic`** | **Companion traffic generator** — fires DNS queries / TCP SYN→RST flows / HTTP-shaped requests / per-source fan-out on `lo` so the L7 and anomaly examples are self-demoable. Userspace sockets only; **no `CAP_NET_RAW` / root required**. Run in one terminal, run the example you want to demo in another. | `tokio` |
---
## Running
The L7 + anomaly examples need real traffic to fire. The
`synthetic_traffic` helper provides enough on `lo` to demo most of
them without root:
```bash
# Terminal A — generator:
just synthetic-traffic 30
# Terminal B — pick a detector:
just example dns_query_burst tokio,dns -- lo 30 50
just example monitor_net_diagnostic monitor-quickstart,icmp -- lo 30
```
Other run patterns:
```bash
# Most examples take an `iface` arg, default `lo`:
cargo run --example multi_protocol_monitor --features tokio,flow,parse -- lo 30
# Full L4+L7 monitor on a real interface (declarative API):
cargo run --example monitor_net_diagnostic --features monitor-quickstart,icmp -- eth0 60
# Offline pcap replay (no privileges needed):
cargo run --example async_pcap_sessions --features tokio,flow,parse,pcap -- trace.pcap
```
```bash
The `justfile` also has shortcut targets (`just async`, `just dpi`,
`just bpf-filter`, etc.) — see `just --list`.