kernel_dna_isoforest

index.html

@@ -92,7 +92,7 @@ <h2>Linux Kernel Ring 0 Visualization</h2>
     <script src="/static/js/active-connections.js?v=4"></script>
     <script src="/static/js/nginx_files.js?v=6"></script>
     <script src="/static/js/right-semicircle-menu.js?v=48"></script>
-    <script src="/static/js/kernel-dna.js?v=32"></script>
+    <script src="/static/js/kernel-dna.js?v=34"></script>
     <script src="/static/js/network-stack.js?v=33"></script>
     <script src="/static/js/devices-belt.js?v=23"></script>
     <script src="/static/js/crypto-belt.js?v=50"></script>

kernel_ai/api/rest.py

@@ -29,6 +29,7 @@
     ("/execution-context", "get_execution_context", h.get_execution_context, None),
     ("/kernel-dna", "kernel_dna", h.kernel_dna, None),
     ("/ml-anomalies", "ml_anomalies", h.ml_anomalies, None),
+    ("/ml-drift", "ml_drift", h.ml_drift, None),
     ("/crypto-realtime", "crypto_realtime", h.crypto_realtime, None),
     ("/security-realtime", "security_realtime", h.security_realtime, None),
     ("/processes-realtime", "processes_realtime", h.processes_realtime, None),

kernel_ai/http/api.py

@@ -7,6 +7,7 @@
     kernel_data,
     kernel_dna,
     ml_anomalies,
+    ml_drift,
     nginx_files,
     process_kernel_map,
     sentry_test,
@@ -65,6 +66,7 @@
     "kernel_data",
     "kernel_dna",
     "ml_anomalies",
+    "ml_drift",
     "network_stack_realtime",
     "nginx_files",
     "process_kernel_map",

kernel_ai/http/api_handlers/kernel.py

@@ -117,6 +117,47 @@ def _payload():
     return api_json(_payload)
 
 
+def ml_drift():
+    """Latest model-drift verdict + short history for the Kernel DNA UI.
+
+    Read-only: surfaces what the drift monitor / retrain job wrote to the shared
+    store, plus the on-disk model artifact age (a proxy for "model freshness").
+    Degrades to ``available: false`` if the store/model is unreachable.
+    """
+
+    def _payload():
+        import os
+
+        from kernel_ai.ml.config import MLConfig
+        from kernel_ai.ml.store import fetch_drift_status
+
+        try:
+            history = int(request.args.get("history", 48))
+        except (TypeError, ValueError):
+            history = 48
+        history = max(1, min(history, 200))
+
+        cfg = MLConfig()
+        status = fetch_drift_status(cfg.dsn, history=history)
+
+        model_age_sec = None
+        try:
+            mtime = os.path.getmtime(cfg.model_path)
+            model_age_sec = max(0.0, datetime.now().timestamp() - mtime)
+        except OSError:
+            model_age_sec = None
+
+        return {
+            "timestamp": datetime.now().isoformat(),
+            "available": status.get("available", False),
+            "model_age_sec": model_age_sec,
+            "latest": status.get("latest"),
+            "history": status.get("history", []),
+        }
+
+    return api_json(_payload)
+
+
 def sentry_test():
     """Temporary endpoint for manual Sentry verification."""
     if not current_app.config.get("SENTRY_TEST_ENDPOINT_ENABLED", False):

kernel_ai/ml/config.py

@@ -13,6 +13,11 @@
 # Project root (parent of the ``kernel_ai`` package), used for default paths.
 _PROJECT_ROOT = Path(__file__).resolve().parents[2]
 
+# All ML runtime data (model artifacts, MLflow sqlite + artifacts) lives in one
+# directory owned by the service user (www-data), so the worker and the retrain
+# job can both read/write it. Override with KERNEL_AI_ML_DATA_DIR.
+_DATA_DIR = Path(os.getenv("KERNEL_AI_ML_DATA_DIR", str(_PROJECT_ROOT / "mldata")))
+
 
 def _env_float(name: str, default: float) -> float:
     try:
@@ -70,13 +75,18 @@ class MLConfig:
     # Saved model artifact the worker loads (decoupled from MLflow).
     model_path: str = os.getenv(
         "KERNEL_AI_ML_MODEL_PATH",
-        str(_PROJECT_ROOT / "models" / "isoforest_latest.joblib"),
+        str(_DATA_DIR / "isoforest_latest.joblib"),
     )
     # MLflow tracking store. sqlite gives both tracking + a model registry while
     # staying file-light (no always-on server). Override on PROD if desired.
     mlflow_uri: str = os.getenv(
         "KERNEL_AI_MLFLOW_URI",
-        f"sqlite:///{_PROJECT_ROOT / 'mlflow.db'}",
+        f"sqlite:///{_DATA_DIR / 'mlflow.db'}",
+    )
+    # Artifact root for MLflow runs (must be writable by the service user).
+    mlflow_artifact_uri: str = os.getenv(
+        "KERNEL_AI_MLFLOW_ARTIFACT_URI",
+        f"file:{_DATA_DIR / 'mlruns'}",
     )
     mlflow_experiment: str = os.getenv("KERNEL_AI_MLFLOW_EXPERIMENT", "kernel_dna_anomaly")
     mlflow_model_name: str = os.getenv("KERNEL_AI_MLFLOW_MODEL", "kernel_dna_isoforest")
@@ -90,6 +100,9 @@ class MLConfig:
     # --- Stage 3 (drift + auto-retrain) ---
     # Window of recent snapshots used to measure drift (minutes).
     drift_window_min: int = _env_int("KERNEL_AI_ML_DRIFT_WINDOW_MIN", 30)
+    # Minimum recent samples before drift verdicts are trusted (avoid declaring
+    # drift off one or two noisy snapshots right after a worker restart).
+    drift_min_recent: int = _env_int("KERNEL_AI_ML_DRIFT_MIN_RECENT", 20)
     # Drift trips when the live flag rate exceeds expected (contamination) by
     # this multiple, or when the mean per-feature distribution shift (in train
     # std units) exceeds the z threshold.
@@ -103,6 +116,36 @@ class MLConfig:
     retrain_min_flag_rate: float = _env_float("KERNEL_AI_ML_RETRAIN_MIN_FLAG", 0.001)
     retrain_max_flag_rate: float = _env_float("KERNEL_AI_ML_RETRAIN_MAX_FLAG", 0.30)
 
+    # --- Stage 4 (syscall sequence model: STIDE n-grams) ---
+    # Detects anomalous *sequences* of syscalls rather than per-feature spikes.
+    # Built on sampled /proc/<pid>/syscall traces (L0), so it is a coarse but
+    # zero-dependency HIDS; eBPF/auditd tracing (L2) is the future upgrade.
+    enable_stage4: bool = os.getenv("KERNEL_AI_ML_STAGE4", "true").lower() == "true"
+    seq_model_path: str = os.getenv(
+        "KERNEL_AI_ML_SEQ_MODEL_PATH",
+        str(_DATA_DIR / "stide_latest.joblib"),
+    )
+    seq_n: int = _env_int("KERNEL_AI_ML_SEQ_N", 3)                  # n-gram size
+    seq_max_pids: int = _env_int("KERNEL_AI_ML_SEQ_MAX_PIDS", 512)  # pids sampled/tick
+    seq_window: int = _env_int("KERNEL_AI_ML_SEQ_WINDOW", 400)      # rolling n-grams scored
+    seq_min_window: int = _env_int("KERNEL_AI_ML_SEQ_MIN_WINDOW", 120)  # before scoring
+    # 2s tick sampling only catches processes *parked* in a syscall. A short burst
+    # of sub-samples per tick captures real syscall transitions (better sequences).
+    seq_subsamples: int = _env_int("KERNEL_AI_ML_SEQ_SUBSAMPLES", 4)
+    seq_subsample_gap_ms: int = _env_int("KERNEL_AI_ML_SEQ_SUBSAMPLE_GAP_MS", 40)
+    # Minimum distinct n-grams required before a STIDE profile is built (a profile
+    # learned from too little data would flag almost everything).
+    seq_min_vocab: int = _env_int("KERNEL_AI_ML_SEQ_MIN_VOCAB", 50)
+    # Window mismatch fraction (unseen n-grams) that counts as a sequence anomaly.
+    seq_mismatch_warn: float = _env_float("KERNEL_AI_ML_SEQ_MISMATCH_WARN", 0.30)
+    seq_mismatch_crit: float = _env_float("KERNEL_AI_ML_SEQ_MISMATCH_CRIT", 0.55)
+    seq_cooldown_sec: float = _env_float("KERNEL_AI_ML_SEQ_COOLDOWN_SEC", 30.0)
+    # Flush newly observed n-grams to the store every N seconds (profile growth).
+    seq_flush_sec: float = _env_float("KERNEL_AI_ML_SEQ_FLUSH_SEC", 30.0)
+    # Training keeps n-grams seen at least this many times (frequency-based poison
+    # guard: a one-off attack sequence never enters the "normal" profile).
+    seq_min_ngram_count: int = _env_int("KERNEL_AI_ML_SEQ_MIN_COUNT", 3)
+
     @property
     def alpha(self) -> float:
         return 2.0 / (max(2, self.baseline_window) + 1.0)

kernel_ai/ml/drift.py

@@ -72,11 +72,13 @@ def compute_drift(cfg: MLConfig | None = None, *, persist: bool = True) -> dict:
     n = len(recent)
     contamination = float(model.meta.get("contamination", cfg.if_contamination))
 
-    if n == 0:
+    # Too few recent samples (e.g. right after a worker restart) -> a single
+    # noisy point must not be allowed to declare "drift".
+    if n < cfg.drift_min_recent:
         result = {
-            "available": True, "n_recent": 0, "flag_rate": 0.0,
+            "available": True, "n_recent": n, "flag_rate": 0.0,
             "expected_rate": contamination, "feature_drift": 0.0, "drifted": False,
-            "detail": {"reason": "no_recent_data"},
+            "detail": {"reason": "insufficient_recent_data", "min_recent": cfg.drift_min_recent},
         }
         if persist:
             insert_drift(cfg.dsn, result)

kernel_ai/ml/sequence.py

@@ -0,0 +1,206 @@
+"""Stage 4: syscall *sequence* anomaly detection (STIDE-style n-grams).
+
+Stages 1-3 score the system on aggregate *features* (rates, pressures). They are
+blind to the **order** of operations. A classic host-IDS insight (Forrest et al.,
+"A Sense of Self for Unix Processes") is that normal programs emit a small, stable
+vocabulary of short syscall *sequences*; intrusions produce sequences never seen
+during normal operation. STIDE (Sequence TIme-Delay Embedding) learns the set of
+normal n-grams and flags windows with a high fraction of unseen n-grams.
+
+Data source (honesty note): we don't have a continuous syscall tracer here, so we
+*sample* ``/proc/<pid>/syscall`` each worker tick (the syscall a task is currently
+in). This is a coarse L0 signal -- it misses fast transitions -- but it needs zero
+privileges/deps and demonstrates the sequence approach end to end. Swapping in an
+eBPF/auditd tracer (L2) later only changes the sampler, not the model.
+
+Components:
+    SyscallSampler  - read current syscall per pid from procfs
+    NgramTracker    - per-pid rolling deques -> stream of syscall n-grams
+    StideModel      - set of "normal" n-grams + window mismatch scoring
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from collections import deque
+from dataclasses import dataclass, field
+
+from kernel_ai.services.kernel_maps import SYSCALL_NAMES
+
+logger = logging.getLogger("kernel_ai.ml.sequence")
+
+# Separator for serialising an n-gram tuple into a stable string key.
+_SEP = "|"
+
+
+class SyscallSampler:
+    """Sample the current syscall of running processes from procfs."""
+
+    def __init__(self, max_pids: int = 160) -> None:
+        self.max_pids = max_pids
+
+    def sample(self) -> dict[int, str]:
+        """Return ``{pid: syscall_name}`` for tasks currently in a syscall."""
+        out: dict[int, str] = {}
+        try:
+            pids = sorted((int(d) for d in os.listdir("/proc") if d.isdigit()))
+        except OSError:
+            return out
+        for pid in pids[: self.max_pids]:
+            path = f"/proc/{pid}/syscall"
+            try:
+                with open(path, "r", encoding="utf-8", errors="ignore") as fh:
+                    line = fh.read().strip()
+            except (OSError, PermissionError):
+                continue
+            if not line or line == "-1" or line.startswith("running"):
+                continue
+            head = line.split(" ", 1)[0]
+            try:
+                num = int(head)
+            except ValueError:
+                continue
+            if num < 0:
+                continue
+            out[int(pid)] = SYSCALL_NAMES.get(num, f"sys_{num}")
+        return out
+
+
+class NgramTracker:
+    """Maintain per-pid syscall histories and emit n-grams as they complete.
+
+    A new n-gram is produced whenever a pid's rolling history reaches length ``n``.
+    Consecutive identical samples (a task parked in one syscall) naturally form
+    homogeneous n-grams -- those are normal and end up in the profile.
+    """
+
+    def __init__(self, n: int = 3, window: int = 400, max_pids: int = 4096) -> None:
+        self.n = max(2, n)
+        self.window = window
+        self.max_pids = max_pids
+        self._hist: dict[int, deque[str]] = {}
+        # Rolling window of recent n-gram keys used for live scoring.
+        self._recent: deque[str] = deque(maxlen=window)
+        # Counts of every n-gram observed since the last flush (profile growth).
+        self._pending: dict[str, int] = {}
+
+    def update(self, samples: dict[int, str]) -> None:
+        # Drop histories for pids that vanished to bound memory.
+        if len(self._hist) > self.max_pids:
+            for dead in [p for p in self._hist if p not in samples]:
+                self._hist.pop(dead, None)
+
+        for pid, name in samples.items():
+            hist = self._hist.get(pid)
+            if hist is None:
+                hist = deque(maxlen=self.n)
+                self._hist[pid] = hist
+            hist.append(name)
+            if len(hist) == self.n:
+                key = _SEP.join(hist)
+                self._recent.append(key)
+                self._pending[key] = self._pending.get(key, 0) + 1
+
+    def recent(self) -> list[str]:
+        return list(self._recent)
+
+    def drain_pending(self) -> dict[str, int]:
+        """Return + clear n-gram counts accumulated since the last drain."""
+        pending = self._pending
+        self._pending = {}
+        return pending
+
+
+@dataclass
+class StideModel:
+    """A "normal" n-gram vocabulary with window-mismatch scoring."""
+
+    n: int
+    ngrams: set[str] = field(default_factory=set)
+    meta: dict = field(default_factory=dict)
+
+    def score_window(self, window: list[str]) -> tuple[float, int]:
+        """Return ``(mismatch_rate, n_mismatches)`` for a window of n-gram keys.
+
+        mismatch_rate = fraction of n-grams in the window not present in the
+        learned normal vocabulary. High rate -> the system is doing things in an
+        order it never did while the profile was learned.
+        """
+        if not window:
+            return 0.0, 0
+        misses = sum(1 for g in window if g not in self.ngrams)
+        return misses / len(window), misses
+
+    def top_unseen(self, window: list[str], limit: int = 3) -> list[str]:
+        """Most frequent unseen n-grams in the window (for an explainable cause)."""
+        counts: dict[str, int] = {}
+        for g in window:
+            if g not in self.ngrams:
+                counts[g] = counts.get(g, 0) + 1
+        ordered = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)
+        return [g.replace(_SEP, "→") for g, _ in ordered[:limit]]
+
+    def save(self, path: str) -> None:
+        import joblib
+
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        joblib.dump({"n": self.n, "ngrams": self.ngrams, "meta": self.meta}, path)
+
+    @classmethod
+    def load(cls, path: str) -> "StideModel":
+        import joblib
+
+        obj = joblib.load(path)
+        return cls(n=int(obj["n"]), ngrams=set(obj["ngrams"]), meta=dict(obj.get("meta", {})))
+
+
+def build_profile(cfg) -> dict:
+    """(Re)build the STIDE normal profile from accumulated n-gram counts.
+
+    Frequency-based poison guard: only n-grams seen at least ``seq_min_ngram_count``
+    times enter the profile, so a one-off attack sequence never becomes "normal".
+    Returns a small metrics dict; raises SystemExit if there isn't enough data yet.
+    """
+    from kernel_ai.ml.store import fetch_ngram_counts
+
+    counts = fetch_ngram_counts(cfg.dsn, n=cfg.seq_n)
+    total = len(counts)
+    if total < cfg.seq_min_vocab:
+        raise SystemExit(
+            f"Not enough syscall n-grams to build STIDE profile "
+            f"(have {total}, need >={cfg.seq_min_vocab})"
+        )
+
+    kept = {g for g, c in counts.items() if c >= cfg.seq_min_ngram_count}
+    if not kept:
+        raise SystemExit("STIDE profile empty after frequency filter; lower SEQ_MIN_COUNT or collect more data")
+
+    meta = {
+        "stage": 4,
+        "n": cfg.seq_n,
+        "vocab_total": total,
+        "vocab_kept": len(kept),
+        "min_count": cfg.seq_min_ngram_count,
+    }
+    model = StideModel(n=cfg.seq_n, ngrams=kept, meta=meta)
+    model.save(cfg.seq_model_path)
+    logger.info(
+        "saved STIDE profile -> %s (kept %d/%d n-grams, n=%d)",
+        cfg.seq_model_path, len(kept), total, cfg.seq_n,
+    )
+    return meta
+
+
+def main() -> None:
+    import logging as _logging
+
+    from kernel_ai.ml.config import MLConfig
+
+    _logging.basicConfig(level=_logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
+    meta = build_profile(MLConfig())
+    logger.info("STIDE build done: %s", meta)
+
+
+if __name__ == "__main__":
+    main()