ML stages

kernel_ai/ml/polygon.py

@@ -105,7 +105,8 @@ def _synthetic_samples(scenario: str) -> list[ProcSample]:
     if scenario == "scanner":
         return [_sample(pid=9104, comm="nmap", parent_comm="bash", age_sec=5.0)]
     if scenario == "privesc":
-        return [_sample(pid=9105, comm="sudo", parent_comm="bash", ruid=1000, euid=0, age_sec=20.0)]
+        # Non-allowlisted setuid-shaped binary (sudo/fusermount are suppressed).
+        return [_sample(pid=9105, comm="evil-root", parent_comm="bash", ruid=1000, euid=0, age_sec=20.0)]
     if scenario == "lineage_shell":
         return [_sample(pid=9106, comm="sleep", parent_comm="bash", age_sec=4.0)]
     raise KeyError(scenario)

kernel_ai/ml/proc_baseline.py

@@ -2,11 +2,22 @@
 
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass, field
 
 from kernel_ai.ml.baseline import EwmaBaseline
 from kernel_ai.ml.proc_features import PROC_MIN_STD, PROC_POSITION, PROC_SUBSYSTEM, ProcSample
 
+# Common setuid helpers: euid=0 with ruid≠0 is normal, not an attack signal.
+_DEFAULT_PRIVESC_ALLOW = (
+    "sudo,su,fusermount,fusermount3,pkexec,passwd,chage,newgrp,sg,mount,umount"
+)
+
+
+def _privesc_allowlist() -> frozenset[str]:
+    raw = os.getenv("KERNEL_AI_ML_PROC_PRIVESC_ALLOW", _DEFAULT_PRIVESC_ALLOW)
+    return frozenset(x.strip() for x in raw.split(",") if x.strip())
+
 
 @dataclass
 class LineageWhitelist:
@@ -66,6 +77,7 @@ def __init__(
         self.lineage = LineageWhitelist(min_count=lineage_min_count)
         self._last_emit: dict[str, float] = {}
         self._seen_pids: set[int] = set()
+        self._privesc_allow = _privesc_allowlist()
 
     def _cooldown_ok(self, key: str, now: float) -> bool:
         last = self._last_emit.get(key, 0.0)
@@ -133,10 +145,12 @@ def score(self, samples: list[ProcSample], *, now: float) -> list[dict]:
             if len(out) >= self.max_emit:
                 return out
 
-        # --- euid root / ruid non-root ---
+        # --- euid root / ruid non-root (skip known setuid helpers) ---
         for sample in samples:
             if not (sample.euid == 0 and sample.ruid != 0):
                 continue
+            if sample.comm in self._privesc_allow:
+                continue
             ckey = f"privesc:{sample.comm}"
             if not self._cooldown_ok(ckey, now):
                 continue

kernel_ai/views/pages.py

@@ -29,6 +29,9 @@
     ("/linux-devices-subsystem", "linux_devices_subsystem_page", h.linux_devices_subsystem_page),
     ("/devices", "devices_page_legacy", h.devices_page_legacy),
     ("/linux-devices-subsystem.html", "linux_devices_subsystem_html", h.linux_devices_subsystem_html),
+    ("/linux-ebpf-subsystem", "linux_ebpf_subsystem_page", h.linux_ebpf_subsystem_page),
+    ("/ebpf", "ebpf_page_legacy", h.ebpf_page_legacy),
+    ("/linux-ebpf-subsystem.html", "linux_ebpf_subsystem_html", h.linux_ebpf_subsystem_html),
     ("/health", "health_check", h.health_check),
 ]
 

tests/test_ml_stage5.py

@@ -62,13 +62,33 @@ def test_detector_privesc_rule():
         max_emit_per_tick=8,
     )
     # Pre-seed lineage so the edge is already normal.
-    det.lineage.load_counts([("bash", "sudo", 10)])
-    s = _sample(pid=7, comm="sudo", parent_comm="bash", ruid=1000, euid=0, age_sec=30.0)
+    det.lineage.load_counts([("bash", "evil-root", 10)])
+    s = _sample(pid=7, comm="evil-root", parent_comm="bash", ruid=1000, euid=0, age_sec=30.0)
     out = det.score([s], now=50.0)
     assert any(a["type"] == "proc_anomaly:euid_root" for a in out)
 
 
-def test_stage5_default_off():
-    from kernel_ai.ml.config import MLConfig
+def test_detector_privesc_allowlist_skips_sudo():
+    det = ProcBaselineDetector(
+        alpha=0.1,
+        warmup_samples=100,
+        z_warn=4.0,
+        z_crit=7.0,
+        lineage_min_count=1,
+        cooldown_sec=0.0,
+        max_emit_per_tick=8,
+    )
+    det.lineage.load_counts([("bash", "sudo", 10)])
+    s = _sample(pid=8, comm="sudo", parent_comm="bash", ruid=1000, euid=0, age_sec=30.0)
+    out = det.score([s], now=50.0)
+    assert not any(a["type"] == "proc_anomaly:euid_root" for a in out)
+
+
+def test_stage5_default_off(monkeypatch):
+    monkeypatch.delenv("KERNEL_AI_ML_STAGE5", raising=False)
+    import importlib
+
+    import kernel_ai.ml.config as cfg_mod
 
-    assert MLConfig().enable_stage5 is False
+    importlib.reload(cfg_mod)
+    assert cfg_mod.MLConfig().enable_stage5 is False