feat(hub): netstats snapshot builder + privileged collector + root wrapper (ref #758)

packages/secubox-hub/api/netstats.py

@@ -25,6 +25,7 @@ CATEGORY_MAP = {
     "sbx_doh_detect_v4": "doh", "sbx_doh_detect_v6": "doh",
     "sbx_drop_wafrl": "waf_ratelimit",
     "sbx_drop_input_policy": "input_policy",
+    "sbx_drop_crowdsec": "crowdsec",
 }
 # Categories that count toward "network_drops" (doh is detect-only, excluded).
 DROP_CATEGORIES = {"blacklist", "quarantine", "waf_ratelimit", "input_policy", "crowdsec"}
@@ -183,3 +184,153 @@ def prune(conn: sqlite3.Connection, keep_s: int) -> None:
         if row and row[0] is not None:
             conn.execute(f"DELETE FROM {tbl} WHERE ts < ?", (row[0] - keep_s,))
     conn.commit()
+
+
+# ---------------------------------------------------------------------------
+# Snapshot builder + privileged collector (Task 6)
+# ---------------------------------------------------------------------------
+
+def build_snapshot(conn: sqlite3.Connection, now: int) -> dict:
+    """Latest cumulative per category + instantaneous rates vs the previous
+    sample. network_drops = sum of DROP_CATEGORIES packet counts (doh excluded).
+    """
+    cats: dict[str, dict] = {}
+    # latest cumulative per counter-name → fold into categories
+    rows = conn.execute(
+        "SELECT name, packets, bytes, ts FROM counter_samples "
+        "WHERE ts=(SELECT MAX(ts) FROM counter_samples)"
+    ).fetchall()
+    for name, pk, by, _ts in rows:
+        cat = category_for(name)
+        if cat is None:
+            continue
+        c = cats.setdefault(cat, {"packets": 0, "bytes": 0})
+        c["packets"] += int(pk)
+        c["bytes"] += int(by)
+    ser = query_series(conn, window_s=120, step_s=30)
+    for cat, pts in ser["drops"].items():
+        if pts:
+            cats.setdefault(cat, {"packets": 0, "bytes": 0})["pps"] = pts[-1][1] / 30.0
+
+    ifaces: dict[str, dict] = {}
+    irow = conn.execute(
+        "SELECT iface, rx_bytes, tx_bytes FROM iface_samples "
+        "WHERE ts=(SELECT MAX(ts) FROM iface_samples)"
+    ).fetchall()
+    for iface, rx, tx in irow:
+        ifaces[iface] = {"rx_bytes": int(rx), "tx_bytes": int(tx)}
+    for iface, pts in ser["in_bps"].items():
+        if pts:
+            ifaces.setdefault(iface, {})["rx_bps"] = pts[-1][1]
+    for iface, pts in ser["out_bps"].items():
+        if pts:
+            ifaces.setdefault(iface, {})["tx_bps"] = pts[-1][1]
+
+    network_drops = sum(
+        cats.get(cat, {}).get("packets", 0) for cat in DROP_CATEGORIES
+    )
+    return {
+        "updated": now,
+        "stale": False,
+        "categories": cats,
+        "interfaces": ifaces,
+        "network_drops": int(network_drops),
+    }
+
+
+def read_snapshot() -> dict:
+    """Read the latest snapshot (API path). Flags stale by `updated` age."""
+    try:
+        d = json.loads(SNAP_PATH.read_text())
+    except Exception:
+        return {"updated": 0, "stale": True, "categories": {}, "interfaces": {}, "network_drops": 0}
+    age = max(0, int(time.time()) - int(d.get("updated", 0)))
+    d["stale"] = age > STALE_AFTER_S
+    return d
+
+
+def _run_nft_json(args: list[str]) -> dict:
+    try:
+        r = subprocess.run(["/usr/sbin/nft", "-j", "list"] + args,
+                           capture_output=True, text=True, timeout=5)
+        if r.returncode == 0:
+            return json.loads(r.stdout or "{}")
+    except Exception:
+        pass
+    return {}
+
+
+def _read_nft_counters() -> dict:
+    return parse_nft_counters_json(_run_nft_json(["counters"]))
+
+
+def _read_crowdsec() -> dict:
+    """Best-effort: sum the externally-managed inet crowdsec table counters
+    into a single synthetic counter mapped to the 'crowdsec' category.
+    """
+    data = _run_nft_json(["table", "inet", "crowdsec"])
+    total = 0
+    for item in data.get("nftables", []):
+        c = item.get("counter")
+        if isinstance(c, dict):
+            total += int(c.get("packets", 0) or 0)
+        rule = item.get("rule")
+        if isinstance(rule, dict):
+            for ex in rule.get("expr", []):
+                cc = ex.get("counter")
+                if isinstance(cc, dict):
+                    total += int(cc.get("packets", 0) or 0)
+    if total:
+        # synthetic name so category_for resolves to 'crowdsec'
+        return {"sbx_drop_crowdsec": {"packets": total, "bytes": 0}}
+    return {}
+
+
+def _read_ifaces() -> dict:
+    try:
+        return parse_proc_net_dev(Path("/proc/net/dev").read_text())
+    except Exception:
+        return {}
+
+
+def _open_db(conn: sqlite3.Connection | None):
+    if conn is not None:
+        return conn, False
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    try:
+        DATA_DIR.chmod(0o755)  # our subdir only — NEVER the /var/lib/secubox parent
+    except Exception:
+        pass
+    c = sqlite3.connect(str(DB_PATH))
+    init_db(c)
+    return c, True
+
+
+def collect_once(now: int | None = None, conn: sqlite3.Connection | None = None) -> dict:
+    if now is None:
+        now = int(time.time())
+    c, owns = _open_db(conn)
+    try:
+        counters = dict(_read_nft_counters())
+        counters.update(_read_crowdsec())
+        ifaces = _read_ifaces()
+        insert_sample(c, now, counters, ifaces)
+        prune(c, keep_s=7 * 86400)
+        snap = build_snapshot(c, now)
+        try:
+            SNAP_PATH.write_text(json.dumps(snap))
+            SNAP_PATH.chmod(0o644)
+        except Exception:
+            pass
+        return snap
+    finally:
+        if owns:
+            c.close()
+
+
+def main() -> None:
+    collect_once()
+
+
+if __name__ == "__main__":
+    main()

packages/secubox-hub/sbin/secubox-netstats-collect

@@ -0,0 +1,6 @@
+#!/bin/sh
+# SPDX-License-Identifier: LicenseRef-CMSD-1.0
+# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
+# #758 — root oneshot: sample nft named counters + /proc/net/dev into
+# /var/lib/secubox/hub/netstats.{db,json}. Runs as root via systemd timer.
+exec /usr/bin/python3 -c "import sys; sys.path.insert(0, '/usr/lib/secubox/hub/api'); import netstats; netstats.main()"

packages/secubox-hub/tests/test_netstats_snapshot.py

@@ -0,0 +1,54 @@
+# SPDX-License-Identifier: LicenseRef-CMSD-1.0
+"""Snapshot builder + collect_once with monkeypatched privileged sources (#758)."""
+import importlib
+import json
+import sqlite3
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "api"))
+netstats = importlib.import_module("netstats")
+
+
+def test_build_snapshot_network_drops_excludes_doh():
+    c = sqlite3.connect(":memory:")
+    netstats.init_db(c)
+    counters = {
+        "sbx_drop_blacklist_v4": {"packets": 4, "bytes": 0},
+        "sbx_drop_wafrl": {"packets": 6, "bytes": 0},
+        "sbx_doh_detect_v4": {"packets": 99, "bytes": 0},
+    }
+    netstats.insert_sample(c, 1000, counters, {})
+    snap = netstats.build_snapshot(c, now=1000)
+    assert snap["network_drops"] == 10           # 4 + 6, doh excluded
+    assert snap["categories"]["doh"]["packets"] == 99
+    assert snap["updated"] == 1000
+
+
+def test_collect_once_writes_row_and_snapshot(tmp_path, monkeypatch):
+    db = tmp_path / "netstats.db"
+    snap = tmp_path / "netstats.json"
+    monkeypatch.setattr(netstats, "DB_PATH", db)
+    monkeypatch.setattr(netstats, "SNAP_PATH", snap)
+    monkeypatch.setattr(netstats, "DATA_DIR", tmp_path)
+    monkeypatch.setattr(netstats, "_read_nft_counters",
+                        lambda: {"sbx_drop_wafrl": {"packets": 3, "bytes": 30}})
+    monkeypatch.setattr(netstats, "_read_crowdsec", lambda: {})
+    monkeypatch.setattr(netstats, "_read_ifaces",
+                        lambda: {"eth0": {"rx_bytes": 1, "rx_packets": 1, "tx_bytes": 1, "tx_packets": 1}})
+    out = netstats.collect_once(now=1234)
+    assert out["network_drops"] == 3
+    assert snap.exists()
+    written = json.loads(snap.read_text())
+    assert written["updated"] == 1234
+    # a second tick must not raise (DB reused)
+    netstats.collect_once(now=1264)
+
+
+def test_read_snapshot_marks_stale(tmp_path, monkeypatch):
+    snap = tmp_path / "netstats.json"
+    snap.write_text(json.dumps({"updated": 0, "categories": {}, "interfaces": {}, "network_drops": 0}))
+    monkeypatch.setattr(netstats, "SNAP_PATH", snap)
+    out = netstats.read_snapshot()
+    assert out["stale"] is True