diff --git a/Dockerfile b/Dockerfile index 5ef2eae..41ba9cf 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,16 @@ ARG BASE_IMAGE=debian:trixie-slim + +FROM ${BASE_IMAGE} AS rust-builder +ARG DEBIAN_FRONTEND=noninteractive +RUN apt-get update \ + && apt-get install -y --no-install-recommends cargo rustc \ + && rm -rf /var/lib/apt/lists/* +WORKDIR /build/tzsp-receiver +COPY rust/tzsp-receiver/Cargo.toml ./Cargo.toml +COPY rust/tzsp-receiver/src ./src +RUN cargo test --release \ + && cargo build --release + FROM ${BASE_IMAGE} ARG DEBIAN_FRONTEND=noninteractive @@ -42,12 +54,13 @@ RUN printf '%s\n' \ WORKDIR /opt/ids +COPY --from=rust-builder /build/tzsp-receiver/target/release/mikrosuricata-tzsp /usr/local/bin/mikrosuricata-tzsp COPY app /opt/ids/app COPY scripts /opt/ids/scripts COPY suricata /opt/ids/suricata RUN chmod +x /opt/ids/scripts/*.sh \ - && mkdir -p /data /var/log/suricata /var/lib/suricata/rules /run/suricata /tmp/suricata-build-test \ + && mkdir -p /data /var/log/suricata /var/lib/suricata/rules /run/suricata /run/mikrosuricata /tmp/suricata-build-test \ && suricata -T \ -c /etc/suricata/suricata.yaml \ --include /opt/ids/suricata/ids-output.yaml \ @@ -63,6 +76,12 @@ RUN chmod +x /opt/ids/scripts/*.sh \ ENV PYTHONUNBUFFERED=1 \ TZSP_BIND=0.0.0.0 \ TZSP_PORT=37008 \ + TZSP_RECEIVER_BIN=/usr/local/bin/mikrosuricata-tzsp \ + TZSP_TELEMETRY_SOCKET=/run/mikrosuricata/tzsp-telemetry.sock \ + TZSP_RCVBUF_BYTES=33554432 \ + TZSP_BATCH_SIZE=256 \ + TZSP_QUEUE_MB=64 \ + TZSP_DATAGRAM_BYTES=12288 \ TAP_NAME=suritap0 \ TAP_MTU=9000 \ WEB_BIND=0.0.0.0 \ @@ -70,7 +89,7 @@ ENV PYTHONUNBUFFERED=1 \ DB_PATH=/data/ids.db \ EVE_PATH=/data/logs/suricata/eve.json \ SURICATA_LOG_MAX_MB=512 \ - FORENSIC_PCAP_MODE=blocks \ + FORENSIC_PCAP_MODE=alerts \ FORENSIC_PCAP_WINDOW_SECONDS=60 \ FORENSIC_PCAP_MEMORY_MB=64 \ FORENSIC_PCAP_MAX_FILES=32 \ diff --git a/README.md b/README.md index bc1e7e7..8da16a1 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,37 @@ # MikroSuricata -Project version: `0.9.1` +Project version: `0.11.2` + + +## What changed in 0.11.2 + +- Replaced full-traffic RouterOS Packet Sniffer capture with a **hybrid TZSP path**: routed IPv4 uses `/ip firewall mangle action=sniff-tzsp`, while Packet Sniffer streams only `filter-mac-protocol=!ip` traffic such as ARP, IPv6, LLDP and PPPoE. +- The capture deploy no longer assumes a production bridge, VLAN or interface name. `TZSP_L2_INTERFACE=` defaults to RouterOS `filter-interface=all`; set it only when a site intentionally wants to constrain the low-volume non-IPv4 stream. +- Packet Sniffer TZSP destination now tries the current `streaming-server=IP:PORT` syntax first and automatically falls back to legacy split `streaming-server`/`streaming-port` syntax. +- The project-owned IPv4 mangle rule is identified by the exact comment `MikroSuricata TZSP IPv4`; deployment replaces only that rule and leaves unrelated mangle rules untouched. RouterOS `sniff-tzsp` continues processing subsequent mangle rules after cloning the packet. +- Added `scripts/configure-routeros-tzsp-hybrid.sh` to migrate an existing installation without rebuilding the container, including `DRY_RUN=true` preview mode. +- Replaced the VLAN-specific RouterOS template with `routeros/02-tzsp-hybrid.rsc` and added detailed capture coverage, limitations and migration documentation in `docs/routeros-tzsp-capture.md`. + +## What changed in 0.11.0 + +- Reworked the Rust TZSP data-plane for RouterOS containers that cannot raise kernel `rmem_max`: UDP receive and TZSP decode/TAP injection now run on **separate threads**. +- The RX thread does only blocking `recvmmsg(MSG_WAITFORONE)` into reusable preallocated batches. It never waits for TAP, Suricata, Python, Redis or the UI. +- Added a bounded **userspace packet queue** (`TZSP_QUEUE_MB`, default 64 MiB). With the default 256-packet batch and 12 KiB slots this provides roughly half a second of burst absorption at 1 Gbit/s even when the kernel socket is limited to ~416 KiB. +- If downstream processing falls behind after the userspace queue fills, the RX thread keeps draining UDP into a scratch batch and records explicit `queue_dropped_datagrams` instead of silently turning all loss into kernel UDP drops. +- Increased the default receive batch from 64 to **256** and made the per-datagram buffer configurable (`TZSP_DATAGRAM_BYTES`, default 12288; up to 65535). Oversized/truncated datagrams are counted explicitly. +- Added live capture diagnostics: TZSP ingress rate, inspected rate, capture efficiency, userspace queue fill, kernel drops/s, queue drops/s, truncated packets and RX/TAP worker health. +- The overview now labels the main value as **Inspected throughput** and separately shows TZSP ingress, so a RouterOS mirror limitation can be distinguished from a Suricata/TAP bottleneck immediately. +- Prometheus exports the new Rust queue/capture-health gauges and drop counters. + +## What changed in 0.10.0 + +- Moved the complete TZSP packet hot-path out of Python into a dedicated **Rust data-plane process**. Rust now owns batched UDP receive (`recvmmsg`), TZSP decoding, direction accounting and TAP injection; packet bytes no longer cross into Python. +- Added configurable UDP receive buffering and batched receive. In 0.11.0 the hot path was further split into dedicated RX/TAP threads with a userspace queue. +- Added 1 Hz Unix-socket telemetry from Rust to the Python control plane. The dashboard now receives live throughput independently from retained Redis history, so **Throughput now** is current instead of reusing a stale historical payload. +- Added receiver observability to the UI/health/Prometheus path: actual receive-buffer size, Rust PID, telemetry age and kernel UDP drop counter. +- Fixed the overview refresh path so throughput history is refreshed during periodic UI refreshes, not only at initial load/range changes. +- The old `FORENSIC_PCAP_MODE=blocks` pre-event ring required Python to process every frame. With the isolated Rust data-plane it is treated as a legacy setting and safely falls back to Suricata alert PCAP capture. The new default is `alerts`. +- Docker now compiles and tests the Rust receiver in a dedicated build stage and copies only the stripped receiver binary into the runtime image. The existing RouterOS amd64/arm64/armv7 image workflow is preserved. ## What changed in 0.9.1 @@ -70,7 +101,7 @@ A lightweight IDS stack designed to run as a **single container on MikroTik Rout - Security analytics add anomalies, NXDOMAIN counts, encrypted/cleartext session counts, local/remote endpoint inventory, passive IP/MAC asset observations, top signatures/severities and JA4/JA3/HASSH fingerprint inventory. - The built-in local ruleset adds NXDOMAIN/DNS-rate signals, outbound SMB/SMTP/FTP policy detections, exposed database-service probes and an internal administrative/lateral-movement burst detector. -RouterOS mirrors selected traffic with TZSP, the container decodes the frames into a TAP interface, Suricata analyzes them, and the Python service stores EVE alerts in SQLite and exposes a small web dashboard. +RouterOS mirrors traffic with a hybrid TZSP path: routed IPv4 is cloned by an IPv4 mangle `sniff-tzsp` rule, while Packet Sniffer handles non-IPv4 Ethernet traffic. A dedicated Rust process decodes and injects the frames into TAP, Suricata analyzes them, and the Python control plane handles EVE/SQLite/NDR/API/UI without sitting in the packet hot-path. ## NDR / MikroTik-specific options @@ -93,28 +124,39 @@ All NDR state, IOC data, Redis persistence, Suricata logs/rules and forensic PCA ## Architecture ```text -VLAN / RouterOS traffic - | - v -RouterOS Packet Sniffer - | - | TZSP UDP/37008 - v +RouterOS routed IPv4 ----> mangle sniff-tzsp ---------+ + | TZSP UDP/37008 +RouterOS non-IPv4 -----> Packet Sniffer (!ip) --------+ + v single RouterOS container Debian slim - + Python TZSP receiver + + Rust TZSP data-plane (recvmmsg -> TZSP -> TAP) + TAP suritap0 + Suricata IDS - + EVE JSON watcher + + Python control plane / EVE JSON watcher + SQLite alerts / assets / NDR incidents / sessions + MikroSuricata behavior + correlation engine + local IOC datasets (IP/domain/SHA256/JA3/JA4/HASSH) - + Redis/RAM bounded traffic history - + WebSocket live stream + + Redis traffic history + + Rust -> Python 1 Hz Unix telemetry + + WebSocket live throughput / event stream + Web UI :8080 + optional RouterOS REST blocking ``` +The capture path is intentionally independent from the UI/control plane: + +```text +RouterOS TZSP UDP + | + v +Rust receiver -- recvmmsg() --> TZSP decode --> TAP write --> Suricata + | + +-- 1 Hz counters only --> Unix datagram --> Python --> Redis / WebSocket / Prometheus +``` + +If Redis, the browser or an analytics request is slow, it cannot block UDP receive/TAP injection. + The RouterOS deployment workflow is: ```text @@ -342,7 +384,7 @@ Every feed update is transactional at the merged-rules level: the existing `suri ```dotenv UPDATE_RULES_ON_START=false RULE_UPDATE_INTERVAL_HOURS=24 -FORENSIC_PCAP_MODE=blocks +FORENSIC_PCAP_MODE=alerts FORENSIC_PCAP_WINDOW_SECONDS=60 FORENSIC_PCAP_MEMORY_MB=64 FORENSIC_PCAP_MAX_FILES=32 @@ -584,15 +626,15 @@ Then deploy by giving the **RouterOS-side TAR path** directly: ./scripts/deploy-routeros.sh routeros-suricata-tzsp-arm64.tar ``` -The deployer no longer builds, detects image architecture, renames, or re-uploads the image. For project version `0.9.0` it creates: +The deployer no longer builds, detects image architecture, renames, or re-uploads the image. For project version `0.11.2` it creates: ```text -name=suricata_0.9.0 +name=suricata_0.11.2 file=routeros-suricata-tzsp-arm64.tar -root-dir=/containers/suricata_0.9.0/root +root-dir=/containers/suricata_0.11.2/root ``` -The remaining deployment work is unchanged: bridge/VETH/NAT, environment, persistent mounts, optional RouterOS REST user/firewall integration, TZSP sniffer configuration, image extraction wait, container start, and final status. Existing containers are not removed. Re-running deployment for the same version stops with `Container suricata_0.9.0 already exists`. +The remaining deployment work is unchanged: private container bridge/VETH/NAT, environment, persistent mounts, optional RouterOS REST user/firewall integration, hybrid TZSP capture configuration, image extraction wait, container start, and final status. Existing containers are not removed. Re-running deployment for the same version stops with `Container suricata_0.11.2 already exists`. For SSH key authentication set: @@ -644,7 +686,7 @@ RouterOS templates are located in `routeros/`: ```text 01-container-network.rsc -02-sniffer-vlan100.rsc +02-tzsp-hybrid.rsc 03-rest-and-firewall.rsc 04-container-import-amd64.rsc 04-container-import-arm64.rsc @@ -686,26 +728,50 @@ Replacing or restarting the application container does not remove any of these f ## TZSP capture -The default deployment configuration can configure RouterOS Packet Sniffer to stream VLAN traffic to the container address on UDP port `37008`. +The default RouterOS deployment uses a hybrid capture model: -Relevant settings in `deploy-routeros.env`: - -```dotenv -CONFIGURE_SNIFFER=true -START_SNIFFER=true -VLAN_ID=100 +```text +routed IPv4 -> /ip firewall mangle action=sniff-tzsp +non-IPv4 L2 -> /tool sniffer filter-mac-protocol=!ip + \_______________________________/ + TZSP UDP/37008 ``` -If the router already uses Packet Sniffer for another purpose, disable automatic sniffer configuration: +Relevant `deploy-routeros.env` settings: ```dotenv -CONFIGURE_SNIFFER=false -START_SNIFFER=false +TZSP_PORT=37008 +CONFIGURE_TZSP_CAPTURE=true +CONFIGURE_IPV4_MANGLE=true +CONFIGURE_L2_SNIFFER=true +START_L2_SNIFFER=true +TZSP_L2_INTERFACE= +TZSP_L2_MAC_PROTOCOL=!ip ``` -Then configure the capture manually. +No production bridge/VLAN name is assumed. Empty `TZSP_L2_INTERFACE` maps to RouterOS `filter-interface=all`. To intentionally constrain non-IPv4 capture, set an exact interface name, for example `TZSP_L2_INTERFACE=bridge-core`. -Hardware-offloaded bridge traffic may require additional verification on the specific RouterOS device because some switched traffic can bypass software capture paths. +The IPv4 rule is added to `chain=forward` and is owned by the comment `MikroSuricata TZSP IPv4`. Only that project-owned rule is replaced; unrelated mangle rules are left intact. `sniff-tzsp` clones the packet and then continues to subsequent mangle rules. + +The Packet Sniffer complement uses `filter-mac-protocol=!ip`. RouterOS defines `ip` as IPv4 EtherType `0x0800` and `ipv6` separately as `0x86DD`, so this stream includes ARP, IPv6, LLDP, PPPoE and other non-IPv4 Ethernet protocols. `filter-stream=yes` prevents the sniffer's own TZSP stream from being recaptured. + +Current RouterOS builds may expose the target as `streaming-server=IP:PORT`; older builds/documentation may expose `streaming-server` plus `streaming-port`. Deployment tries the combined form first and falls back automatically. + +To migrate an already running installation without changing the container image: + +```bash +./scripts/configure-routeros-tzsp-hybrid.sh +``` + +Preview without modifying RouterOS: + +```bash +DRY_RUN=true ./scripts/configure-routeros-tzsp-hybrid.sh +``` + +Packet Sniffer is global RouterOS state. If another administrative workflow owns `/tool/sniffer`, use `CONFIGURE_L2_SNIFFER=false` and manage the low-volume L2 complement separately. Hardware-offloaded bridge-only traffic may also be invisible to Packet Sniffer on some devices. + +Full coverage details, caveats, migration behavior and verification commands are in [`docs/routeros-tzsp-capture.md`](docs/routeros-tzsp-capture.md). --- @@ -721,15 +787,29 @@ The pipeline is: ```text TZSP datagram - -> Python decoder + -> Rust batched UDP receiver + -> Rust TZSP decoder -> Ethernet frame -> TAP suritap0 -> Suricata -> eve.json - -> Python EVE watcher + -> Python control plane / EVE watcher -> SQLite / Web UI ``` +The Rust receiver emits only a compact 1 Hz telemetry sample to Python. That sample drives the live throughput card and is persisted asynchronously for historical throughput charts. + +Useful receive-path tuning: + +```dotenv +TZSP_RCVBUF_BYTES=33554432 +TZSP_BATCH_SIZE=256 +TZSP_QUEUE_MB=64 +TZSP_DATAGRAM_BYTES=12288 +``` + +On RouterOS the kernel may cap the actual socket buffer around a few hundred KiB regardless of `TZSP_RCVBUF_BYTES`. The 0.11.0 receiver therefore does not rely on sysctl: a dedicated RX thread drains UDP into the preallocated userspace queue. The UI exposes both kernel drops and userspace queue drops, plus TZSP ingress versus traffic actually injected into Suricata. + If the RouterOS container cannot create the TAP interface, the application will fail early with an error related to `/dev/net/tun`, `TUNSETIFF`, or permissions. This is the main platform-specific capability to validate on the target router. --- @@ -778,11 +858,14 @@ Full RouterOS/container application entry point app/dev_web.py Local web-only entry point -app/tzsp.py -TZSP receiver and decoder +rust/tzsp-receiver/ +Production Rust TZSP receiver: batched UDP receive, decoder, counters and TAP injection -app/tap.py -TAP interface handling +app/tzsp_rust.py +Rust process supervision plus 1 Hz telemetry bridge; no packet bytes enter Python + +app/tzsp.py / app/tap.py +Legacy decoder/TAP helpers retained for focused tests; not used by the production packet path app/eve.py Suricata EVE JSON watcher @@ -862,7 +945,7 @@ The default configuration is observation-oriented: AUTO_BLOCK=false ALERT_MAX_SEVERITY=2 UPDATE_RULES_ON_START=false -FORENSIC_PCAP_MODE=blocks +FORENSIC_PCAP_MODE=alerts ROUTEROS_PASSWORD=CHANGE_ME ADMIN_USERNAME=admin ADMIN_PASSWORD= @@ -875,7 +958,7 @@ Keep automatic firewall actions disabled until the capture path and alert qualit ## Image-only upgrade on an already configured RouterOS -After the first deployment, when `veth-ids`, bridge/NAT, TZSP sniffer, `IDS_ENV` and `IDS_MOUNTS` already exist, do not run the full deploy just to change the image. +After the first deployment, when the VETH/private bridge/NAT, hybrid TZSP capture, `IDS_ENV` and `IDS_MOUNTS` already exist, do not run the full deploy just to change the image. 1. Upload the ready TAR only: @@ -889,18 +972,18 @@ After the first deployment, when `veth-ids`, bridge/NAT, TZSP sniffer, `IDS_ENV` ./scripts/upgrade-routeros-container.sh routeros-suricata-tzsp-arm64.tar ``` -For version `0.9.0` the second command creates: +For version `0.11.2` the second command creates: ```text -name=suricata_0.9.0 +name=suricata_0.11.2 file=routeros-suricata-tzsp-arm64.tar -root-dir=/containers/suricata_0.9.0/root +root-dir=/containers/suricata_0.11.2/root interface=veth-ids envlist=IDS_ENV mountlists=IDS_MOUNTS ``` -The upgrade helper does not modify the bridge, IP addresses, NAT, veth, TZSP/sniffer, firewall, REST user or envlist definitions. It stops older `suricata_*` containers, normalizes `IDS_MOUNTS` to the single `/containers/suricata-data -> /data` mapping, creates the new versioned container, waits for image extraction and starts it. Older containers are kept stopped for rollback. +The upgrade helper does not modify the bridge, IP addresses, NAT, VETH, hybrid TZSP capture, firewall, REST user or envlist definitions. It stops older `suricata_*` containers, normalizes `IDS_MOUNTS` to the single `/containers/suricata-data -> /data` mapping, creates the new versioned container, waits for image extraction and starts it. Older containers are kept stopped for rollback. All new mutable state is written below `/data`, so subsequent image upgrades need only that one persistent mount. diff --git a/VERSION b/VERSION index e3e1807..bc859cb 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.9.8 +0.11.2 diff --git a/app/config.py b/app/config.py index 582e372..c572f9f 100644 --- a/app/config.py +++ b/app/config.py @@ -37,6 +37,12 @@ class Config: tzsp_port: int tap_name: str tap_mtu: int + tzsp_receiver_bin: str + tzsp_telemetry_socket: str + tzsp_rcvbuf_bytes: int + tzsp_batch_size: int + tzsp_queue_mb: int + tzsp_datagram_bytes: int suricata_config: str suricata_output_config: str suricata_home_net: str @@ -111,6 +117,12 @@ class Config: tzsp_port=_int("TZSP_PORT", 37008), tap_name=os.getenv("TAP_NAME", "suritap0"), tap_mtu=_int("TAP_MTU", 9000), + tzsp_receiver_bin=os.getenv("TZSP_RECEIVER_BIN", "/usr/local/bin/mikrosuricata-tzsp"), + tzsp_telemetry_socket=os.getenv("TZSP_TELEMETRY_SOCKET", "/run/mikrosuricata/tzsp-telemetry.sock"), + tzsp_rcvbuf_bytes=max(1024 * 1024, _int("TZSP_RCVBUF_BYTES", 32 * 1024 * 1024)), + tzsp_batch_size=max(1, min(1024, _int("TZSP_BATCH_SIZE", 256))), + tzsp_queue_mb=max(8, min(512, _int("TZSP_QUEUE_MB", 64))), + tzsp_datagram_bytes=max(2048, min(65535, _int("TZSP_DATAGRAM_BYTES", 12288))), suricata_config=os.getenv("SURICATA_CONFIG", "/etc/suricata/suricata.yaml"), suricata_output_config=os.getenv( "SURICATA_OUTPUT_CONFIG", "/opt/ids/suricata/ids-output.yaml" @@ -150,7 +162,7 @@ class Config: db_path=os.getenv("DB_PATH", "/data/ids.db"), eve_path=os.getenv("EVE_PATH", "/data/logs/suricata/eve.json"), suricata_log_max_mb=_int("SURICATA_LOG_MAX_MB", 512), - forensic_pcap_mode=_choice("FORENSIC_PCAP_MODE", "blocks", {"blocks", "alerts", "all", "off"}), + forensic_pcap_mode=_choice("FORENSIC_PCAP_MODE", "alerts", {"blocks", "alerts", "all", "off"}), forensic_pcap_window_seconds=max(5, _int("FORENSIC_PCAP_WINDOW_SECONDS", 60)), forensic_pcap_memory_mb=max(1, _int("FORENSIC_PCAP_MEMORY_MB", 64)), forensic_pcap_max_files=max(1, _int("FORENSIC_PCAP_MAX_FILES", 32)), @@ -213,6 +225,11 @@ class Config: "tzsp_port": self.tzsp_port, "tap_name": self.tap_name, "tap_mtu": self.tap_mtu, + "tzsp_receiver_engine": "rust", + "tzsp_rcvbuf_bytes": self.tzsp_rcvbuf_bytes, + "tzsp_batch_size": self.tzsp_batch_size, + "tzsp_queue_mb": self.tzsp_queue_mb, + "tzsp_datagram_bytes": self.tzsp_datagram_bytes, "suricata_home_net": self.suricata_home_net, "suricata_extra_rules_glob": self.suricata_extra_rules_glob, "web_port": self.web_port, diff --git a/app/main.py b/app/main.py index 5037af1..d06c1bc 100644 --- a/app/main.py +++ b/app/main.py @@ -17,7 +17,6 @@ from .analytics_cache import AnalyticsSnapshotCache from .backup import BackupManager from .config import Config from .eve import EVEWatcher -from .flow_tracker import FlowTracker from .forensics import ForensicPcapRing from .live import EventBus, LiveEventPipeline, TrafficHistory, TrafficNormalizer from .maintenance import clear_suricata_logs, storage_info @@ -30,9 +29,8 @@ from .routeros import RouterOSClient from .rules import RuleManager from .state import RuntimeStats from .store import AlertStore -from .tap import TapDevice from .tuning import AlertTuner -from .tzsp import TZSPReceiver +from .tzsp_rust import RustTZSPReceiver from .webui import WebServer @@ -50,6 +48,18 @@ def _ensure_suricata_state(cfg: Config) -> None: def _prepare_suricata_output_config(cfg: Config) -> Config: + # The legacy "blocks" mode depended on Python seeing every TZSP frame to + # maintain a pre-event RAM ring. The Rust data-plane intentionally removes + # Python from that packet path. Keep forensic evidence without reintroducing + # the bottleneck by falling back to Suricata's alert-associated PCAP output. + if cfg.forensic_pcap_mode == "blocks": + print( + "[forensics] FORENSIC_PCAP_MODE=blocks is not used with the Rust data-plane; " + "using Suricata alert PCAP capture instead", + flush=True, + ) + cfg = replace(cfg, forensic_pcap_mode="alerts") + source = Path(cfg.suricata_output_config) text = source.read_text(encoding="utf-8") match = re.search(r"(?ms)^ - pcap-log:\n.*?(?=^ - |\Z)", text) @@ -120,16 +130,24 @@ def main() -> int: if purged: print(f"[db] purged {purged} old alerts", flush=True) - tap = TapDevice(cfg.tap_name, cfg.tap_mtu) + receiver = RustTZSPReceiver( + binary=cfg.tzsp_receiver_bin, + telemetry_socket=cfg.tzsp_telemetry_socket, + stats=stats, + stop_event=stop_event, + ) try: - tap.open() + receiver.start() + if not receiver.wait_ready(timeout=8.0): + raise RuntimeError("Rust receiver did not report ready state") except Exception as exc: - print(f"[fatal] cannot create TAP {cfg.tap_name}: {exc}", file=sys.stderr, flush=True) + print(f"[fatal] cannot start Rust TZSP data-plane: {exc}", file=sys.stderr, flush=True) print("[fatal] container needs /dev/net/tun and NET_ADMIN capability", file=sys.stderr, flush=True) + receiver.close() store.close() return 2 - print(f"[tap] {cfg.tap_name} is up, mtu={cfg.tap_mtu}", flush=True) + print(f"[tap] {cfg.tap_name} is owned by Rust TZSP receiver, mtu={cfg.tap_mtu}", flush=True) log_dir = os.path.dirname(cfg.eve_path) or "/var/log/suricata" suricata_cmd = [ @@ -156,7 +174,7 @@ def main() -> int: file=sys.stderr, flush=True, ) - tap.close() + receiver.close() store.close() return test.returncode or 3 @@ -235,15 +253,9 @@ def main() -> int: ) live_pipeline = LiveEventPipeline(event_bus, traffic_history) normalizer = TrafficNormalizer(cfg.monitored_networks) - flow_tracker = FlowTracker(normalizer, live_pipeline, update_interval_seconds=cfg.live_flow_update_seconds) - - def observe_frame(frame: bytes) -> None: - forensic_pcap.observe(frame) - flow_tracker.observe(frame) - - receiver = TZSPReceiver( - cfg.tzsp_bind, cfg.tzsp_port, tap.write, stats, stop_event, frame_observer=observe_frame - ) + # The Rust data-plane emits one compact rate sample per second. Persisting it + # is asynchronous and never sits in the packet receive/injection path. + receiver.set_throughput_sink(live_pipeline.publish_throughput) watcher = EVEWatcher( cfg.eve_path, store, @@ -268,22 +280,35 @@ def main() -> int: def health() -> dict: suricata_up = suricata.poll() is None - tzsp_up = receiver.is_alive() and receiver.sock is not None - tap_up = tap.fd is not None and os.path.exists(f"/sys/class/net/{cfg.tap_name}") + tzsp_up = receiver.is_alive() + tap_up = os.path.exists(f"/sys/class/net/{cfg.tap_name}") eve_up = watcher.is_alive() routeros_status = "configured" if routeros.configured else "disabled" db = store.database_info() storage = storage_info(cfg.db_path, cfg.eve_path) rules = rule_manager.status() runtime = stats.snapshot() + receiver_status = receiver.status() redis_status = redis_supervisor.status() suri_stats = runtime.get("suricata") or {} kernel_packets = int(suri_stats.get("capture.kernel_packets", 0) or 0) kernel_drops = int(suri_stats.get("capture.kernel_drops", 0) or 0) alert_overflow = int(suri_stats.get("detect.alert_queue_overflow", 0) or 0) inject_errors = int(runtime.get("inject_errors", 0) or 0) + tzsp_udp_drops = int(receiver_status.get("kernel_udp_drops", 0) or 0) + tzsp_udp_drops_now = int(receiver_status.get("kernel_udp_drops_interval", 0) or 0) + tzsp_queue_drops = int(receiver_status.get("queue_dropped_datagrams", 0) or 0) + tzsp_queue_drops_now = int(receiver_status.get("queue_drops_interval", 0) or 0) + tzsp_truncated_now = int(receiver_status.get("truncated_interval", 0) or 0) drop_pct = round((kernel_drops / kernel_packets) * 100.0, 3) if kernel_packets else 0.0 - sensor_degraded = (kernel_packets >= 1000 and drop_pct >= 1.0) or alert_overflow > 0 or inject_errors > 0 + sensor_degraded = ( + (kernel_packets >= 1000 and drop_pct >= 1.0) + or alert_overflow > 0 + or inject_errors > 0 + or tzsp_udp_drops_now > 0 + or tzsp_queue_drops_now > 0 + or tzsp_truncated_now > 0 + ) core_up = suricata_up and tzsp_up and tap_up and eve_up and db["ok"] routeros_required_ok = (not cfg.auto_block) or routeros.configured operational = core_up and routeros_required_ok @@ -313,7 +338,13 @@ def main() -> int: "tzsp": { "name": "TZSP receiver", "status": "up" if tzsp_up else "down", - "details": f"Listening on UDP {cfg.tzsp_bind}:{cfg.tzsp_port}", + "details": ( + f"Rust PID {receiver.pid or '—'} · UDP {cfg.tzsp_bind}:{cfg.tzsp_port} · " + f"socket={int(receiver_status.get('rcvbuf_bytes', 0) or 0)} B · " + f"queue={int(receiver_status.get('queue_depth_batches', 0) or 0)}/" + f"{int(receiver_status.get('queue_capacity_batches', 0) or 0)} batches · " + f"kernel drops={tzsp_udp_drops} · queue drops={tzsp_queue_drops}" + ), }, "tap": { "name": "TAP interface", @@ -328,7 +359,7 @@ def main() -> int: "sensor_quality": { "name": "Sensor quality / packet loss", "status": "degraded" if sensor_degraded else "up", - "details": f"capture packets={kernel_packets}; kernel drops={kernel_drops} ({drop_pct}%); alert queue overflow={alert_overflow}; inject errors={inject_errors}", + "details": f"Suricata packets={kernel_packets}; Suricata kernel drops={kernel_drops} ({drop_pct}%); TZSP UDP drops={tzsp_udp_drops}; alert queue overflow={alert_overflow}; TAP inject errors={inject_errors}", }, "eve": { "name": "EVE JSON watcher", @@ -346,9 +377,9 @@ def main() -> int: "details": f"{storage['path']}; {storage['used_percent']}% used", }, "live_flows": { - "name": "Immediate TZSP sessions", - "status": "up" if tzsp_up else "down", - "details": f"{flow_tracker.status()['active_flows']} active; non-persistent {flow_tracker.status()['update_interval_seconds']:g}s updates", + "name": "Live session stream", + "status": "up" if eve_up else "down", + "details": "Suricata EVE sessions; packet capture is isolated in the Rust data-plane", }, "traffic_history": { "name": "Live traffic history", @@ -433,8 +464,8 @@ def main() -> int: return { "components": { "web": True, - "tzsp": receiver.is_alive() and receiver.sock is not None, - "tap": tap.fd is not None, + "tzsp": receiver.is_alive(), + "tap": os.path.exists(f"/sys/class/net/{cfg.tap_name}"), "suricata": suricata.poll() is None, "eve": watcher.is_alive(), }, @@ -454,7 +485,7 @@ def main() -> int: mode="full", started_at=started_at, state_provider=metrics_state, - flow_tracker=flow_tracker, + flow_tracker=receiver, event_bus=event_bus, live_pipeline=live_pipeline, ndr_analyzer=ndr_analyzer, @@ -477,6 +508,7 @@ def main() -> int: ndr_analyzer=ndr_analyzer, backup_manager=backup_manager, forensic_pcap=forensic_pcap, + traffic_source=receiver, metrics_provider=prometheus_metrics.render, ) @@ -535,7 +567,6 @@ def main() -> int: analytics_cache.start() notifier.start() ndr_analyzer.start() - receiver.start() watcher.start() housekeeping_thread.start() web.start() @@ -548,6 +579,10 @@ def main() -> int: print(f"[fatal] Suricata exited with rc={suricata_rc}", file=sys.stderr, flush=True) rc = suricata_rc or 4 break + if not receiver.is_alive(): + print("[fatal] Rust TZSP data-plane exited", file=sys.stderr, flush=True) + rc = 5 + break time.sleep(0.5) finally: stop_event.set() @@ -567,7 +602,6 @@ def main() -> int: os.remove("/run/suricata.pid") except FileNotFoundError: pass - tap.close() live_pipeline.stop() analytics_cache.stop() ndr_analyzer.stop() diff --git a/app/metrics.py b/app/metrics.py index e78b121..381f6db 100644 --- a/app/metrics.py +++ b/app/metrics.py @@ -238,22 +238,71 @@ class PrometheusMetrics: if self.flow_tracker is None: return status = self.flow_tracker.status() - for key in ("active_flows", "max_flows", "update_interval_seconds"): - if key in status and _number(status[key]) is not None: - self._emit( - lines, - f"mikrosuricata_flow_tracker_{_metric_name(key)}", - status[key], - metric_type="gauge", - ) - for key in ("published_updates", "evicted_flows", "parse_errors", "throughput_samples"): - if key in status and _number(status[key]) is not None: - self._emit( - lines, - f"mikrosuricata_flow_tracker_{_metric_name(key)}_total", - status[key], - metric_type="counter", - ) + is_rust = str(status.get("engine") or "").lower() == "rust" + if is_rust: + self._emit( + lines, + "mikrosuricata_tzsp_receiver_info", + 1, + metric_type="gauge", + help_text="TZSP packet data-plane implementation.", + labels={"engine": "rust"}, + ) + for key in ( + "rcvbuf_bytes", + "batch_size", + "datagram_bytes", + "queue_capacity_batches", + "queue_capacity_bytes", + "queue_depth_batches", + "queue_high_water_batches", + "capture_efficiency_pct", + "rx_thread_alive", + "worker_thread_alive", + "telemetry_age_ms", + "process_alive", + "ready", + ): + if key in status and _number(status[key]) is not None: + self._emit( + lines, + f"mikrosuricata_tzsp_receiver_{_metric_name(key)}", + status[key], + metric_type="gauge", + ) + for key in ( + "kernel_udp_drops", + "queue_dropped_datagrams", + "truncated_datagrams", + "tzsp_datagrams", + "tzsp_rx_bytes", + "rx_batches", + "telemetry_errors", + ): + if key in status and _number(status[key]) is not None: + self._emit( + lines, + f"mikrosuricata_tzsp_receiver_{_metric_name(key)}_total", + status[key], + metric_type="counter", + ) + else: + for key in ("active_flows", "max_flows", "update_interval_seconds"): + if key in status and _number(status[key]) is not None: + self._emit( + lines, + f"mikrosuricata_flow_tracker_{_metric_name(key)}", + status[key], + metric_type="gauge", + ) + for key in ("published_updates", "evicted_flows", "parse_errors", "throughput_samples"): + if key in status and _number(status[key]) is not None: + self._emit( + lines, + f"mikrosuricata_flow_tracker_{_metric_name(key)}_total", + status[key], + metric_type="counter", + ) traffic = status.get("traffic_counters") or {} if not isinstance(traffic, Mapping): diff --git a/app/state.py b/app/state.py index 34d435a..b1de7c2 100644 --- a/app/state.py +++ b/app/state.py @@ -20,6 +20,9 @@ class RuntimeStats: "tzsp_unsupported": 0, "frames_injected": 0, "inject_errors": 0, + "tzsp_kernel_udp_drops": 0, + "tzsp_queue_drops": 0, + "tzsp_truncated_datagrams": 0, "eve_events": 0, "eve_alerts": 0, "eve_parse_errors": 0, @@ -44,6 +47,17 @@ class RuntimeStats: with self._lock: self._data[key] = datetime.now(timezone.utc).isoformat() + def update_tzsp_receiver(self, values: dict[str, Any]) -> None: + """Replace Rust receiver counters from its compact telemetry sample.""" + with self._lock: + for key, value in values.items(): + if key == "last_packet_at": + if value: + self._data[key] = value + continue + if isinstance(value, (int, float)) and not isinstance(value, bool): + self._data[key] = value + def update_suricata(self, stats: dict[str, Any], timestamp: str | None = None) -> None: flattened: dict[str, int | float] = {} _flatten_numeric("", stats, flattened, 240) diff --git a/app/static/css/app.css b/app/static/css/app.css index f58f800..21309e9 100644 --- a/app/static/css/app.css +++ b/app/static/css/app.css @@ -68,3 +68,5 @@ html,body{min-height:100%;background:var(--bg);color:var(--text)}body{overflow-x /* Compact secondary navigation inside data-heavy primary views. */ .subtabs{display:flex;align-items:center;gap:5px;margin:0 0 12px;padding:5px;border:1px solid var(--line-soft);border-radius:9px;background:#0e0f11;overflow-x:auto;scrollbar-width:thin}.subtab-button{flex:0 0 auto;border:1px solid transparent;border-radius:6px;background:transparent;color:#777880;padding:7px 11px;font-size:11px;font-weight:600;cursor:pointer;white-space:nowrap;transition:.14s ease}.subtab-button:hover{color:#d6d6db;background:#141517}.subtab-button.active{color:#f4f4f5;background:#191a1d;border-color:#2b2c30;box-shadow:0 1px 2px rgba(0,0,0,.18)}.subtab-button.active::before{content:'';display:inline-block;width:6px;height:6px;margin-right:7px;border-radius:999px;background:var(--green);vertical-align:1px}.subtab-panel{display:none}.subtab-panel.active{display:block}.panel-stack{display:grid;gap:12px;align-content:start;min-width:0} @media(max-width:780px){.subtabs{margin-bottom:10px}.subtab-button{padding:7px 10px}.panel-stack{width:100%}} + +.metric-sub.metric-sub-bad{color:#d99a9a} diff --git a/app/static/js/app.js b/app/static/js/app.js index e70521e..bcdc8a4 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -14,7 +14,7 @@ liveEnabled: false, paused: false, live: [], liveById: new Map(), liveSequence: 0, liveRenderTimer: null, liveFilterTimer: null, historyLoaded: false, snapshot: [], batchTimes: [], uiDropped: 0, serverDropped: 0, - incidents: [], analytics: null, analyticsWindow: 0, throughput: null, throughputWindow: 0, status: null, config: null, ruleSources: [], ruleSourcesLoaded: false, + incidents: [], analytics: null, analyticsWindow: 0, throughput: null, throughputWindow: 0, currentThroughput: null, currentThroughputWindow: 0, status: null, config: null, ruleSources: [], ruleSourcesLoaded: false, selectedRuleSources: new Set(), sourceQueue: null, sourceQueueTimer: null, ndrIncidents: [], assets: [], iocs: [], pcaps: [], pcapMode: 'blocks', ndrSummary: {}, ruleIntelligence: [], ruleSnapshots: [], mergedRulesOffset: 0, mergedRulesQuery: '', backups: [], audit: [], @@ -374,16 +374,40 @@ state.analyticsPollTimer=setTimeout(()=>{if(Number(windowSec)===selectedWindow())loadAnalytics(windowSec,true);},delay); } + function renderCurrentThroughput(t) { + const windowSec=Number(t?.window_seconds||selectedWindow()); + if(windowSec && windowSec!==selectedWindow())return; + state.currentThroughput=t||{}; state.currentThroughputWindow=selectedWindow(); + const total=Math.max(0,Number(t?.current_bps||0)), inbound=Math.max(0,Number(t?.current_in_bps||0)), outbound=Math.max(0,Number(t?.current_out_bps||0)); + const ingress=Math.max(0,Number(t?.current_ingress_bps||0)); + const other=Math.max(0,Number(t?.current_other_bps ?? (total-inbound-outbound))), pps=Math.max(0,Number(t?.current_pps||0)); + if($('metricThroughput'))$('metricThroughput').textContent=fmtBits(total); + if($('metricThroughputSplit'))$('metricThroughputSplit').textContent=`IN ${fmtBits(inbound)} · OUT ${fmtBits(outbound)}${other>0?` · OTHER ${fmtBits(other)}`:''} · ${Math.round(pps).toLocaleString()} pps`; + const quality=$('metricThroughputQuality'); + if(quality){ + const age=t?.current_sample_age_ms; + const ageText=Number.isFinite(Number(age))?`${(Number(age)/1000).toFixed(1)}s`:'waiting'; + const capture=Math.max(0,Math.min(100,Number(t?.capture_efficiency_pct ?? 0))); + const queue=Math.max(0,Math.min(100,Number(t?.queue_fill_pct||0))); + const loss=Math.max(0,Number(t?.loss_pps||0)); + const inspectRatio=Math.max(0,Math.min(100,Number(t?.inspection_ratio_pct ?? 100))); + const kernelDrops=Math.max(0,Number(t?.kernel_udp_drops||0)); + const queueDrops=Math.max(0,Number(t?.queue_dropped_datagrams||0)); + const truncated=Math.max(0,Number(t?.truncated_datagrams||0)); + const buffer=Number(t?.rcvbuf_bytes||0), queueBytes=Number(t?.queue_capacity_bytes||0); + quality.textContent=`TZSP RX ${fmtBits(ingress)} · capture ${capture.toFixed(1)}% · queue ${queue.toFixed(0)}% · loss ${loss.toFixed(loss<10?1:0)}/s`; + quality.title=`sample ${ageText} old · inspected/TZSP ${inspectRatio.toFixed(1)}% · socket ${fmtBytes(buffer)} · userspace queue ${fmtBytes(queueBytes)} · kernel drops ${kernelDrops.toLocaleString()} · queue drops ${queueDrops.toLocaleString()} · truncated ${truncated.toLocaleString()}`; + const pipelineBehind=ingress>10_000_000 && inspectRatio<90; + const unhealthy=loss>0||queue>=80||pipelineBehind||t?.current_sample_fresh===false||t?.rx_thread_alive===false||t?.worker_thread_alive===false; + quality.classList.toggle('metric-sub-bad',unhealthy); + } + } + function renderThroughput(t) { const windowSec=Number(t?.window_seconds||selectedWindow()); if(windowSec!==selectedWindow())return; state.throughput=t; state.throughputWindow=windowSec; - if($('metricThroughput'))$('metricThroughput').textContent=fmtBits(t.current_bps||0); - if($('metricThroughputSplit')){ - const total=Math.max(0,Number(t.current_bps||0)), inbound=Math.max(0,Number(t.current_in_bps||0)), outbound=Math.max(0,Number(t.current_out_bps||0)); - const other=Math.max(0,Number(t.current_other_bps ?? (total-inbound-outbound))); - $('metricThroughputSplit').textContent=`IN ${fmtBits(inbound)} · OUT ${fmtBits(outbound)}${other>0?` · OTHER ${fmtBits(other)}`:''}`; - } + renderCurrentThroughput(t); if($('metricPeakThroughput'))$('metricPeakThroughput').textContent=fmtBits(t.peak_bps||0); if($('metricBytes'))$('metricBytes').textContent=fmtBytes(t.bytes||0); const charts=window.MikroSuricataCharts; if(charts?.drawThroughput)charts.drawThroughput($('throughputChart'),t.timeline||[]); @@ -420,12 +444,8 @@ if(!state.throughput || state.throughputWindow!==windowSec){state.throughput=a;state.throughputWindow=windowSec;} $('metricEvents').textContent = Number(a.events||0).toLocaleString(); const traffic=(state.throughput && state.throughputWindow===windowSec)?state.throughput:a; - if($('metricThroughput'))$('metricThroughput').textContent=fmtBits(traffic.current_bps||0); - if($('metricThroughputSplit')){ - const total=Math.max(0,Number(traffic.current_bps||0)), inbound=Math.max(0,Number(traffic.current_in_bps||0)), outbound=Math.max(0,Number(traffic.current_out_bps||0)); - const other=Math.max(0,Number(traffic.current_other_bps ?? (total-inbound-outbound))); - $('metricThroughputSplit').textContent=`IN ${fmtBits(inbound)} · OUT ${fmtBits(outbound)}${other>0?` · OTHER ${fmtBits(other)}`:''}`; - } + const live=(state.currentThroughput && state.currentThroughputWindow===windowSec)?state.currentThroughput:a; + renderCurrentThroughput(live); if($('metricPeakThroughput'))$('metricPeakThroughput').textContent=fmtBits(traffic.peak_bps||0); $('metricBytes').textContent = fmtBytes(traffic.bytes||0); $('metricAlerts').textContent = Number(a.alerts||0).toLocaleString(); $('metricBlocked').textContent = Number(a.blocked||0).toLocaleString(); $('metricEventRate').textContent = `${Math.round(Number(a.events||0)/(Number(a.window_seconds||3600)/60)).toLocaleString()} / min`; @@ -542,7 +562,7 @@ $('ndrIncidentRows').innerHTML=state.ndrIncidents.length?state.ndrIncidents.map(x=>`${Number(x.risk_score||0)}${fmtTime(x.last_seen)}${esc(x.subject_ip||'—')}${esc((x.stages||[]).join(' → ')||'detection')}${renderAttack(x.mitre)}${esc(x.summary||x.title||'—')}${Number(x.event_count||0).toLocaleString()}${x.blocked?' · blocked':''}${esc(x.status||'open')} · `).join(''):'No correlated NDR incidents yet.'; $('assetRows').innerHTML=state.assets.length?state.assets.map(x=>`${Number(x.risk_score||0)}${esc(x.ip)}${esc(x.hostname||'—')}
${esc(x.mac||x.identity_source||'—')}
${esc((x.protocols||[]).slice(0,8).join(', ')||'—')}${esc((x.ports||[]).slice(0,12).join(', ')||'—')}${Number(x.alert_count||0).toLocaleString()}${fmtTime(x.last_seen)}`).join(''):'Assets appear after traffic or RouterOS inventory sync.'; $('iocRows').innerHTML=state.iocs.length?state.iocs.map(x=>`${esc(x.indicator_type)}${esc(x.indicator)}${Number(x.confidence||0)}%S${esc(x.severity||'—')}${esc(x.source||'—')}${Number(x.hit_count||0).toLocaleString()}${fmtTime(x.last_hit_at)}`).join(''):'No local IOCs configured.'; - const pcapDescriptions={blocks:'Mode: blocks · PCAP is persisted only after a successful RouterOS block; recent packets come from the bounded RAM ring.',alerts:'Mode: alerts · Suricata persists packets associated with alerts.',all:'Mode: all · Suricata persists all observed packets into the rotating PCAP log.',off:'Mode: off · forensic PCAP persistence is disabled.'}; + const pcapDescriptions={blocks:'Mode: blocks (legacy) · with the Rust data-plane this is converted to Suricata alert capture so Python never processes every packet.',alerts:'Mode: alerts · Suricata persists packets associated with alerts.',all:'Mode: all · Suricata persists all observed packets into the rotating PCAP log.',off:'Mode: off · forensic PCAP persistence is disabled.'}; if($('pcapMeta'))$('pcapMeta').textContent=pcapDescriptions[state.pcapMode]||`Mode: ${state.pcapMode}`; $('pcapRows').innerHTML=state.pcaps.length?state.pcaps.map(x=>{const url=`/api/forensics/pcap?name=${encodeURIComponent(x.name)}`;return `${esc(x.name)}${fmtBytes(x.size_bytes)}${fmtTime(Number(x.modified_at||0)*1000)}download`;}).join(''):'No forensic PCAP files yet.'; } @@ -708,6 +728,7 @@ if(msg.data?.status)renderStatus(msg.data.status); if(msg.data?.analytics)renderAnalytics(msg.data.analytics); } else if(msg.type==='status')renderStatus(msg.data||{}); + else if(msg.type==='throughput')renderCurrentThroughput(msg.data||{}); else if(msg.type==='analytics')renderAnalytics(msg.data||{}); }; ws.onclose=()=>{ if (!state.authEnabled || state.authenticated) scheduleReconnect(); }; @@ -790,6 +811,7 @@ api('/api/stats').then(renderStats), api('/api/alerts?limit=250').then(alerts=>{state.incidents=alerts.alerts||[];renderIncidents();}), loadOverviewSnapshot(windowSec,true), + loadThroughput(windowSec,true), loadAnalytics(windowSec,true), state.view==='intelligence'?loadIntelligence(true):Promise.resolve(), ]); diff --git a/app/templates/index.html b/app/templates/index.html index 44eda22..dae04d0 100644 --- a/app/templates/index.html +++ b/app/templates/index.html @@ -51,14 +51,14 @@
Events
0
0 / min
-
Throughput now
0 bps
IN 0 bps · OUT 0 bps
-
Observed traffic
0 B
TZSP bytes in selected range
+
Inspected throughput
0 bps
IN 0 bps · OUT 0 bps · 0 pps
Rust TZSP · waiting for sample
+
Observed traffic
0 B
Ethernet frame bytes delivered to Suricata in selected range
Peak throughput
0 bps
Selected time range
Threats
0
0 incidents
Blocked
0
Policy actions
-

Traffic throughput

Total, inbound and outbound network speed sampled from TZSP traffic and retained in Redis.

loading
TotalInboundOutbound
+

Traffic throughput

Total, inbound and outbound speed from the Rust TZSP data-plane; 1 s samples are retained in Redis.

loading
TotalInboundOutbound

Traffic direction

Inbound / outbound / internal

Events & alerts

Complete retained event history for the selected time range.

EventsAlerts

Event mix

Flow, DNS, TLS, HTTP and alerts

@@ -152,7 +152,7 @@

Threat intelligence repository

IOC hits increase incident risk and remain persistent in SQLite.

TypeIndicatorConfidenceSeveritySourceHitsLast hit
-

Forensic PCAP ring

Persistent evidence mode is loading…

FileSizeModified
+

Forensic PCAP

Persistent evidence mode is loading…

FileSizeModified
diff --git a/app/tzsp_rust.py b/app/tzsp_rust.py new file mode 100644 index 0000000..f19a7b2 --- /dev/null +++ b/app/tzsp_rust.py @@ -0,0 +1,287 @@ +from __future__ import annotations + +import json +import os +import socket +import subprocess +import threading +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable + +from .state import RuntimeStats + + +class RustTZSPReceiver: + """Supervise the Rust TZSP data-plane and ingest its 1 Hz telemetry. + + Packet bytes never cross into Python. The Rust process owns UDP reception, + TZSP decoding and TAP injection. Python receives only compact telemetry over + a Unix datagram socket, so UI/Redis work cannot back-pressure packet capture. + """ + + def __init__( + self, + *, + binary: str, + telemetry_socket: str, + stats: RuntimeStats, + stop_event: threading.Event, + throughput_sink: Callable[[dict[str, Any]], None] | None = None, + ) -> None: + self.binary = str(binary) + self.telemetry_socket = str(telemetry_socket) + self.stats = stats + self.stop_event = stop_event + self._throughput_sink = throughput_sink + self._process: subprocess.Popen | None = None + self._socket: socket.socket | None = None + self._thread = threading.Thread(target=self._run_telemetry, name="tzsp-rust-telemetry", daemon=True) + self._lock = threading.RLock() + self._ready = threading.Event() + self._last: dict[str, Any] = {} + self._samples = 0 + self._telemetry_errors = 0 + self._started_at = time.monotonic() + + def start(self) -> None: + if self._process is not None: + return + binary = Path(self.binary) + if not binary.is_file(): + raise RuntimeError(f"Rust TZSP receiver binary not found: {self.binary}") + + path = Path(self.telemetry_socket) + path.parent.mkdir(parents=True, exist_ok=True) + try: + path.unlink() + except FileNotFoundError: + pass + + sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + sock.bind(self.telemetry_socket) + sock.settimeout(0.5) + self._socket = sock + self._thread.start() + + env = os.environ.copy() + env["TZSP_TELEMETRY_SOCKET"] = self.telemetry_socket + self._process = subprocess.Popen([self.binary], env=env) + print(f"[tzsp] Rust data-plane started, pid={self._process.pid}", flush=True) + + def wait_ready(self, timeout: float = 8.0) -> bool: + deadline = time.monotonic() + max(0.1, float(timeout)) + while time.monotonic() < deadline: + process = self._process + if process is not None and process.poll() is not None: + return False + if self._ready.wait(timeout=min(0.1, max(0.0, deadline - time.monotonic()))): + return True + return False + + def set_throughput_sink(self, sink: Callable[[dict[str, Any]], None] | None) -> None: + with self._lock: + self._throughput_sink = sink + + def is_alive(self) -> bool: + process = self._process + return bool(process is not None and process.poll() is None and self._thread.is_alive()) + + @property + def pid(self) -> int | None: + process = self._process + return process.pid if process is not None and process.poll() is None else None + + def close(self, timeout: float = 3.0) -> None: + process = self._process + if process is not None and process.poll() is None: + process.terminate() + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + process.kill() + try: + process.wait(timeout=1) + except subprocess.TimeoutExpired: + pass + sock = self._socket + self._socket = None + if sock is not None: + try: + sock.close() + except OSError: + pass + if self._thread.is_alive(): + self._thread.join(timeout=1.0) + try: + Path(self.telemetry_socket).unlink() + except FileNotFoundError: + pass + self._process = None + + def status(self) -> dict[str, Any]: + with self._lock: + data = dict(self._last) + now_ms = int(time.time() * 1000) + ts_ms = int(data.get("ts_ms") or 0) + data.update( + { + "engine": "rust", + "process_alive": self.is_alive(), + "pid": self.pid or data.get("pid"), + "ready": bool(self._ready.is_set() and self.is_alive()), + "telemetry_age_ms": max(0, now_ms - ts_ms) if ts_ms else None, + "throughput_samples": self._samples, + "telemetry_errors": self._telemetry_errors, + } + ) + return data + + def current_throughput(self, window_seconds: int | None = None) -> dict[str, Any]: + with self._lock: + sample = dict(self._last) + now_ms = int(time.time() * 1000) + ts_ms = int(sample.get("ts_ms") or 0) + interval = max(float(sample.get("interval_ms") or 1000) / 1000.0, 0.001) + age_ms = max(0, now_ms - ts_ms) if ts_ms else 10**9 + fresh = bool(ts_ms and age_ms <= max(3000, int(interval * 3000))) + if fresh: + total = round(max(int(sample.get("bytes_total") or 0), 0) * 8 / interval) + inbound = round(max(int(sample.get("bytes_in") or 0), 0) * 8 / interval) + outbound = round(max(int(sample.get("bytes_out") or 0), 0) * 8 / interval) + pps = round(max(int(sample.get("packets_total") or 0), 0) / interval, 2) + ingress_bps = round(max(int(sample.get("rx_bytes_interval") or 0), 0) * 8 / interval) + ingress_pps = round(max(int(sample.get("rx_datagrams_interval") or 0), 0) / interval, 2) + else: + total = inbound = outbound = ingress_bps = 0 + pps = ingress_pps = 0.0 + queue_depth = max(int(sample.get("queue_depth_batches") or 0), 0) + queue_capacity = max(int(sample.get("queue_capacity_batches") or 0), 0) + queue_fill_pct = round((queue_depth / queue_capacity) * 100.0, 1) if queue_capacity else 0.0 + inspection_ratio_pct = ( + round(min(100.0, (total / ingress_bps) * 100.0), 1) + if ingress_bps > 0 + else (100.0 if total == 0 else 0.0) + ) + loss_per_second = round( + ( + max(int(sample.get("kernel_udp_drops_interval") or 0), 0) + + max(int(sample.get("queue_drops_interval") or 0), 0) + + max(int(sample.get("truncated_interval") or 0), 0) + ) + / interval, + 2, + ) if fresh else 0.0 + return { + "window_seconds": int(window_seconds or 0), + "current_bps": total, + "current_in_bps": inbound, + "current_out_bps": outbound, + "current_other_bps": max(0, total - inbound - outbound), + "current_pps": pps, + "current_ingress_bps": ingress_bps, + "current_ingress_pps": ingress_pps, + "inspection_ratio_pct": inspection_ratio_pct, + "capture_efficiency_pct": float(sample.get("capture_efficiency_pct") or 0.0), + "loss_pps": loss_per_second, + "current_sample_ts_ms": ts_ms, + "current_sample_age_ms": age_ms if ts_ms else None, + "current_sample_fresh": fresh, + "receiver_engine": "rust", + "receiver_pid": self.pid, + "kernel_udp_drops": int(sample.get("kernel_udp_drops") or 0), + "kernel_udp_drops_interval": int(sample.get("kernel_udp_drops_interval") or 0), + "queue_dropped_datagrams": int(sample.get("queue_dropped_datagrams") or 0), + "queue_drops_interval": int(sample.get("queue_drops_interval") or 0), + "truncated_datagrams": int(sample.get("truncated_datagrams") or 0), + "truncated_interval": int(sample.get("truncated_interval") or 0), + "queue_depth_batches": queue_depth, + "queue_capacity_batches": queue_capacity, + "queue_capacity_bytes": int(sample.get("queue_capacity_bytes") or 0), + "queue_high_water_batches": int(sample.get("queue_high_water_batches") or 0), + "queue_fill_pct": queue_fill_pct, + "rx_thread_alive": bool(sample.get("rx_thread_alive", False)), + "worker_thread_alive": bool(sample.get("worker_thread_alive", False)), + "rcvbuf_bytes": int(sample.get("rcvbuf_bytes") or 0), + "batch_size": int(sample.get("batch_size") or 0), + "datagram_bytes": int(sample.get("datagram_bytes") or 0), + } + + def overlay_current(self, payload: dict[str, Any], window_seconds: int | None = None) -> dict[str, Any]: + result = dict(payload) + result.update(self.current_throughput(window_seconds or int(result.get("window_seconds") or 0))) + return result + + def _run_telemetry(self) -> None: + while not self.stop_event.is_set(): + sock = self._socket + if sock is None: + break + try: + raw = sock.recv(64 * 1024) + except socket.timeout: + continue + except OSError: + if self.stop_event.is_set() or self._socket is None: + break + self._telemetry_errors += 1 + continue + try: + message = json.loads(raw.decode("utf-8")) + if not isinstance(message, dict) or message.get("type") != "tzsp_sample": + continue + self._ingest(message) + except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError): + self._telemetry_errors += 1 + + def _ingest(self, message: dict[str, Any]) -> None: + with self._lock: + self._last = dict(message) + sink = self._throughput_sink + if message.get("ready"): + self._ready.set() + + last_packet_ms = int(message.get("last_packet_ms") or 0) + last_packet_at = None + if last_packet_ms: + last_packet_at = datetime.fromtimestamp(last_packet_ms / 1000.0, tz=timezone.utc).isoformat() + self.stats.update_tzsp_receiver( + { + "tzsp_datagrams": int(message.get("tzsp_datagrams") or 0), + "tzsp_decode_errors": int(message.get("tzsp_decode_errors") or 0), + "tzsp_unsupported": int(message.get("tzsp_unsupported") or 0), + "frames_injected": int(message.get("frames_injected") or 0), + "inject_errors": int(message.get("inject_errors") or 0), + "tzsp_kernel_udp_drops": int(message.get("kernel_udp_drops") or 0), + "tzsp_queue_drops": int(message.get("queue_dropped_datagrams") or 0), + "tzsp_truncated_datagrams": int(message.get("truncated_datagrams") or 0), + "last_packet_at": last_packet_at, + } + ) + + sample = { + key: int(message.get(key) or 0) + for key in ( + "ts_ms", + "interval_ms", + "bytes_total", + "bytes_in", + "bytes_out", + "bytes_internal", + "bytes_external", + "packets_total", + "packets_in", + "packets_out", + "packets_internal", + "packets_external", + ) + } + self._samples += 1 + if sink is not None and sample["interval_ms"] > 0: + try: + sink(sample) + except Exception: + # Telemetry persistence is best-effort and is deliberately never + # allowed to affect the independent Rust packet data-plane. + self._telemetry_errors += 1 diff --git a/app/webui.py b/app/webui.py index 191d1ea..e136e12 100644 --- a/app/webui.py +++ b/app/webui.py @@ -20,7 +20,7 @@ import urllib.parse from collections import defaultdict, deque from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from typing import Callable +from typing import Any, Callable from .config import Config from .auth import SessionAuth @@ -122,6 +122,7 @@ class WebServer: ndr_analyzer: NDRAnalyzer | None = None, backup_manager: BackupManager | None = None, forensic_pcap: ForensicPcapRing | None = None, + traffic_source: Any | None = None, metrics_provider: Callable[[], str] | None = None, ) -> None: self.config = config @@ -137,6 +138,7 @@ class WebServer: self.threat_intel = threat_intel self.ndr_analyzer = ndr_analyzer self.forensic_pcap = forensic_pcap + self.traffic_source = traffic_source self.metrics_provider = metrics_provider self.metrics_access = MetricsAccessControl(config) if metrics_provider is not None else None self.backup_manager = backup_manager or BackupManager(config.db_path, os.path.dirname(config.db_path) or ".") @@ -162,10 +164,30 @@ class WebServer: def _analytics_payload(self, window_seconds: int) -> dict: if self.analytics_cache is not None: - return self.analytics_cache.get(window_seconds) - if self.traffic_history is not None: - return self.traffic_history.analytics(window_seconds) - return {"window_seconds": window_seconds, "events": 0, "timeline": []} + payload = self.analytics_cache.get(window_seconds) + elif self.traffic_history is not None: + payload = self.traffic_history.analytics(window_seconds) + else: + payload = {"window_seconds": window_seconds, "events": 0, "timeline": []} + return self._overlay_current_throughput(payload, window_seconds) + + def _overlay_current_throughput(self, payload: dict, window_seconds: int) -> dict: + source = self.traffic_source + if source is None or not hasattr(source, "overlay_current"): + return payload + try: + return source.overlay_current(payload, window_seconds) + except Exception: + return payload + + def _current_throughput_payload(self, window_seconds: int) -> dict: + source = self.traffic_source + if source is None or not hasattr(source, "current_throughput"): + return {"window_seconds": window_seconds, "current_bps": 0} + try: + return source.current_throughput(window_seconds) + except Exception: + return {"window_seconds": window_seconds, "current_bps": 0} def _login_allowed(self, client_ip: str) -> bool: now = time.monotonic() @@ -297,7 +319,8 @@ class WebServer: query = urllib.parse.parse_qs(parsed.query) window = self._query_int(query, "window", 3600, 60, config.traffic_retention_hours * 3600) try: - self._json(traffic_history.throughput_analytics(window)) + payload = traffic_history.throughput_analytics(window) + self._json(outer._overlay_current_throughput(payload, window)) except RedisUnavailableError as exc: self._json({"error": f"Redis throughput history unavailable: {exc}"}, status=503) return @@ -778,6 +801,7 @@ class WebServer: self._ws_send_json(bootstrap) last_status = time.monotonic() last_analytics = last_status + last_throughput = 0.0 while True: if not self._ws_client_control(): return @@ -819,6 +843,9 @@ class WebServer: time.sleep(0.25) now = time.monotonic() + if now - last_throughput >= 1: + self._ws_send_json({"type": "throughput", "data": outer._current_throughput_payload(window)}) + last_throughput = now if now - last_status >= 5: self._ws_send_json({"type": "status", "data": outer._status_payload()}) last_status = now diff --git a/deploy-routeros.env.example b/deploy-routeros.env.example index 2343dbf..92b975e 100644 --- a/deploy-routeros.env.example +++ b/deploy-routeros.env.example @@ -17,17 +17,33 @@ CONTAINER_SUBNET=172.31.255.0/30 CONTAINER_BRIDGE=br-ids CONTAINER_VETH=veth-ids -# Packet Sniffer -> TZSP -VLAN_ID=100 +# RouterOS -> TZSP hybrid capture. +# Routed IPv4 uses /ip firewall mangle action=sniff-tzsp for high throughput. +# Packet Sniffer is kept only for non-IPv4 Ethernet frames (ARP, IPv6, LLDP, +# PPPoE discovery, etc.) using filter-mac-protocol=!ip. TZSP_PORT=37008 -CONFIGURE_SNIFFER=true -START_SNIFFER=true +CONFIGURE_TZSP_CAPTURE=true +CONFIGURE_IPV4_MANGLE=true +CONFIGURE_L2_SNIFFER=true +START_L2_SNIFFER=true +# Empty means all RouterOS interfaces. Set a real interface/bridge name only +# when you intentionally want to constrain the low-volume non-IPv4 stream. +TZSP_L2_INTERFACE= +TZSP_L2_MAC_PROTOCOL=!ip + +# Rust receive-path tuning. RouterOS may cap the kernel socket buffer; the +# userspace batch queue is the primary burst buffer for high-throughput mirrors. +TZSP_RCVBUF_BYTES=33554432 +TZSP_BATCH_SIZE=256 +TZSP_QUEUE_MB=64 +TZSP_DATAGRAM_BYTES=12288 # Suricata/app SURICATA_HOME_NET=[192.168.0.0/16,10.0.0.0/8,172.16.0.0/12] SURICATA_LOG_MAX_MB=512 -# Forensic PCAP: blocks (default), alerts, all, off -FORENSIC_PCAP_MODE=blocks +# Forensic PCAP: alerts (default), all, off. Legacy "blocks" is accepted but +# falls back to alert capture so Python never returns to the per-packet path. +FORENSIC_PCAP_MODE=alerts FORENSIC_PCAP_WINDOW_SECONDS=60 FORENSIC_PCAP_MEMORY_MB=64 FORENSIC_PCAP_MAX_FILES=32 diff --git a/docs/routeros-tzsp-capture.md b/docs/routeros-tzsp-capture.md new file mode 100644 index 0000000..e69116d --- /dev/null +++ b/docs/routeros-tzsp-capture.md @@ -0,0 +1,276 @@ +# RouterOS TZSP capture architecture + +MikroSuricata 0.11.2 uses a **hybrid RouterOS capture path**. The goal is to keep high-volume routed IPv4 off `/tool/sniffer` while still preserving visibility into ARP and other non-IPv4 Ethernet traffic. + +## Default topology + +```text + routed IPv4 +RouterOS forwarding --------------------> /ip firewall mangle + action=sniff-tzsp + | + | TZSP UDP/37008 + v + MikroSuricata Rust + | + v + TAP -> Suricata + + non-IPv4 Ethernet +RouterOS interfaces --------------------> /tool/sniffer + filter-mac-protocol=!ip + filter-stream=yes + | + +---- TZSP UDP/37008 ----+ +``` + +Both paths use the same TZSP receiver in the container. + +## Why hybrid capture + +A full `/tool/sniffer` TZSP stream can become the limiting stage on some RouterOS systems long before the Rust receiver, TAP or Suricata are saturated. In testing on the system that motivated this change, full Packet Sniffer streaming plateaued around 150-270 Mbit/s while an IPv4 `mangle action=sniff-tzsp` rule delivered roughly 1.2 Gbit/s of encapsulated traffic to the container VETH. + +This is not treated as a universal RouterOS speed guarantee. The deployer simply chooses the path that proved suitable for high-volume routed IPv4 and keeps Packet Sniffer for the much smaller non-IPv4 stream. + +## Routed IPv4 + +The deployer manages one rule: + +```routeros +/ip/firewall/mangle +add chain=forward \ + action=sniff-tzsp \ + sniff-target=172.31.255.2 \ + sniff-target-port=37008 \ + comment="MikroSuricata TZSP IPv4" +``` + +The target IP and port come from `CONTAINER_IP` and `TZSP_PORT`; the values above are examples. + +`sniff-tzsp` clones the matching packet to the TZSP receiver and then continues processing the next mangle rule. MikroSuricata places its rule at the beginning of the IPv4 `forward` mangle chain so later existing mangle rules still run normally. + +The deployer removes/replaces **only** the rule whose comment is exactly: + +```text +MikroSuricata TZSP IPv4 +``` + +It does not delete, disable or reorder unrelated mangle rules. + +### What this IPv4 rule covers + +`chain=forward` covers IPv4 routed **through** RouterOS. This is the main IDS path for client-to-Internet and inter-subnet traffic. + +It intentionally does not capture: + +- IPv4 traffic terminating on RouterOS itself (`input`), +- IPv4 traffic generated by RouterOS itself (`output`), +- pure L2 IPv4 switching that never enters the routed IPv4 `forward` chain. + +The first two exclusions also avoid any risk of recursively observing RouterOS-generated TZSP traffic. If router-local IPv4 inspection is required later, it should be added as a separate, explicitly designed path rather than by broadening the default rule blindly. + +## ARP, IPv6 and other non-IPv4 Ethernet traffic + +Packet Sniffer is configured as the complement: + +```routeros +/tool/sniffer/set \ + streaming-enabled=yes \ + filter-stream=yes \ + filter-interface=all \ + filter-mac-protocol=!ip \ + filter-direction=any +``` + +RouterOS names EtherType `0x0800` as `ip` and EtherType `0x86DD` as `ipv6`. Therefore `filter-mac-protocol=!ip` means **everything except IPv4 EtherType**, not "everything except all IP". + +With the default filter the stream can include, among other protocols: + +- ARP (`0x0806`), +- IPv6 (`0x86DD`), +- LLDP, +- 802.1X/EAPoL, +- LACP, +- PPPoE session/discovery, +- MPLS EtherTypes, +- RARP, +- VLAN/service-VLAN frames when visible to the sniffer. + +`filter-stream=yes` is kept enabled so packets destined for the configured sniffer server are ignored by Packet Sniffer rather than captured again. + +### IPv6 performance note + +IPv6 is deliberately left in the Packet Sniffer complement in 0.11.2 because the requested design is "mangle for IPv4, stream everything except IPv4". If the network later carries sustained high-volume IPv6, the Packet Sniffer performance ceiling can reappear for that traffic. At that point IPv6 should be split into its own high-throughput capture path and the L2 sniffer narrowed to explicit non-IP EtherTypes. + +## No bridge or interface name is assumed + +The deployer does **not** assume names such as `bridge`, `bridge-trunk`, `br0`, `LAN`, or any site-specific VLAN interface. + +Default: + +```dotenv +TZSP_L2_INTERFACE= +``` + +An empty value is translated to RouterOS: + +```text +filter-interface=all +``` + +If an operator intentionally wants to constrain non-IPv4 capture, set the exact RouterOS interface name: + +```dotenv +TZSP_L2_INTERFACE=bridge-core +``` + +The deployer verifies that a non-empty interface name exists before applying the sniffer configuration. + +`CONTAINER_BRIDGE=br-ids` is different: that is the **new private bridge created for the IDS container**, not the bridge from which production traffic is assumed to arrive. It remains configurable through `deploy-routeros.env`. + +## RouterOS sniffer endpoint syntax + +Newer RouterOS CLI builds can expose the TZSP destination as a composite value: + +```routeros +streaming-server=172.31.255.2:37008 +``` + +Some older documentation/builds expose the destination as separate properties: + +```routeros +streaming-server=172.31.255.2 +streaming-port=37008 +``` + +The deploy and migration helper try the composite `IP:PORT` form first and automatically fall back to the split form if RouterOS rejects it. + +## Deployment variables + +Default capture settings: + +```dotenv +TZSP_PORT=37008 +CONFIGURE_TZSP_CAPTURE=true +CONFIGURE_IPV4_MANGLE=true +CONFIGURE_L2_SNIFFER=true +START_L2_SNIFFER=true +TZSP_L2_INTERFACE= +TZSP_L2_MAC_PROTOCOL=!ip +``` + +Meaning: + +| Variable | Meaning | +|---|---| +| `CONFIGURE_TZSP_CAPTURE` | Master switch for capture changes made by full deploy | +| `CONFIGURE_IPV4_MANGLE` | Manage the project-owned IPv4 `forward` `sniff-tzsp` rule | +| `CONFIGURE_L2_SNIFFER` | Reconfigure global Packet Sniffer as the non-IPv4 complement | +| `START_L2_SNIFFER` | Start Packet Sniffer after configuring it | +| `TZSP_L2_INTERFACE` | Empty = all interfaces; otherwise exact interface/bridge name | +| `TZSP_L2_MAC_PROTOCOL` | Default `!ip`; normally should not be changed | +| `TZSP_PORT` | UDP port used by Rust TZSP receiver | + +For backward compatibility, an old `deploy-routeros.env` containing `CONFIGURE_SNIFFER` and `START_SNIFFER` is still accepted as an alias for the new master/start settings. `VLAN_ID` is no longer used by the capture deploy. + +## Full deployment + +Prepare configuration: + +```bash +cp deploy-routeros.env.example deploy-routeros.env +``` + +Build/upload the image as usual, then deploy: + +```bash +./scripts/deploy-routeros.sh routeros-suricata-tzsp-arm64.tar +``` + +The deployment: + +1. creates/updates the IDS container infrastructure, +2. starts the new container, +3. replaces only the MikroSuricata IPv4 mangle capture rule, +4. stops the global Packet Sniffer, +5. clears stale Packet Sniffer filters, +6. configures Packet Sniffer for `!ip`, +7. uses all RouterOS interfaces unless `TZSP_L2_INTERFACE` is explicitly set, +8. starts the L2 sniffer when `START_L2_SNIFFER=true`. + +## Switch an existing installation without rebuilding the container + +The capture migration is independent from the container image: + +```bash +./scripts/configure-routeros-tzsp-hybrid.sh +``` + +It reads the same `deploy-routeros.env` file. + +Preview the exact generated RouterOS script without making changes: + +```bash +DRY_RUN=true ./scripts/configure-routeros-tzsp-hybrid.sh +``` + +The helper only manages the project-owned IPv4 mangle rule. Packet Sniffer itself is a **global RouterOS facility**, so enabling `CONFIGURE_L2_SNIFFER=true` necessarily replaces its current global filters/settings. If the router already uses `/tool/sniffer` for another administrative purpose, either coordinate that use or set: + +```dotenv +CONFIGURE_L2_SNIFFER=false +``` + +and manage non-IPv4 capture manually. + +## Manual RouterOS template + +A standalone template is included: + +```text +routeros/02-tzsp-hybrid.rsc +``` + +Edit the local variables at the top before importing if the default container address differs. + +## Verification + +Check the RouterOS capture configuration: + +```routeros +/ip/firewall/mangle/print stats where comment="MikroSuricata TZSP IPv4" +/tool/sniffer/print +/interface/monitor-traffic veth-ids +``` + +Or from the project host: + +```bash +./scripts/routeros-status.sh +``` + +During a large IPv4 download, expect the mangle rule packet/byte counters and the container VETH TX counters to increase rapidly. In the MikroSuricata UI compare: + +```text +TZSP RX +Inspected throughput +capture % +queue % +loss/s +``` + +If `TZSP RX` is close to the expected traffic and capture remains close to 100%, RouterOS -> Rust and Rust -> TAP are keeping up. + +## Coverage and caveats + +- Packet Sniffer cannot necessarily see traffic switched entirely in hardware by a hardware-offloaded bridge. Broadcast/multicast behavior can differ by platform. +- The default mangle rule covers routed IPv4 only, not bridge-only IPv4 switching. +- IPv6 remains on Packet Sniffer in 0.11.2 and can therefore inherit Packet Sniffer throughput limits under sustained high-rate IPv6 traffic. +- `filter-interface=all` can expose the same low-level broadcast/L2 frame on more than one logical/physical observation point on some topologies. If this is noisy, set `TZSP_L2_INTERFACE` explicitly. +- The global Packet Sniffer has no per-consumer instance. MikroSuricata therefore cannot preserve a second independent sniffer configuration while also owning the non-IPv4 TZSP stream. +- Do not run an additional full-traffic Packet Sniffer stream to the same TZSP destination at the same time; duplicate packets will inflate traffic and Suricata processing. + +## Why existing mangle rules continue to work + +RouterOS defines `sniff-tzsp` as an action that sends a copy to a TZSP target and then passes the matched packet to the next mangle rule, similar to `passthrough`. The MikroSuricata rule therefore observes traffic without accepting, dropping, marking or rerouting the original packet. + +This is why the rule can safely sit at the beginning of `chain=forward`: later QoS, marking, policy-routing and other mangle rules remain in the processing path. diff --git a/grafana/README.md b/grafana/README.md index afd1b97..cbcf41d 100644 --- a/grafana/README.md +++ b/grafana/README.md @@ -8,8 +8,8 @@ already available in-memory counters and the last Suricata EVE stats snapshot. A scrape does not query SQLite, Redis, RouterOS, the Suricata control socket, or analytics endpoints. -The dashboard also uses monotonic TZSP traffic counters exported directly from -the in-memory flow tracker: +The dashboard also uses monotonic TZSP traffic counters exported from the Rust +receiver's 1 Hz telemetry bridge: ```text mikrosuricata_traffic_bytes_total{direction="total|inbound|outbound|internal|external"} @@ -21,6 +21,11 @@ bits/s. The top of the dashboard therefore shows current total, inbound and outbound throughput, packet rate, capture drops, a large throughput chart and traffic volume by direction for the selected time range. +Rust receive-path quality is exported separately. In particular, +`mikrosuricata_tzsp_receiver_kernel_udp_drops_total` is the Linux UDP socket +drop counter and should stay at zero during high-rate capture tests. The +dashboard includes it in **Sensor loss & decode errors**. + `/metrics` is protected by an IP/CIDR ACL configured through environment variables. The default allows loopback only: @@ -73,3 +78,7 @@ scrape_configs: Import `mikrosuricata-prometheus.json` in Grafana and select the Prometheus datasource from the dashboard variable. Rate, percentage and ratio panels are calculated in PromQL, not by MikroSuricata. + +## Rust TZSP data-plane health (0.11.0+) + +The receiver now exports both kernel and userspace back-pressure signals. In addition to `mikrosuricata_tzsp_receiver_kernel_udp_drops_total`, watch `mikrosuricata_tzsp_receiver_queue_dropped_datagrams_total`, `mikrosuricata_tzsp_receiver_queue_depth_batches`, `mikrosuricata_tzsp_receiver_queue_capacity_batches` and `mikrosuricata_tzsp_receiver_capture_efficiency_pct`. A small RouterOS kernel socket buffer is expected; loss should be judged by the drop counters, not by buffer size alone. diff --git a/grafana/mikrosuricata-prometheus.json b/grafana/mikrosuricata-prometheus.json index 61d7951..a8d4280 100644 --- a/grafana/mikrosuricata-prometheus.json +++ b/grafana/mikrosuricata-prometheus.json @@ -648,7 +648,7 @@ { "id": 25, "type": "bargauge", - "title": "Traffic mix \u00b7 selected range", + "title": "Traffic mix · selected range", "datasource": { "type": "prometheus", "uid": "${datasource}" @@ -873,8 +873,8 @@ { "id": 26, "type": "stat", - "title": "Active flows", - "description": "Currently tracked live L3/L4 flows.", + "title": "TZSP RX buffer", + "description": "Actual Linux UDP receive buffer allocated to the Rust TZSP receiver.", "datasource": { "type": "prometheus", "uid": "${datasource}" @@ -887,7 +887,7 @@ }, "fieldConfig": { "defaults": { - "unit": "short", + "unit": "bytes", "thresholds": { "mode": "absolute", "steps": [ @@ -926,7 +926,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "max(mikrosuricata_flow_tracker_active_flows{instance=~\"$instance\"})", + "expr": "max(mikrosuricata_tzsp_receiver_rcvbuf_bytes{instance=~\"$instance\"})", "legendFormat": "", "range": true, "refId": "A" @@ -1322,7 +1322,7 @@ "id": 8, "type": "timeseries", "title": "Sensor loss & decode errors", - "description": "Rates that should stay near zero: kernel drops, injection errors, TZSP decode errors and Suricata alert queue overflow.", + "description": "Rates that should stay near zero: Rust UDP socket drops, Suricata kernel drops, TAP injection errors, TZSP decode errors and alert queue overflow.", "datasource": { "type": "prometheus", "uid": "${datasource}" @@ -1438,6 +1438,17 @@ "legendFormat": "Alert queue overflow", "range": true, "refId": "D" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rate(mikrosuricata_tzsp_receiver_kernel_udp_drops_total{instance=~\"$instance\"}[$__rate_interval]))", + "legendFormat": "TZSP UDP drops", + "range": true, + "refId": "E" } ] }, @@ -2048,8 +2059,8 @@ { "id": 14, "type": "timeseries", - "title": "Live sessions", - "description": "", + "title": "Rust dataplane packet path", + "description": "Inspected packet rate versus kernel UDP loss and userspace queue loss in the Rust TZSP data-plane.", "datasource": { "type": "prometheus", "uid": "${datasource}" @@ -2062,7 +2073,7 @@ }, "fieldConfig": { "defaults": { - "unit": "short", + "unit": "pps", "custom": { "drawStyle": "line", "lineInterpolation": "smooth", @@ -2128,8 +2139,8 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "max(mikrosuricata_flow_tracker_active_flows{instance=~\"$instance\"})", - "legendFormat": "Active flows", + "expr": "sum(rate(mikrosuricata_traffic_packets_total{instance=~\"$instance\",direction=\"total\"}[$__rate_interval]))", + "legendFormat": "Inspected packets/s", "range": true, "refId": "A" }, @@ -2139,8 +2150,8 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "max(mikrosuricata_event_bus_subscribers{instance=~\"$instance\"})", - "legendFormat": "WebSocket subscribers", + "expr": "sum(rate(mikrosuricata_tzsp_receiver_kernel_udp_drops_total{instance=~\"$instance\"}[$__rate_interval]))", + "legendFormat": "Kernel UDP drops/s", "range": true, "refId": "B" }, @@ -2150,8 +2161,8 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rate(mikrosuricata_flow_tracker_evicted_flows_total{instance=~\"$instance\"}[$__rate_interval]))", - "legendFormat": "Evictions/s", + "expr": "sum(rate(mikrosuricata_tzsp_receiver_queue_dropped_datagrams_total{instance=~\"$instance\"}[$__rate_interval]))", + "legendFormat": "Userspace queue drops/s", "range": true, "refId": "C" } diff --git a/routeros/02-tzsp-hybrid.rsc b/routeros/02-tzsp-hybrid.rsc new file mode 100644 index 0000000..6ad03b7 --- /dev/null +++ b/routeros/02-tzsp-hybrid.rsc @@ -0,0 +1,40 @@ +# MikroSuricata hybrid TZSP capture template. +# Review the target IP/port before importing. l2Interface="all" means all interfaces; +# set it to a real RouterOS interface/bridge only if you intentionally want to +# constrain the low-volume non-IPv4 stream. + +:local target "172.31.255.2" +:local port 37008 +:local l2Interface "all" +:local l2MacProtocol "!ip" + +# Routed IPv4 uses mangle sniff-tzsp. The action passes processing to the next +# mangle rule after cloning the packet, so existing mangle rules remain active. +/ip/firewall/mangle/remove [find where comment="MikroSuricata TZSP IPv4"] +:if ([:len [/ip/firewall/mangle/find]] > 0) do={ + /ip/firewall/mangle/add chain=forward action=sniff-tzsp sniff-target=$target sniff-target-port=$port comment="MikroSuricata TZSP IPv4" place-before=0 +} else={ + /ip/firewall/mangle/add chain=forward action=sniff-tzsp sniff-target=$target sniff-target-port=$port comment="MikroSuricata TZSP IPv4" +} + +# Everything except IPv4 EtherType is streamed by Packet Sniffer. This includes +# ARP, IPv6, LLDP, PPPoE discovery and other non-IPv4 Ethernet protocols. +/tool/sniffer/stop +:if ($l2Interface != "all") do={ + :if ([:len [/interface/find where name=$l2Interface]] = 0) do={ + :error ("L2 capture interface not found: " . $l2Interface) + } +} +/tool/sniffer/set only-headers=no max-packet-size=2048 streaming-enabled=yes filter-stream=yes filter-interface=$l2Interface filter-mac-address="" filter-src-mac-address="" filter-dst-mac-address="" filter-mac-protocol=$l2MacProtocol filter-ip-address="" filter-src-ip-address="" filter-dst-ip-address="" filter-ipv6-address="" filter-src-ipv6-address="" filter-dst-ipv6-address="" filter-ip-protocol="" filter-port="" filter-src-port="" filter-dst-port="" filter-vlan="" filter-cpu="" filter-size="" filter-direction=any filter-operator-between-entries=or + +# Current RouterOS uses streaming-server=IP:PORT. Fallback supports older builds +# that expose streaming-port separately. +:do { + /tool/sniffer/set streaming-server=($target . ":" . $port) +} on-error={ + /tool/sniffer/set streaming-server=$target streaming-port=$port +} +/tool/sniffer/start + +/ip/firewall/mangle/print stats where comment="MikroSuricata TZSP IPv4" +/tool/sniffer/print diff --git a/routeros/rollback.rsc b/routeros/rollback.rsc index e09fddf..1e6f6f8 100644 --- a/routeros/rollback.rsc +++ b/routeros/rollback.rsc @@ -2,6 +2,7 @@ # directories on the external disk. Review before importing. /tool/sniffer/stop +/ip/firewall/mangle/remove [find where comment="MikroSuricata TZSP IPv4"] :if ([:len [/container/find where name="suricata-ids"]] > 0) do={ :local cid [/container/find where name="suricata-ids"] :if ([/container/get $cid status] = "running") do={ diff --git a/rust/tzsp-receiver/Cargo.toml b/rust/tzsp-receiver/Cargo.toml new file mode 100644 index 0000000..a558ba1 --- /dev/null +++ b/rust/tzsp-receiver/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "mikrosuricata-tzsp" +version = "0.11.2" +edition = "2021" +publish = false + +[profile.release] +lto = "thin" +codegen-units = 1 +panic = "abort" +strip = true diff --git a/rust/tzsp-receiver/src/main.rs b/rust/tzsp-receiver/src/main.rs new file mode 100644 index 0000000..9a35ba4 --- /dev/null +++ b/rust/tzsp-receiver/src/main.rs @@ -0,0 +1,1144 @@ +use std::env; +use std::ffi::{c_int, c_ulong, c_void}; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Write}; +use std::mem; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, UdpSocket}; +use std::os::fd::{AsRawFd, RawFd}; +use std::os::unix::net::UnixDatagram; +use std::path::Path; +use std::process::{self, Command}; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::mpsc::{sync_channel, SyncSender, TryRecvError}; +use std::sync::Arc; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +const TZSP_VERSION: u8 = 1; +const TZSP_TYPE_RECEIVED: u8 = 0; +const TZSP_TYPE_TRANSMIT: u8 = 1; +const TZSP_ENCAP_ETHERNET: u16 = 1; +const TAG_PADDING: u8 = 0; +const TAG_END: u8 = 1; + +const TUNSETIFF: c_ulong = 0x400454ca; +const IFF_TAP: i16 = 0x0002; +const IFF_NO_PI: i16 = 0x1000; + +const SOL_SOCKET: c_int = 1; +const SO_REUSEADDR: c_int = 2; +const SO_RCVBUF: c_int = 8; +const SO_RCVBUFFORCE: c_int = 33; +const MSG_TRUNC: c_int = 0x20; +const MSG_WAITFORONE: c_int = 0x10000; +const EINTR: i32 = 4; + +const DEFAULT_BATCH: usize = 256; +const MAX_BATCH: usize = 1024; +const DEFAULT_QUEUE_MB: usize = 64; +const MIN_POOL_BATCHES: usize = 6; +const MAX_POOL_BATCHES: usize = 256; + +#[repr(C)] +struct Iovec { + iov_base: *mut c_void, + iov_len: usize, +} + +#[repr(C)] +struct Msghdr { + msg_name: *mut c_void, + msg_namelen: u32, + msg_iov: *mut Iovec, + msg_iovlen: usize, + msg_control: *mut c_void, + msg_controllen: usize, + msg_flags: c_int, +} + +#[repr(C)] +struct Mmsghdr { + msg_hdr: Msghdr, + msg_len: u32, +} + +extern "C" { + fn ioctl(fd: c_int, request: c_ulong, ...) -> c_int; + fn setsockopt( + fd: c_int, + level: c_int, + optname: c_int, + optval: *const c_void, + optlen: u32, + ) -> c_int; + fn getsockopt( + fd: c_int, + level: c_int, + optname: c_int, + optval: *mut c_void, + optlen: *mut u32, + ) -> c_int; + fn recvmmsg( + fd: c_int, + msgvec: *mut Mmsghdr, + vlen: u32, + flags: c_int, + timeout: *mut c_void, + ) -> c_int; +} + +#[derive(Clone, Copy, Debug, Default)] +struct TrafficCounters { + bytes_total: u64, + bytes_in: u64, + bytes_out: u64, + bytes_internal: u64, + bytes_external: u64, + packets_total: u64, + packets_in: u64, + packets_out: u64, + packets_internal: u64, + packets_external: u64, +} + +impl TrafficCounters { + fn observe(&mut self, bytes: usize, direction: Direction) { + let bytes = bytes as u64; + self.bytes_total = self.bytes_total.saturating_add(bytes); + self.packets_total = self.packets_total.saturating_add(1); + match direction { + Direction::Inbound => { + self.bytes_in = self.bytes_in.saturating_add(bytes); + self.packets_in = self.packets_in.saturating_add(1); + } + Direction::Outbound => { + self.bytes_out = self.bytes_out.saturating_add(bytes); + self.packets_out = self.packets_out.saturating_add(1); + } + Direction::Internal => { + self.bytes_internal = self.bytes_internal.saturating_add(bytes); + self.packets_internal = self.packets_internal.saturating_add(1); + } + Direction::External => { + self.bytes_external = self.bytes_external.saturating_add(bytes); + self.packets_external = self.packets_external.saturating_add(1); + } + } + } + + fn delta(self, previous: Self) -> Self { + Self { + bytes_total: self.bytes_total.saturating_sub(previous.bytes_total), + bytes_in: self.bytes_in.saturating_sub(previous.bytes_in), + bytes_out: self.bytes_out.saturating_sub(previous.bytes_out), + bytes_internal: self.bytes_internal.saturating_sub(previous.bytes_internal), + bytes_external: self.bytes_external.saturating_sub(previous.bytes_external), + packets_total: self.packets_total.saturating_sub(previous.packets_total), + packets_in: self.packets_in.saturating_sub(previous.packets_in), + packets_out: self.packets_out.saturating_sub(previous.packets_out), + packets_internal: self.packets_internal.saturating_sub(previous.packets_internal), + packets_external: self.packets_external.saturating_sub(previous.packets_external), + } + } +} + +#[derive(Clone, Copy, Debug)] +enum Direction { + Inbound, + Outbound, + Internal, + External, +} + +#[derive(Clone, Copy, Debug)] +enum Network { + V4 { network: u32, mask: u32 }, + V6 { network: u128, mask: u128 }, +} + +impl Network { + fn parse(text: &str) -> Option { + let (addr_text, prefix_text) = text.trim().split_once('/')?; + let addr = IpAddr::from_str(addr_text.trim()).ok()?; + let prefix: u8 = prefix_text.trim().parse().ok()?; + match addr { + IpAddr::V4(value) if prefix <= 32 => { + let mask = if prefix == 0 { 0 } else { u32::MAX << (32 - prefix) }; + Some(Self::V4 { + network: u32::from(value) & mask, + mask, + }) + } + IpAddr::V6(value) if prefix <= 128 => { + let mask = if prefix == 0 { 0 } else { u128::MAX << (128 - prefix) }; + Some(Self::V6 { + network: u128::from(value) & mask, + mask, + }) + } + _ => None, + } + } + + fn contains(&self, ip: IpAddr) -> bool { + match (*self, ip) { + (Self::V4 { network, mask }, IpAddr::V4(value)) => (u32::from(value) & mask) == network, + (Self::V6 { network, mask }, IpAddr::V6(value)) => (u128::from(value) & mask) == network, + _ => false, + } + } +} + +struct Config { + bind: String, + port: u16, + tap_name: String, + tap_mtu: u32, + telemetry_socket: String, + rcvbuf_bytes: c_int, + batch_size: usize, + datagram_bytes: usize, + queue_mb: usize, + networks: Vec, +} + +impl Config { + fn from_env() -> Self { + let tap_mtu = env_u64("TAP_MTU", 9000).min(u32::MAX as u64) as u32; + let default_datagram = round_up((tap_mtu as usize).saturating_add(2048), 1024) + .clamp(2048, 65_535); + let networks = env::var("MONITORED_NETWORKS") + .unwrap_or_else(|_| "192.168.0.0/16,10.0.0.0/8,172.16.0.0/12".to_string()) + .split(',') + .filter_map(Network::parse) + .collect(); + Self { + bind: env::var("TZSP_BIND").unwrap_or_else(|_| "0.0.0.0".to_string()), + port: env_u64("TZSP_PORT", 37008).min(u16::MAX as u64) as u16, + tap_name: env::var("TAP_NAME").unwrap_or_else(|_| "suritap0".to_string()), + tap_mtu, + telemetry_socket: env::var("TZSP_TELEMETRY_SOCKET") + .unwrap_or_else(|_| "/run/mikrosuricata/tzsp-telemetry.sock".to_string()), + rcvbuf_bytes: env_u64("TZSP_RCVBUF_BYTES", 32 * 1024 * 1024) + .min(c_int::MAX as u64) as c_int, + batch_size: env_u64("TZSP_BATCH_SIZE", DEFAULT_BATCH as u64) + .clamp(1, MAX_BATCH as u64) as usize, + datagram_bytes: env_u64("TZSP_DATAGRAM_BYTES", default_datagram as u64) + .clamp(2048, 65_535) as usize, + queue_mb: env_u64("TZSP_QUEUE_MB", DEFAULT_QUEUE_MB as u64) + .clamp(8, 512) as usize, + networks, + } + } + + fn pool_batches(&self) -> usize { + pool_batches(self.queue_mb, self.batch_size, self.datagram_bytes) + } +} + +/// One reusable recvmmsg() buffer. The raw pointers always refer to backing +/// allocations owned by this struct. Moving the struct only moves the Vec +/// handles, not those allocations, and no Vec is resized after construction. +struct BatchBuffer { + storage: Vec, + iovecs: Vec, + messages: Vec, + received: usize, +} + +// A BatchBuffer has one owner at a time and is transferred producer -> worker +// -> producer through bounded channels. Its internal pointers are never shared. +unsafe impl Send for BatchBuffer {} + +impl BatchBuffer { + fn new(batch_size: usize, datagram_bytes: usize) -> Self { + let mut storage = vec![0u8; batch_size.saturating_mul(datagram_bytes)]; + let mut iovecs = Vec::with_capacity(batch_size); + for index in 0..batch_size { + let base = unsafe { storage.as_mut_ptr().add(index * datagram_bytes) }; + iovecs.push(Iovec { + iov_base: base.cast::(), + iov_len: datagram_bytes, + }); + } + let mut messages = Vec::with_capacity(batch_size); + for index in 0..batch_size { + messages.push(Mmsghdr { + msg_hdr: Msghdr { + msg_name: std::ptr::null_mut(), + msg_namelen: 0, + msg_iov: &mut iovecs[index] as *mut Iovec, + msg_iovlen: 1, + msg_control: std::ptr::null_mut(), + msg_controllen: 0, + msg_flags: 0, + }, + msg_len: 0, + }); + } + Self { + storage, + iovecs, + messages, + received: 0, + } + } + + fn receive_blocking(&mut self, fd: RawFd, datagram_bytes: usize) -> io::Result { + for (index, message) in self.messages.iter_mut().enumerate() { + self.iovecs[index].iov_len = datagram_bytes; + message.msg_len = 0; + message.msg_hdr.msg_flags = 0; + } + let received = unsafe { + recvmmsg( + fd, + self.messages.as_mut_ptr(), + self.messages.len() as u32, + MSG_WAITFORONE, + std::ptr::null_mut(), + ) + }; + if received < 0 { + return Err(io::Error::last_os_error()); + } + self.received = received as usize; + Ok(self.received) + } + + fn datagram(&self, index: usize, datagram_bytes: usize) -> &[u8] { + let length = self.messages[index].msg_len as usize; + let start = index * datagram_bytes; + &self.storage[start..start + length.min(datagram_bytes)] + } + + fn is_truncated(&self, index: usize) -> bool { + (self.messages[index].msg_hdr.msg_flags & MSG_TRUNC) != 0 + } + + fn received_bytes(&self) -> u64 { + self.messages[..self.received] + .iter() + .map(|message| message.msg_len as u64) + .sum() + } +} + +#[derive(Default)] +struct SharedStats { + tzsp_datagrams: AtomicU64, + tzsp_rx_bytes: AtomicU64, + rx_batches: AtomicU64, + queue_dropped_datagrams: AtomicU64, + truncated_datagrams: AtomicU64, + tzsp_decode_errors: AtomicU64, + tzsp_unsupported: AtomicU64, + frames_injected: AtomicU64, + inject_errors: AtomicU64, + bytes_total: AtomicU64, + bytes_in: AtomicU64, + bytes_out: AtomicU64, + bytes_internal: AtomicU64, + bytes_external: AtomicU64, + packets_total: AtomicU64, + packets_in: AtomicU64, + packets_out: AtomicU64, + packets_internal: AtomicU64, + packets_external: AtomicU64, + queued_batches: AtomicUsize, + queue_high_water_batches: AtomicUsize, + last_packet_ms: AtomicU64, + rx_alive: AtomicBool, + worker_alive: AtomicBool, + fatal: AtomicBool, +} + +impl SharedStats { + fn traffic(&self) -> TrafficCounters { + TrafficCounters { + bytes_total: self.bytes_total.load(Ordering::Relaxed), + bytes_in: self.bytes_in.load(Ordering::Relaxed), + bytes_out: self.bytes_out.load(Ordering::Relaxed), + bytes_internal: self.bytes_internal.load(Ordering::Relaxed), + bytes_external: self.bytes_external.load(Ordering::Relaxed), + packets_total: self.packets_total.load(Ordering::Relaxed), + packets_in: self.packets_in.load(Ordering::Relaxed), + packets_out: self.packets_out.load(Ordering::Relaxed), + packets_internal: self.packets_internal.load(Ordering::Relaxed), + packets_external: self.packets_external.load(Ordering::Relaxed), + } + } + + fn add_traffic(&self, traffic: TrafficCounters) { + self.bytes_total.fetch_add(traffic.bytes_total, Ordering::Relaxed); + self.bytes_in.fetch_add(traffic.bytes_in, Ordering::Relaxed); + self.bytes_out.fetch_add(traffic.bytes_out, Ordering::Relaxed); + self.bytes_internal.fetch_add(traffic.bytes_internal, Ordering::Relaxed); + self.bytes_external.fetch_add(traffic.bytes_external, Ordering::Relaxed); + self.packets_total.fetch_add(traffic.packets_total, Ordering::Relaxed); + self.packets_in.fetch_add(traffic.packets_in, Ordering::Relaxed); + self.packets_out.fetch_add(traffic.packets_out, Ordering::Relaxed); + self.packets_internal.fetch_add(traffic.packets_internal, Ordering::Relaxed); + self.packets_external.fetch_add(traffic.packets_external, Ordering::Relaxed); + } + + fn queue_push(&self) { + let depth = self.queued_batches.fetch_add(1, Ordering::Relaxed) + 1; + let mut high = self.queue_high_water_batches.load(Ordering::Relaxed); + while depth > high { + match self.queue_high_water_batches.compare_exchange_weak( + high, + depth, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(value) => high = value, + } + } + } + + fn queue_pop(&self) { + let _ = self.queued_batches.fetch_update( + Ordering::Relaxed, + Ordering::Relaxed, + |value| Some(value.saturating_sub(1)), + ); + } +} + +fn main() { + if let Err(error) = run() { + eprintln!("[tzsp-rust] fatal: {error}"); + process::exit(1); + } +} + +fn run() -> io::Result<()> { + let cfg = Arc::new(Config::from_env()); + if cfg.tap_name.as_bytes().len() >= 16 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "TAP_NAME must be shorter than 16 bytes", + )); + } + + let tap = open_tap(&cfg.tap_name, cfg.tap_mtu)?; + let socket = UdpSocket::bind((cfg.bind.as_str(), cfg.port))?; + configure_udp_socket(socket.as_raw_fd(), cfg.rcvbuf_bytes)?; + let actual_rcvbuf = socket_rcvbuf(socket.as_raw_fd()).unwrap_or(0); + let socket_inode = socket_inode(socket.as_raw_fd()); + if actual_rcvbuf > 0 && actual_rcvbuf < cfg.rcvbuf_bytes { + eprintln!( + "[tzsp-rust] kernel capped SO_RCVBUF at {} B (requested {} B); dedicated RX + userspace queue will absorb bursts", + actual_rcvbuf, cfg.rcvbuf_bytes + ); + } + + let telemetry = UnixDatagram::unbound()?; + telemetry.set_nonblocking(true)?; + + let pool_batches = cfg.pool_batches(); + let queue_capacity_batches = pool_batches.saturating_sub(2).max(1); + let batch_bytes = cfg.batch_size.saturating_mul(cfg.datagram_bytes); + let queue_capacity_bytes = queue_capacity_batches.saturating_mul(batch_bytes); + + println!( + "[tzsp-rust] udp://{}:{} -> {} MTU {} | batch={} datagram={}B pool={} queue~{}MiB SO_RCVBUF={}B", + cfg.bind, + cfg.port, + cfg.tap_name, + cfg.tap_mtu, + cfg.batch_size, + cfg.datagram_bytes, + pool_batches, + queue_capacity_bytes / (1024 * 1024), + actual_rcvbuf + ); + + let (free_tx, free_rx) = sync_channel::(pool_batches); + let (filled_tx, filled_rx) = sync_channel::(pool_batches); + for _ in 0..pool_batches { + free_tx + .send(BatchBuffer::new(cfg.batch_size, cfg.datagram_bytes)) + .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "cannot initialize batch pool"))?; + } + + let stats = Arc::new(SharedStats::default()); + + let worker_stats = Arc::clone(&stats); + let worker_cfg = Arc::clone(&cfg); + let worker_free_tx = free_tx.clone(); + thread::Builder::new() + .name("tzsp-tap-worker".to_string()) + .spawn(move || { + worker_stats.worker_alive.store(true, Ordering::Release); + let result = worker_loop(tap, filled_rx, worker_free_tx, &worker_cfg, &worker_stats); + if let Err(error) = result { + eprintln!("[tzsp-rust] TAP worker stopped: {error}"); + worker_stats.fatal.store(true, Ordering::Release); + } + worker_stats.worker_alive.store(false, Ordering::Release); + }) + .map_err(|error| io::Error::new(io::ErrorKind::Other, format!("spawn TAP worker: {error}")))?; + + let rx_stats = Arc::clone(&stats); + let rx_cfg = Arc::clone(&cfg); + thread::Builder::new() + .name("tzsp-udp-rx".to_string()) + .spawn(move || { + rx_stats.rx_alive.store(true, Ordering::Release); + let result = rx_loop(socket, free_rx, filled_tx, &rx_cfg, &rx_stats); + if let Err(error) = result { + eprintln!("[tzsp-rust] UDP RX stopped: {error}"); + rx_stats.fatal.store(true, Ordering::Release); + } + rx_stats.rx_alive.store(false, Ordering::Release); + }) + .map_err(|error| io::Error::new(io::ErrorKind::Other, format!("spawn UDP RX: {error}")))?; + + let started = Instant::now(); + let mut previous_traffic = TrafficCounters::default(); + let mut previous_tzsp_datagrams = 0u64; + let mut previous_tzsp_rx_bytes = 0u64; + let mut previous_queue_drops = 0u64; + let mut previous_truncated = 0u64; + let mut previous_kernel_drops = 0u64; + let mut sample_started = Instant::now(); + + // Rust owns TAP/socket at this point. Let Python start Suricata immediately; + // packet ingestion and TAP injection continue independently of telemetry. + send_telemetry( + &telemetry, + &cfg, + &stats, + TrafficCounters::default(), + 1, + actual_rcvbuf, + 0, + 0, + 0, + 0, + 0, + 0, + queue_capacity_batches, + queue_capacity_bytes, + started.elapsed(), + ); + + loop { + thread::sleep(Duration::from_millis(200)); + if stats.fatal.load(Ordering::Acquire) { + return Err(io::Error::new(io::ErrorKind::Other, "TZSP data-plane worker failed")); + } + + let elapsed = sample_started.elapsed(); + if elapsed < Duration::from_secs(1) { + continue; + } + + let traffic = stats.traffic(); + let interval = traffic.delta(previous_traffic); + previous_traffic = traffic; + + let tzsp_datagrams = stats.tzsp_datagrams.load(Ordering::Relaxed); + let tzsp_rx_bytes = stats.tzsp_rx_bytes.load(Ordering::Relaxed); + let queue_drops = stats.queue_dropped_datagrams.load(Ordering::Relaxed); + let truncated = stats.truncated_datagrams.load(Ordering::Relaxed); + let kernel_drops = socket_inode + .as_deref() + .and_then(udp_socket_drops) + .unwrap_or(previous_kernel_drops); + + let rx_datagrams_interval = tzsp_datagrams.saturating_sub(previous_tzsp_datagrams); + let rx_bytes_interval = tzsp_rx_bytes.saturating_sub(previous_tzsp_rx_bytes); + let queue_drops_interval = queue_drops.saturating_sub(previous_queue_drops); + let truncated_interval = truncated.saturating_sub(previous_truncated); + let kernel_drops_interval = kernel_drops.saturating_sub(previous_kernel_drops); + + previous_tzsp_datagrams = tzsp_datagrams; + previous_tzsp_rx_bytes = tzsp_rx_bytes; + previous_queue_drops = queue_drops; + previous_truncated = truncated; + previous_kernel_drops = kernel_drops; + + send_telemetry( + &telemetry, + &cfg, + &stats, + interval, + elapsed.as_millis().max(1) as u64, + actual_rcvbuf, + kernel_drops, + kernel_drops_interval, + queue_drops_interval, + truncated_interval, + rx_datagrams_interval, + rx_bytes_interval, + queue_capacity_batches, + queue_capacity_bytes, + started.elapsed(), + ); + + sample_started = Instant::now(); + } +} + +fn rx_loop( + socket: UdpSocket, + free_rx: std::sync::mpsc::Receiver, + filled_tx: SyncSender, + cfg: &Config, + stats: &SharedStats, +) -> io::Result<()> { + let fd = socket.as_raw_fd(); + let mut scratch = BatchBuffer::new(cfg.batch_size, cfg.datagram_bytes); + + loop { + let maybe_buffer = match free_rx.try_recv() { + Ok(buffer) => Some(buffer), + Err(TryRecvError::Empty) => None, + Err(TryRecvError::Disconnected) => { + return Err(io::Error::new(io::ErrorKind::BrokenPipe, "batch pool disconnected")); + } + }; + + if let Some(mut buffer) = maybe_buffer { + let received = loop { + match buffer.receive_blocking(fd, cfg.datagram_bytes) { + Ok(value) => break value, + Err(error) if error.raw_os_error() == Some(EINTR) => continue, + Err(error) => return Err(error), + } + }; + if received == 0 { + continue; + } + stats.tzsp_datagrams.fetch_add(received as u64, Ordering::Relaxed); + stats.tzsp_rx_bytes.fetch_add(buffer.received_bytes(), Ordering::Relaxed); + stats.rx_batches.fetch_add(1, Ordering::Relaxed); + stats.queue_push(); + if filled_tx.send(buffer).is_err() { + stats.queue_pop(); + return Err(io::Error::new(io::ErrorKind::BrokenPipe, "TAP worker disconnected")); + } + } else { + // Downstream is behind. Never stop draining the tiny kernel UDP + // queue: receive into a reusable scratch batch and account the loss + // explicitly. This prevents a TAP/Suricata stall from turning into + // opaque kernel drops and keeps the RX thread hot. + let received = loop { + match scratch.receive_blocking(fd, cfg.datagram_bytes) { + Ok(value) => break value, + Err(error) if error.raw_os_error() == Some(EINTR) => continue, + Err(error) => return Err(error), + } + }; + if received == 0 { + continue; + } + stats.tzsp_datagrams.fetch_add(received as u64, Ordering::Relaxed); + stats.tzsp_rx_bytes.fetch_add(scratch.received_bytes(), Ordering::Relaxed); + stats.rx_batches.fetch_add(1, Ordering::Relaxed); + stats + .queue_dropped_datagrams + .fetch_add(received as u64, Ordering::Relaxed); + } + } +} + +fn worker_loop( + mut tap: File, + filled_rx: std::sync::mpsc::Receiver, + free_tx: SyncSender, + cfg: &Config, + stats: &SharedStats, +) -> io::Result<()> { + while let Ok(buffer) = filled_rx.recv() { + stats.queue_pop(); + let mut traffic = TrafficCounters::default(); + let mut decoded_errors = 0u64; + let mut unsupported = 0u64; + let mut truncated = 0u64; + let mut injected = 0u64; + let mut inject_errors = 0u64; + + for index in 0..buffer.received { + if buffer.is_truncated(index) { + truncated = truncated.saturating_add(1); + continue; + } + let datagram = buffer.datagram(index, cfg.datagram_bytes); + match decode_tzsp(datagram) { + Ok((encapsulation, frame)) => { + if encapsulation != TZSP_ENCAP_ETHERNET { + unsupported = unsupported.saturating_add(1); + continue; + } + let direction = frame_direction(frame, &cfg.networks); + // Each TAP write is exactly one Ethernet frame. write_all() + // must not be used because a short write would create a + // second TAP packet from the remainder. + match tap.write(frame) { + Ok(written) if written == frame.len() => { + injected = injected.saturating_add(1); + traffic.observe(frame.len(), direction); + } + Ok(written) => { + inject_errors = inject_errors.saturating_add(1); + if inject_errors <= 3 { + eprintln!("[tzsp-rust] short TAP write: {written}/{} bytes", frame.len()); + } + } + Err(error) => { + inject_errors = inject_errors.saturating_add(1); + if inject_errors <= 3 { + eprintln!("[tzsp-rust] TAP write failed: {error}"); + } + } + } + } + Err(_) => decoded_errors = decoded_errors.saturating_add(1), + } + } + + if truncated > 0 { + stats.truncated_datagrams.fetch_add(truncated, Ordering::Relaxed); + } + if decoded_errors > 0 { + stats.tzsp_decode_errors.fetch_add(decoded_errors, Ordering::Relaxed); + } + if unsupported > 0 { + stats.tzsp_unsupported.fetch_add(unsupported, Ordering::Relaxed); + } + if injected > 0 { + stats.frames_injected.fetch_add(injected, Ordering::Relaxed); + stats.last_packet_ms.store(unix_ms(), Ordering::Relaxed); + } + if inject_errors > 0 { + stats.inject_errors.fetch_add(inject_errors, Ordering::Relaxed); + } + stats.add_traffic(traffic); + + if free_tx.send(buffer).is_err() { + return Err(io::Error::new(io::ErrorKind::BrokenPipe, "UDP RX disconnected")); + } + } + Err(io::Error::new(io::ErrorKind::BrokenPipe, "filled batch channel disconnected")) +} + +fn decode_tzsp(data: &[u8]) -> Result<(u16, &[u8]), ()> { + if data.len() < 5 || data[0] != TZSP_VERSION { + return Err(()); + } + if data[1] != TZSP_TYPE_RECEIVED && data[1] != TZSP_TYPE_TRANSMIT { + return Err(()); + } + let encapsulation = u16::from_be_bytes([data[2], data[3]]); + let mut offset = 4usize; + let mut found_end = false; + while offset < data.len() { + let tag = data[offset]; + offset += 1; + if tag == TAG_PADDING { + continue; + } + if tag == TAG_END { + found_end = true; + break; + } + if offset >= data.len() { + return Err(()); + } + let length = data[offset] as usize; + offset += 1; + if offset.saturating_add(length) > data.len() { + return Err(()); + } + offset += length; + } + if !found_end || offset >= data.len() { + return Err(()); + } + Ok((encapsulation, &data[offset..])) +} + +fn frame_direction(frame: &[u8], networks: &[Network]) -> Direction { + let Some((src, dst)) = ethernet_endpoints(frame) else { + return Direction::External; + }; + let src_local = networks.iter().any(|network| network.contains(src)); + let dst_local = networks.iter().any(|network| network.contains(dst)); + match (src_local, dst_local) { + (true, false) => Direction::Outbound, + (false, true) => Direction::Inbound, + (true, true) => Direction::Internal, + (false, false) => Direction::External, + } +} + +fn ethernet_endpoints(frame: &[u8]) -> Option<(IpAddr, IpAddr)> { + if frame.len() < 14 { + return None; + } + let mut offset = 14usize; + let mut ethertype = u16::from_be_bytes([frame[12], frame[13]]); + for _ in 0..2 { + if !matches!(ethertype, 0x8100 | 0x88a8 | 0x9100) { + break; + } + if frame.len() < offset + 4 { + return None; + } + ethertype = u16::from_be_bytes([frame[offset + 2], frame[offset + 3]]); + offset += 4; + } + match ethertype { + 0x0800 => { + if frame.len() < offset + 20 || frame[offset] >> 4 != 4 { + return None; + } + let src = Ipv4Addr::new( + frame[offset + 12], + frame[offset + 13], + frame[offset + 14], + frame[offset + 15], + ); + let dst = Ipv4Addr::new( + frame[offset + 16], + frame[offset + 17], + frame[offset + 18], + frame[offset + 19], + ); + Some((IpAddr::V4(src), IpAddr::V4(dst))) + } + 0x86dd => { + if frame.len() < offset + 40 || frame[offset] >> 4 != 6 { + return None; + } + let src = Ipv6Addr::from(<[u8; 16]>::try_from(&frame[offset + 8..offset + 24]).ok()?); + let dst = Ipv6Addr::from(<[u8; 16]>::try_from(&frame[offset + 24..offset + 40]).ok()?); + Some((IpAddr::V6(src), IpAddr::V6(dst))) + } + _ => None, + } +} + +fn open_tap(name: &str, mtu: u32) -> io::Result { + let file = OpenOptions::new().read(true).write(true).open("/dev/net/tun")?; + let mut ifreq = [0u8; 40]; + let name_bytes = name.as_bytes(); + ifreq[..name_bytes.len()].copy_from_slice(name_bytes); + let flags = (IFF_TAP | IFF_NO_PI).to_ne_bytes(); + ifreq[16] = flags[0]; + ifreq[17] = flags[1]; + let result = unsafe { ioctl(file.as_raw_fd(), TUNSETIFF, ifreq.as_mut_ptr()) }; + if result < 0 { + return Err(io::Error::last_os_error()); + } + let mtu_text = mtu.to_string(); + run_ip(["link", "set", "dev", name, "mtu", mtu_text.as_str()])?; + run_ip(["link", "set", "dev", name, "up"])?; + Ok(file) +} + +fn run_ip(args: [&str; N]) -> io::Result<()> { + let status = Command::new("ip").args(args).status()?; + if status.success() { + Ok(()) + } else { + Err(io::Error::new( + io::ErrorKind::Other, + format!("ip command exited with {status}"), + )) + } +} + +fn configure_udp_socket(fd: RawFd, requested: c_int) -> io::Result<()> { + let reuse: c_int = 1; + let reuse_result = unsafe { + setsockopt( + fd, + SOL_SOCKET, + SO_REUSEADDR, + (&reuse as *const c_int).cast::(), + mem::size_of::() as u32, + ) + }; + if reuse_result < 0 { + return Err(io::Error::last_os_error()); + } + + // RouterOS containers frequently cannot raise net.core.rmem_max. Try the + // force variant, but correctness/performance does not depend on it anymore: + // a dedicated RX thread drains this small queue into the userspace pool. + let force_result = unsafe { + setsockopt( + fd, + SOL_SOCKET, + SO_RCVBUFFORCE, + (&requested as *const c_int).cast::(), + mem::size_of::() as u32, + ) + }; + if force_result < 0 { + let regular_result = unsafe { + setsockopt( + fd, + SOL_SOCKET, + SO_RCVBUF, + (&requested as *const c_int).cast::(), + mem::size_of::() as u32, + ) + }; + if regular_result < 0 { + return Err(io::Error::last_os_error()); + } + } + Ok(()) +} + +fn socket_rcvbuf(fd: RawFd) -> io::Result { + let mut value: c_int = 0; + let mut length = mem::size_of::() as u32; + let result = unsafe { + getsockopt( + fd, + SOL_SOCKET, + SO_RCVBUF, + (&mut value as *mut c_int).cast::(), + &mut length, + ) + }; + if result < 0 { + Err(io::Error::last_os_error()) + } else { + Ok(value) + } +} + +fn socket_inode(fd: RawFd) -> Option { + let link = fs::read_link(format!("/proc/self/fd/{fd}")).ok()?; + let text = link.to_string_lossy(); + let start = text.find('[')? + 1; + let end = text[start..].find(']')? + start; + Some(text[start..end].to_string()) +} + +fn udp_socket_drops(inode: &str) -> Option { + for path in ["/proc/net/udp", "/proc/net/udp6"] { + let Ok(text) = fs::read_to_string(path) else { + continue; + }; + for line in text.lines().skip(1) { + let fields: Vec<&str> = line.split_whitespace().collect(); + if fields.get(9).copied() != Some(inode) { + continue; + } + return fields.last()?.parse().ok(); + } + } + None +} + +#[allow(clippy::too_many_arguments)] +fn send_telemetry( + socket: &UnixDatagram, + cfg: &Config, + stats: &SharedStats, + interval: TrafficCounters, + interval_ms: u64, + rcvbuf_bytes: c_int, + kernel_udp_drops: u64, + kernel_udp_drops_interval: u64, + queue_drops_interval: u64, + truncated_interval: u64, + rx_datagrams_interval: u64, + rx_bytes_interval: u64, + queue_capacity_batches: usize, + queue_capacity_bytes: usize, + uptime: Duration, +) { + let ts_ms = unix_ms(); + let traffic = stats.traffic(); + let queued_batches = stats.queued_batches.load(Ordering::Relaxed); + let queue_high_water = stats.queue_high_water_batches.load(Ordering::Relaxed); + let rx_alive = stats.rx_alive.load(Ordering::Acquire); + let worker_alive = stats.worker_alive.load(Ordering::Acquire); + let ready = rx_alive && worker_alive && !stats.fatal.load(Ordering::Acquire); + let received = stats.tzsp_datagrams.load(Ordering::Relaxed); + let total_loss_interval = kernel_udp_drops_interval + .saturating_add(queue_drops_interval) + .saturating_add(truncated_interval); + let denominator = rx_datagrams_interval.saturating_add(kernel_udp_drops_interval); + let retained = denominator.saturating_sub(total_loss_interval); + let capture_efficiency = if denominator == 0 { + 100.0 + } else { + (retained as f64 * 100.0 / denominator as f64).clamp(0.0, 100.0) + }; + + let payload = format!( + concat!( + "{{\"type\":\"tzsp_sample\",\"engine\":\"rust\",\"version\":2,", + "\"pid\":{},\"ready\":{},\"rx_thread_alive\":{},\"worker_thread_alive\":{},", + "\"ts_ms\":{},\"interval_ms\":{},\"last_packet_ms\":{},\"uptime_ms\":{},", + "\"rcvbuf_bytes\":{},\"kernel_udp_drops\":{},\"kernel_udp_drops_interval\":{},", + "\"batch_size\":{},\"datagram_bytes\":{},\"queue_capacity_batches\":{},", + "\"queue_capacity_bytes\":{},\"queue_depth_batches\":{},\"queue_high_water_batches\":{},", + "\"queue_dropped_datagrams\":{},\"queue_drops_interval\":{},", + "\"truncated_datagrams\":{},\"truncated_interval\":{},", + "\"capture_efficiency_pct\":{:.3},", + "\"rx_datagrams_interval\":{},\"rx_bytes_interval\":{},", + "\"tzsp_datagrams\":{},\"tzsp_rx_bytes\":{},\"rx_batches\":{},", + "\"tzsp_decode_errors\":{},\"tzsp_unsupported\":{},", + "\"frames_injected\":{},\"inject_errors\":{},", + "\"bytes_total\":{},\"bytes_in\":{},\"bytes_out\":{},", + "\"bytes_internal\":{},\"bytes_external\":{},", + "\"packets_total\":{},\"packets_in\":{},\"packets_out\":{},", + "\"packets_internal\":{},\"packets_external\":{},", + "\"traffic_counters\":{{", + "\"bytes_total\":{},\"bytes_in\":{},\"bytes_out\":{},", + "\"bytes_internal\":{},\"bytes_external\":{},", + "\"packets_total\":{},\"packets_in\":{},\"packets_out\":{},", + "\"packets_internal\":{},\"packets_external\":{}", + "}}}}" + ), + process::id(), + ready, + rx_alive, + worker_alive, + ts_ms, + interval_ms, + stats.last_packet_ms.load(Ordering::Relaxed), + uptime.as_millis(), + rcvbuf_bytes.max(0), + kernel_udp_drops, + kernel_udp_drops_interval, + cfg.batch_size, + cfg.datagram_bytes, + queue_capacity_batches, + queue_capacity_bytes, + queued_batches, + queue_high_water, + stats.queue_dropped_datagrams.load(Ordering::Relaxed), + queue_drops_interval, + stats.truncated_datagrams.load(Ordering::Relaxed), + truncated_interval, + capture_efficiency, + rx_datagrams_interval, + rx_bytes_interval, + received, + stats.tzsp_rx_bytes.load(Ordering::Relaxed), + stats.rx_batches.load(Ordering::Relaxed), + stats.tzsp_decode_errors.load(Ordering::Relaxed), + stats.tzsp_unsupported.load(Ordering::Relaxed), + stats.frames_injected.load(Ordering::Relaxed), + stats.inject_errors.load(Ordering::Relaxed), + interval.bytes_total, + interval.bytes_in, + interval.bytes_out, + interval.bytes_internal, + interval.bytes_external, + interval.packets_total, + interval.packets_in, + interval.packets_out, + interval.packets_internal, + interval.packets_external, + traffic.bytes_total, + traffic.bytes_in, + traffic.bytes_out, + traffic.bytes_internal, + traffic.bytes_external, + traffic.packets_total, + traffic.packets_in, + traffic.packets_out, + traffic.packets_internal, + traffic.packets_external, + ); + if !Path::new(&cfg.telemetry_socket).exists() { + return; + } + let _ = socket.send_to(payload.as_bytes(), &cfg.telemetry_socket); +} + +fn unix_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .min(u64::MAX as u128) as u64 +} + +fn round_up(value: usize, multiple: usize) -> usize { + if multiple == 0 { + value + } else { + value.saturating_add(multiple - 1) / multiple * multiple + } +} + +fn pool_batches(queue_mb: usize, batch_size: usize, datagram_bytes: usize) -> usize { + let batch_bytes = batch_size.saturating_mul(datagram_bytes).max(1); + let target = queue_mb.saturating_mul(1024 * 1024); + let queue_batches = target.saturating_add(batch_bytes - 1) / batch_bytes; + queue_batches + .saturating_add(2) + .clamp(MIN_POOL_BATCHES, MAX_POOL_BATCHES) +} + +fn env_u64(name: &str, default: u64) -> u64 { + env::var(name) + .ok() + .and_then(|value| value.trim().parse::().ok()) + .unwrap_or(default) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tzsp_decoder_accepts_ethernet_payload() { + let payload = [1u8, 0, 0, 1, 1, 0xaa, 0xbb]; + let (encapsulation, frame) = decode_tzsp(&payload).unwrap(); + assert_eq!(encapsulation, 1); + assert_eq!(frame, &[0xaa, 0xbb]); + } + + #[test] + fn tzsp_decoder_skips_tags() { + let payload = [1u8, 0, 0, 1, 10, 2, 0xaa, 0xbb, 1, 0xcc]; + let (_, frame) = decode_tzsp(&payload).unwrap(); + assert_eq!(frame, &[0xcc]); + } + + #[test] + fn network_direction_matches_python_semantics() { + let networks = vec![Network::parse("192.168.100.0/24").unwrap()]; + let mut frame = vec![0u8; 14 + 20]; + frame[12..14].copy_from_slice(&0x0800u16.to_be_bytes()); + frame[14] = 0x45; + frame[26..30].copy_from_slice(&[8, 8, 8, 8]); + frame[30..34].copy_from_slice(&[192, 168, 100, 10]); + assert!(matches!(frame_direction(&frame, &networks), Direction::Inbound)); + } + + #[test] + fn queue_pool_reserves_requested_userspace_buffer() { + let batches = pool_batches(64, 256, 11_264); + let usable = batches.saturating_sub(2) * 256 * 11_264; + assert!(usable >= 64 * 1024 * 1024); + assert!(batches >= MIN_POOL_BATCHES); + } + + #[test] + fn traffic_delta_is_saturating() { + let current = TrafficCounters { bytes_total: 100, packets_total: 10, ..Default::default() }; + let previous = TrafficCounters { bytes_total: 75, packets_total: 8, ..Default::default() }; + let delta = current.delta(previous); + assert_eq!(delta.bytes_total, 25); + assert_eq!(delta.packets_total, 2); + } +} diff --git a/scripts/configure-routeros-tzsp-hybrid.sh b/scripts/configure-routeros-tzsp-hybrid.sh new file mode 100755 index 0000000..6c2decb --- /dev/null +++ b/scripts/configure-routeros-tzsp-hybrid.sh @@ -0,0 +1,165 @@ +#!/bin/sh +set -eu + +cd "$(dirname "$0")/.." +CONFIG_FILE="${DEPLOY_CONFIG:-deploy-routeros.env}" +if [ -f "$CONFIG_FILE" ]; then + # shellcheck disable=SC1090 + case "$CONFIG_FILE" in + /*) . "$CONFIG_FILE" ;; + *) . "./$CONFIG_FILE" ;; + esac +fi + +: "${ROUTER_HOST:=192.168.88.1}" +: "${ROUTER_USER:=admin}" +: "${ROUTER_PORT:=22}" +: "${ROUTER_IDENTITY_FILE:=}" +: "${ROUTER_SCP_DIR:=/}" +: "${CONTAINER_IP:=172.31.255.2/30}" +: "${TZSP_PORT:=37008}" +: "${CONFIGURE_IPV4_MANGLE:=true}" +: "${CONFIGURE_L2_SNIFFER:=true}" +: "${START_L2_SNIFFER:=true}" +: "${TZSP_L2_INTERFACE:=}" +: "${TZSP_L2_MAC_PROTOCOL:=!ip}" +TZSP_L2_FILTER_INTERFACE="${TZSP_L2_INTERFACE:-all}" +: "${KEEP_REMOTE_RSC:=false}" +: "${DRY_RUN:=false}" + +case "$ROUTER_PORT" in *[!0-9]*|'') echo "ROUTER_PORT must be numeric" >&2; exit 2 ;; esac +case "$TZSP_PORT" in *[!0-9]*|'') echo "TZSP_PORT must be numeric" >&2; exit 2 ;; esac +for bool_pair in \ + "CONFIGURE_IPV4_MANGLE=$CONFIGURE_IPV4_MANGLE" \ + "CONFIGURE_L2_SNIFFER=$CONFIGURE_L2_SNIFFER" \ + "START_L2_SNIFFER=$START_L2_SNIFFER" \ + "DRY_RUN=$DRY_RUN" +do + case "${bool_pair#*=}" in + true|false) ;; + *) echo "${bool_pair%%=*} must be true or false" >&2; exit 2 ;; + esac +done + +check_ros_value() { + label="$1" + value="$2" + case "$value" in + *'"'*|*'\\'*|*'$'*|*';'*|*'`'*) + echo "$label contains a character not supported by this helper: $value" >&2 + exit 3 + ;; + esac +} +check_ros_value CONTAINER_IP "$CONTAINER_IP" +check_ros_value TZSP_L2_INTERFACE "$TZSP_L2_INTERFACE" +check_ros_value TZSP_L2_MAC_PROTOCOL "$TZSP_L2_MAC_PROTOCOL" + +TARGET="${ROUTER_USER}@${ROUTER_HOST}" +ssh_run() { + if [ -n "$ROUTER_IDENTITY_FILE" ]; then + ssh -i "$ROUTER_IDENTITY_FILE" -p "$ROUTER_PORT" "$TARGET" "$1" + else + ssh -p "$ROUTER_PORT" "$TARGET" "$1" + fi +} +scp_put() { + src="$1" + dst="$2" + if [ -n "$ROUTER_IDENTITY_FILE" ]; then + scp -i "$ROUTER_IDENTITY_FILE" -P "$ROUTER_PORT" "$src" "${TARGET}:$dst" + else + scp -P "$ROUTER_PORT" "$src" "${TARGET}:$dst" + fi +} + +CONTAINER_IP_ONLY="${CONTAINER_IP%/*}" +ID="$(date -u +%Y%m%d%H%M%S)" +mkdir -p build +LOCAL_RSC="build/configure-tzsp-hybrid-${ID}.rsc" +REMOTE_NAME="configure-tzsp-hybrid-${ID}.rsc" +REMOTE_PATH="${ROUTER_SCP_DIR%/}/${REMOTE_NAME}" +[ "$ROUTER_SCP_DIR" = "/" ] && REMOTE_PATH="/${REMOTE_NAME}" + +cat > "$LOCAL_RSC" <> "$LOCAL_RSC" < 0) do={ + /ip/firewall/mangle/add chain=forward action=sniff-tzsp sniff-target="${CONTAINER_IP_ONLY}" sniff-target-port=${TZSP_PORT} comment="MikroSuricata TZSP IPv4" place-before=0 +} else={ + /ip/firewall/mangle/add chain=forward action=sniff-tzsp sniff-target="${CONTAINER_IP_ONLY}" sniff-target-port=${TZSP_PORT} comment="MikroSuricata TZSP IPv4" +} +RSC +fi + +if [ "$CONFIGURE_L2_SNIFFER" = "true" ]; then + cat >> "$LOCAL_RSC" <> "$LOCAL_RSC" <<'RSC' +/tool/sniffer/start +RSC + fi +fi + +cat >> "$LOCAL_RSC" <<'RSC' + +/ip/firewall/mangle/print stats where comment="MikroSuricata TZSP IPv4" +/tool/sniffer/print +RSC + +if [ "$DRY_RUN" = "true" ]; then + cat "$LOCAL_RSC" + echo "[capture] dry run only; RouterOS was not modified" + exit 0 +fi + +command -v ssh >/dev/null 2>&1 || { echo "ssh is required" >&2; exit 2; } +command -v scp >/dev/null 2>&1 || { echo "scp is required" >&2; exit 2; } + +echo "[capture] preflight ${TARGET}" +ssh_run '/ip/firewall/mangle/print' >/dev/null +if [ "$CONFIGURE_L2_SNIFFER" = "true" ]; then + ssh_run '/tool/sniffer/print' >/dev/null +fi + +echo "[capture] uploading $REMOTE_PATH" +scp_put "$LOCAL_RSC" "$REMOTE_PATH" +echo "[capture] applying hybrid capture" +ssh_run "/import file-name=\"${REMOTE_NAME}\"" + +if [ "$KEEP_REMOTE_RSC" != "true" ]; then + ssh_run "/file/remove [find where name=\"${REMOTE_NAME}\"]" || true +fi + +echo "[capture] done" +echo "[capture] IPv4: mangle sniff-tzsp -> ${CONTAINER_IP_ONLY}:${TZSP_PORT}" +if [ "$CONFIGURE_L2_SNIFFER" = "true" ]; then + if [ -n "$TZSP_L2_INTERFACE" ]; then + echo "[capture] non-IPv4: Packet Sniffer on ${TZSP_L2_INTERFACE} -> ${CONTAINER_IP_ONLY}:${TZSP_PORT}" + else + echo "[capture] non-IPv4: Packet Sniffer on all interfaces -> ${CONTAINER_IP_ONLY}:${TZSP_PORT}" + fi +fi diff --git a/scripts/deploy-routeros.sh b/scripts/deploy-routeros.sh index 5b2ff66..9081c8d 100755 --- a/scripts/deploy-routeros.sh +++ b/scripts/deploy-routeros.sh @@ -21,6 +21,14 @@ esac CONTAINER_NAME="suricata_${VERSION}" ROOT_DIR="/containers/${CONTAINER_NAME}/root" +# Backward-compatible aliases for pre-0.11.2 deploy-routeros.env files. +if [ "${CONFIGURE_TZSP_CAPTURE+x}" != x ] && [ "${CONFIGURE_SNIFFER+x}" = x ]; then + CONFIGURE_TZSP_CAPTURE="$CONFIGURE_SNIFFER" +fi +if [ "${START_L2_SNIFFER+x}" != x ] && [ "${START_SNIFFER+x}" = x ]; then + START_L2_SNIFFER="$START_SNIFFER" +fi + : "${ROUTER_HOST:=192.168.88.1}" : "${ROUTER_USER:=admin}" : "${ROUTER_PORT:=22}" @@ -32,14 +40,22 @@ ROOT_DIR="/containers/${CONTAINER_NAME}/root" : "${CONTAINER_SUBNET:=172.31.255.0/30}" : "${CONTAINER_BRIDGE:=br-ids}" : "${CONTAINER_VETH:=veth-ids}" -: "${VLAN_ID:=100}" : "${TZSP_PORT:=37008}" -: "${CONFIGURE_SNIFFER:=true}" -: "${START_SNIFFER:=true}" +: "${CONFIGURE_TZSP_CAPTURE:=true}" +: "${CONFIGURE_IPV4_MANGLE:=true}" +: "${CONFIGURE_L2_SNIFFER:=true}" +: "${START_L2_SNIFFER:=true}" +: "${TZSP_L2_INTERFACE:=}" +: "${TZSP_L2_MAC_PROTOCOL:=!ip}" +TZSP_L2_FILTER_INTERFACE="${TZSP_L2_INTERFACE:-all}" +: "${TZSP_RCVBUF_BYTES:=33554432}" +: "${TZSP_BATCH_SIZE:=256}" +: "${TZSP_QUEUE_MB:=64}" +: "${TZSP_DATAGRAM_BYTES:=12288}" : "${SURICATA_HOME_NET:=[192.168.0.0/16,10.0.0.0/8,172.16.0.0/12]}" : "${MONITORED_NETWORKS:=192.168.0.0/16,10.0.0.0/8,172.16.0.0/12}" : "${SURICATA_LOG_MAX_MB:=512}" -: "${FORENSIC_PCAP_MODE:=blocks}" +: "${FORENSIC_PCAP_MODE:=alerts}" : "${FORENSIC_PCAP_WINDOW_SECONDS:=60}" : "${FORENSIC_PCAP_MEMORY_MB:=64}" : "${FORENSIC_PCAP_MAX_FILES:=32}" @@ -126,9 +142,6 @@ need scp case "$ROUTER_PORT" in *[!0-9]*|'') echo "ROUTER_PORT must be numeric" >&2; exit 2 ;; esac -case "$VLAN_ID" in - *[!0-9]*|'') echo "VLAN_ID must be numeric" >&2; exit 2 ;; -esac case "$TZSP_PORT" in *[!0-9]*|'') echo "TZSP_PORT must be numeric" >&2; exit 2 ;; esac @@ -136,6 +149,17 @@ case "$FORENSIC_PCAP_MODE" in blocks|alerts|all|off) ;; *) echo "FORENSIC_PCAP_MODE must be one of: blocks, alerts, all, off" >&2; exit 2 ;; esac +for bool_pair in \ + "CONFIGURE_TZSP_CAPTURE=$CONFIGURE_TZSP_CAPTURE" \ + "CONFIGURE_IPV4_MANGLE=$CONFIGURE_IPV4_MANGLE" \ + "CONFIGURE_L2_SNIFFER=$CONFIGURE_L2_SNIFFER" \ + "START_L2_SNIFFER=$START_L2_SNIFFER" +do + case "${bool_pair#*=}" in + true|false) ;; + *) echo "${bool_pair%%=*} must be true or false" >&2; exit 2 ;; + esac +done if { [ -n "$METRICS_BASIC_AUTH_USERNAME" ] && [ -z "$METRICS_BASIC_AUTH_PASSWORD" ]; } || \ { [ -z "$METRICS_BASIC_AUTH_USERNAME" ] && [ -n "$METRICS_BASIC_AUTH_PASSWORD" ]; }; then echo "METRICS_BASIC_AUTH_USERNAME and METRICS_BASIC_AUTH_PASSWORD must both be set or both be empty" >&2 @@ -143,6 +167,10 @@ if { [ -n "$METRICS_BASIC_AUTH_USERNAME" ] && [ -z "$METRICS_BASIC_AUTH_PASSWORD fi for numeric_pair in \ "REDIS_PORT=$REDIS_PORT" \ + "TZSP_RCVBUF_BYTES=$TZSP_RCVBUF_BYTES" \ + "TZSP_BATCH_SIZE=$TZSP_BATCH_SIZE" \ + "TZSP_QUEUE_MB=$TZSP_QUEUE_MB" \ + "TZSP_DATAGRAM_BYTES=$TZSP_DATAGRAM_BYTES" \ "SURICATA_LOG_MAX_MB=$SURICATA_LOG_MAX_MB" \ "FORENSIC_PCAP_WINDOW_SECONDS=$FORENSIC_PCAP_WINDOW_SECONDS" \ "FORENSIC_PCAP_MEMORY_MB=$FORENSIC_PCAP_MEMORY_MB" \ @@ -193,6 +221,8 @@ for pair in \ "CONTAINER_SUBNET=$CONTAINER_SUBNET" \ "CONTAINER_BRIDGE=$CONTAINER_BRIDGE" \ "CONTAINER_VETH=$CONTAINER_VETH" \ + "TZSP_L2_INTERFACE=$TZSP_L2_INTERFACE" \ + "TZSP_L2_MAC_PROTOCOL=$TZSP_L2_MAC_PROTOCOL" \ "SURICATA_HOME_NET=$SURICATA_HOME_NET" \ "MONITORED_NETWORKS=$MONITORED_NETWORKS" \ "BLOCK_TIMEOUT=$BLOCK_TIMEOUT" \ @@ -248,10 +278,18 @@ if ! ssh_run "/file/print without-paging where name=\"${IMAGE_TAR_ROS}\"" | grep echo "Upload it first with scripts/upload-routeros-image.sh." >&2 exit 5 fi -if [ "$CONFIGURE_SNIFFER" = "true" ]; then - if ! ssh_run '/tool/sniffer/print' >/dev/null; then - echo "RouterOS sniffer is unavailable." >&2 - exit 4 +if [ "$CONFIGURE_TZSP_CAPTURE" = "true" ]; then + if [ "$CONFIGURE_IPV4_MANGLE" = "true" ]; then + if ! ssh_run '/ip/firewall/mangle/print' >/dev/null; then + echo "RouterOS IPv4 mangle is unavailable." >&2 + exit 4 + fi + fi + if [ "$CONFIGURE_L2_SNIFFER" = "true" ]; then + if ! ssh_run '/tool/sniffer/print' >/dev/null; then + echo "RouterOS sniffer is unavailable." >&2 + exit 4 + fi fi fi @@ -298,6 +336,10 @@ cat > "$LOCAL_RSC" <> "$LOCAL_RSC" <> "$LOCAL_RSC" <> "$LOCAL_RSC" < 0) do={ + /ip/firewall/mangle/add chain=forward action=sniff-tzsp sniff-target="${CONTAINER_IP_ONLY}" sniff-target-port=${TZSP_PORT} comment="MikroSuricata TZSP IPv4" place-before=0 +} else={ + /ip/firewall/mangle/add chain=forward action=sniff-tzsp sniff-target="${CONTAINER_IP_ONLY}" sniff-target-port=${TZSP_PORT} comment="MikroSuricata TZSP IPv4" +} RSC - if [ "$START_SNIFFER" = "true" ]; then - cat >> "$LOCAL_RSC" <<'RSC' + fi + + if [ "$CONFIGURE_L2_SNIFFER" = "true" ]; then + cat >> "$LOCAL_RSC" <> "$LOCAL_RSC" <<'RSC' /tool/sniffer/start RSC + fi fi fi diff --git a/scripts/entrypoint.sh b/scripts/entrypoint.sh index b7d58c7..cdd15a5 100755 --- a/scripts/entrypoint.sh +++ b/scripts/entrypoint.sh @@ -11,7 +11,8 @@ mkdir -p \ "$PERSIST_LOG_DIR" \ "$PERSIST_LIB_DIR/rules" \ "$PERSIST_STATE_DIR" \ - /run/suricata + /run/suricata \ + /run/mikrosuricata if ! id -u suricata >/dev/null 2>&1 || ! getent group suricata >/dev/null 2>&1; then echo "[entrypoint] FATAL: missing suricata user/group in the image; rebuild the image from the current Dockerfile" >&2 diff --git a/scripts/routeros-status.sh b/scripts/routeros-status.sh index 881f93a..1b80c69 100755 --- a/scripts/routeros-status.sh +++ b/scripts/routeros-status.sh @@ -18,5 +18,6 @@ run() { fi } run "/container/print detail where name=\"${CONTAINER_NAME}\"" +run "/ip/firewall/mangle/print stats where comment=\"MikroSuricata TZSP IPv4\"" run "/tool/sniffer/print" run "/log/print without-paging where message~\"suricata|TZSP|IDS\"" diff --git a/tests/test_deploy_script.py b/tests/test_deploy_script.py index 8bddb09..7752766 100644 --- a/tests/test_deploy_script.py +++ b/tests/test_deploy_script.py @@ -68,3 +68,43 @@ def test_routeros_deploy_forwards_ndr_and_persistence_controls(): "METRICS_BASIC_AUTH_PASSWORD", ): assert f"key={key}" in script + + +def test_routeros_deploy_uses_hybrid_tzsp_capture_without_site_bridge_assumption(): + script = (ROOT / "scripts" / "deploy-routeros.sh").read_text() + env_example = (ROOT / "deploy-routeros.env.example").read_text() + + assert 'action=sniff-tzsp' in script + assert 'chain=forward' in script + assert 'comment="MikroSuricata TZSP IPv4"' in script + assert '/ip/firewall/mangle/remove [find where comment="MikroSuricata TZSP IPv4"]' in script + assert 'filter-mac-protocol=${TZSP_L2_MAC_PROTOCOL}' in script + assert 'TZSP_L2_FILTER_INTERFACE="${TZSP_L2_INTERFACE:-all}"' in script + assert 'filter-interface="${TZSP_L2_FILTER_INTERFACE}"' in script + assert 'streaming-server=${CONTAINER_IP_ONLY}:${TZSP_PORT}' in script + assert 'streaming-port=${TZSP_PORT}' in script # compatibility fallback only + assert 'VLAN_ID' not in env_example + assert 'TZSP_L2_INTERFACE=' in env_example + assert 'TZSP_L2_MAC_PROTOCOL=!ip' in env_example + assert 'bridge-trunk' not in script + + +def test_hybrid_capture_migration_helper_supports_dry_run_and_same_owned_rule(): + script = (ROOT / "scripts" / "configure-routeros-tzsp-hybrid.sh").read_text() + assert 'DRY_RUN' in script + assert 'action=sniff-tzsp' in script + assert 'comment="MikroSuricata TZSP IPv4"' in script + assert 'filter-mac-protocol=${TZSP_L2_MAC_PROTOCOL}' in script + assert 'TZSP_L2_FILTER_INTERFACE="${TZSP_L2_INTERFACE:-all}"' in script + assert 'streaming-server=${CONTAINER_IP_ONLY}:${TZSP_PORT}' in script + + +def test_routeros_manual_capture_template_is_hybrid_and_not_vlan_specific(): + hybrid = ROOT / "routeros" / "02-tzsp-hybrid.rsc" + assert hybrid.exists() + assert not (ROOT / "routeros" / "02-sniffer-vlan100.rsc").exists() + text = hybrid.read_text() + assert 'action=sniff-tzsp' in text + assert 'filter-mac-protocol=$l2MacProtocol' in text + assert ':local l2Interface "all"' in text + assert 'filter-vlan=100' not in text diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 68dc8b4..0fa09bd 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -27,6 +27,50 @@ class _Status: class PrometheusMetricsTests(unittest.TestCase): + def test_render_exports_rust_receiver_quality_metrics(self): + stats = RuntimeStats() + metrics = PrometheusMetrics( + stats, + version="0.11.0", + mode="full", + started_at=datetime(2026, 8, 16, 6, 0, tzinfo=timezone.utc), + state_provider=lambda: {"components": {}, "features": {}}, + flow_tracker=_Status( + engine="rust", + rcvbuf_bytes=67_108_864, + batch_size=256, + datagram_bytes=12288, + queue_capacity_batches=24, + queue_capacity_bytes=67_108_864, + queue_depth_batches=2, + queue_high_water_batches=9, + capture_efficiency_pct=99.9, + rx_thread_alive=True, + worker_thread_alive=True, + telemetry_age_ms=125, + process_alive=True, + ready=True, + kernel_udp_drops=7, + queue_dropped_datagrams=11, + truncated_datagrams=0, + telemetry_errors=0, + traffic_counters={ + "bytes_total": 125_000_000, + "packets_total": 80_000, + }, + ), + ) + rendered = metrics.render() + self.assertIn('mikrosuricata_tzsp_receiver_info{engine="rust"} 1', rendered) + self.assertIn("mikrosuricata_tzsp_receiver_rcvbuf_bytes 67108864", rendered) + self.assertIn("mikrosuricata_tzsp_receiver_batch_size 256", rendered) + self.assertIn("mikrosuricata_tzsp_receiver_queue_depth_batches 2", rendered) + self.assertIn("mikrosuricata_tzsp_receiver_capture_efficiency_pct 99.9", rendered) + self.assertIn("mikrosuricata_tzsp_receiver_kernel_udp_drops_total 7", rendered) + self.assertIn("mikrosuricata_tzsp_receiver_queue_dropped_datagrams_total 11", rendered) + self.assertIn('mikrosuricata_traffic_bytes_total{direction="total"} 125000000', rendered) + self.assertNotIn("mikrosuricata_flow_tracker_", rendered) + def test_render_exports_raw_runtime_suricata_and_in_memory_component_state(self): stats = RuntimeStats() stats.inc("tzsp_datagrams", 5) diff --git a/tests/test_rust_dataplane_layout.py b/tests/test_rust_dataplane_layout.py new file mode 100644 index 0000000..e0ac110 --- /dev/null +++ b/tests/test_rust_dataplane_layout.py @@ -0,0 +1,36 @@ +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + + +def test_production_main_uses_rust_receiver_not_python_packet_receiver(): + main = (ROOT / "app" / "main.py").read_text() + assert "from .tzsp_rust import RustTZSPReceiver" in main + assert "from .tzsp import TZSPReceiver" not in main + assert "TapDevice(" not in main + assert "flow_tracker.observe(" not in main + + +def test_docker_builds_and_tests_rust_receiver_before_runtime_copy(): + dockerfile = (ROOT / "Dockerfile").read_text() + assert "AS rust-builder" in dockerfile + assert "cargo test --release" in dockerfile + assert "cargo build --release" in dockerfile + assert "COPY --from=rust-builder" in dockerfile + assert "/usr/local/bin/mikrosuricata-tzsp" in dockerfile + + +def test_rust_receiver_is_dependency_free_and_uses_batched_linux_receive(): + cargo = (ROOT / "rust" / "tzsp-receiver" / "Cargo.toml").read_text() + source = (ROOT / "rust" / "tzsp-receiver" / "src" / "main.rs").read_text() + assert "[dependencies]" not in cargo + assert "recvmmsg(" in source + assert "MSG_WAITFORONE" in source + assert "sync_channel::" in source + assert 'name("tzsp-udp-rx"' in source + assert 'name("tzsp-tap-worker"' in source + assert "queue_dropped_datagrams" in source + assert "SO_RCVBUFFORCE" in source + assert "TUNSETIFF" in source + assert "tap.write(frame)" in source + assert "write_all(frame)" not in source diff --git a/tests/test_tzsp_rust.py b/tests/test_tzsp_rust.py new file mode 100644 index 0000000..5c9dda0 --- /dev/null +++ b/tests/test_tzsp_rust.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import threading +import time +import unittest + +from app.state import RuntimeStats +from app.tzsp_rust import RustTZSPReceiver + + +class RustTZSPReceiverTelemetryTests(unittest.TestCase): + def _receiver(self, sink=None): + return RustTZSPReceiver( + binary="/does/not/start-in-unit-tests", + telemetry_socket="/tmp/unused-tzsp-test.sock", + stats=RuntimeStats(), + stop_event=threading.Event(), + throughput_sink=sink, + ) + + def test_telemetry_sample_drives_live_rate_without_packet_bytes_in_python(self): + written = [] + receiver = self._receiver(written.append) + now_ms = int(time.time() * 1000) + receiver._ingest( + { + "type": "tzsp_sample", + "engine": "rust", + "ready": True, + "ts_ms": now_ms, + "interval_ms": 1000, + "last_packet_ms": now_ms, + "bytes_total": 125_000_000, + "bytes_in": 100_000_000, + "bytes_out": 25_000_000, + "packets_total": 80_000, + "traffic_counters": {"bytes_total": 125_000_000, "packets_total": 80_000}, + "kernel_udp_drops": 3, + "kernel_udp_drops_interval": 0, + "queue_dropped_datagrams": 0, + "queue_drops_interval": 0, + "truncated_datagrams": 0, + "truncated_interval": 0, + "queue_depth_batches": 1, + "queue_capacity_batches": 24, + "queue_capacity_bytes": 64 * 1024 * 1024, + "capture_efficiency_pct": 100.0, + "rx_datagrams_interval": 80_000, + "rx_bytes_interval": 126_000_000, + "rx_thread_alive": True, + "worker_thread_alive": True, + "rcvbuf_bytes": 425_984, + "batch_size": 256, + "datagram_bytes": 12_288, + } + ) + current = receiver.current_throughput(3600) + self.assertEqual(current["current_bps"], 1_000_000_000) + self.assertEqual(current["current_in_bps"], 800_000_000) + self.assertEqual(current["current_out_bps"], 200_000_000) + self.assertEqual(current["current_pps"], 80_000) + self.assertEqual(current["kernel_udp_drops"], 3) + self.assertEqual(current["current_ingress_bps"], 1_008_000_000) + self.assertEqual(current["capture_efficiency_pct"], 100.0) + self.assertEqual(current["queue_fill_pct"], 4.2) + self.assertEqual(current["loss_pps"], 0) + self.assertTrue(current["rx_thread_alive"]) + self.assertTrue(current["worker_thread_alive"]) + self.assertEqual(current["receiver_engine"], "rust") + self.assertEqual(len(written), 1) + self.assertEqual(written[0]["bytes_total"], 125_000_000) + + def test_ingress_and_inspection_are_reported_separately_when_pipeline_is_behind(self): + receiver = self._receiver() + now_ms = int(time.time() * 1000) + receiver._last = { + "ts_ms": now_ms, + "interval_ms": 1000, + "bytes_total": 31_250_000, + "packets_total": 24_000, + "rx_bytes_interval": 125_000_000, + "rx_datagrams_interval": 80_000, + "kernel_udp_drops_interval": 10, + "queue_drops_interval": 90, + "truncated_interval": 0, + "queue_depth_batches": 20, + "queue_capacity_batches": 22, + "capture_efficiency_pct": 99.875, + } + current = receiver.current_throughput(900) + self.assertEqual(current["current_bps"], 250_000_000) + self.assertEqual(current["current_ingress_bps"], 1_000_000_000) + self.assertEqual(current["inspection_ratio_pct"], 25.0) + self.assertEqual(current["queue_fill_pct"], 90.9) + self.assertEqual(current["loss_pps"], 100.0) + + def test_stale_sample_reports_zero_current_rate(self): + receiver = self._receiver() + receiver._last = { + "ts_ms": int(time.time() * 1000) - 10_000, + "interval_ms": 1000, + "bytes_total": 125_000_000, + "packets_total": 80_000, + } + current = receiver.current_throughput(900) + self.assertFalse(current["current_sample_fresh"]) + self.assertEqual(current["current_bps"], 0) + self.assertEqual(current["current_pps"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_webui.py b/tests/test_webui.py index 9498985..b5f0823 100644 --- a/tests/test_webui.py +++ b/tests/test_webui.py @@ -65,7 +65,7 @@ class WebUITests(unittest.TestCase): self.assertIn(f'data-subtab-panel="intelligence:{tab}"', DASHBOARD) self.assertIn("Asset intelligence", DASHBOARD) self.assertIn("Threat intelligence repository", DASHBOARD) - self.assertIn("Forensic PCAP ring", DASHBOARD) + self.assertIn("Forensic PCAP", DASHBOARD) def test_system_has_dedicated_redis_status_panel(self): self.assertIn('

Redis

', DASHBOARD)