diff --git a/index.html b/index.html
index 4a2ffc0..2abf4be 100755
--- a/index.html
+++ b/index.html
@@ -92,7 +92,7 @@ Linux Kernel Ring 0 Visualization
-
+
diff --git a/kernel_ai/api/rest.py b/kernel_ai/api/rest.py
index 027e267..016fc3b 100644
--- a/kernel_ai/api/rest.py
+++ b/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),
diff --git a/kernel_ai/http/api.py b/kernel_ai/http/api.py
index 6642670..61486a2 100644
--- a/kernel_ai/http/api.py
+++ b/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",
diff --git a/kernel_ai/http/api_handlers/kernel.py b/kernel_ai/http/api_handlers/kernel.py
index f17db72..42de9f0 100644
--- a/kernel_ai/http/api_handlers/kernel.py
+++ b/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):
diff --git a/kernel_ai/ml/config.py b/kernel_ai/ml/config.py
index 0995664..1bf7766 100644
--- a/kernel_ai/ml/config.py
+++ b/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//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)
diff --git a/kernel_ai/ml/drift.py b/kernel_ai/ml/drift.py
index 1c144a3..ee84397 100644
--- a/kernel_ai/ml/drift.py
+++ b/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)
diff --git a/kernel_ai/ml/sequence.py b/kernel_ai/ml/sequence.py
new file mode 100644
index 0000000..ffc539c
--- /dev/null
+++ b/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//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()
diff --git a/kernel_ai/ml/store.py b/kernel_ai/ml/store.py
index c8443af..aad053a 100644
--- a/kernel_ai/ml/store.py
+++ b/kernel_ai/ml/store.py
@@ -62,6 +62,15 @@
detail jsonb
);
CREATE INDEX IF NOT EXISTS ml_drift_ts_idx ON ml_drift (ts);
+
+CREATE TABLE IF NOT EXISTS ml_syscall_ngrams (
+ ngram text PRIMARY KEY,
+ n smallint NOT NULL,
+ count bigint NOT NULL DEFAULT 0,
+ first_seen timestamptz NOT NULL DEFAULT now(),
+ last_seen timestamptz NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS ml_syscall_ngrams_n_idx ON ml_syscall_ngrams (n);
"""
@@ -122,6 +131,22 @@ def insert_anomalies(self, anomalies: list[dict]) -> None:
],
)
+ def upsert_ngram_counts(self, n: int, counts: dict[str, int]) -> None:
+ """Accumulate observed syscall n-gram counts (profile growth)."""
+ if not counts:
+ return
+ with self.conn.cursor() as cur:
+ cur.executemany(
+ """
+ INSERT INTO ml_syscall_ngrams (ngram, n, count)
+ VALUES (%(ngram)s, %(n)s, %(count)s)
+ ON CONFLICT (ngram) DO UPDATE
+ SET count = ml_syscall_ngrams.count + EXCLUDED.count,
+ last_seen = now()
+ """,
+ [{"ngram": g, "n": n, "count": c} for g, c in counts.items()],
+ )
+
def save_baseline(self, rows: list[dict]) -> None:
if not rows:
return
@@ -237,6 +262,46 @@ def insert_drift(dsn: str, record: dict) -> None:
logger.warning("insert_drift failed: %s", exc)
+def fetch_ngram_counts(dsn: str, *, n: int) -> dict[str, int]:
+ """Read accumulated syscall n-gram counts for STIDE profile building."""
+ with connect(dsn) as conn, conn.cursor() as cur:
+ cur.execute("SELECT ngram, count FROM ml_syscall_ngrams WHERE n = %s", (n,))
+ return {row[0]: int(row[1]) for row in cur.fetchall()}
+
+
+def fetch_drift_status(dsn: str, *, history: int = 48) -> dict:
+ """Read the latest drift verdict plus a short history for the API.
+
+ Returns ``{"available": bool, "latest": {...}|None, "history": [...]}``.
+ Never raises on DB trouble (API must degrade gracefully).
+ """
+ try:
+ with connect(dsn) as conn, conn.cursor() as cur:
+ cur.execute(
+ """
+ SELECT ts, flag_rate, expected_rate, feature_drift, n_recent, drifted
+ FROM ml_drift
+ ORDER BY ts DESC
+ LIMIT %s
+ """,
+ (max(1, int(history)),),
+ )
+ cols = [d.name for d in cur.description]
+ rows = []
+ for row in cur.fetchall():
+ rec = dict(zip(cols, row))
+ if rec.get("ts") is not None:
+ rec["ts"] = rec["ts"].isoformat()
+ rows.append(rec)
+ if not rows:
+ return {"available": True, "latest": None, "history": []}
+ # rows are newest-first; history is returned oldest-first for charting.
+ return {"available": True, "latest": rows[0], "history": list(reversed(rows))}
+ except Exception as exc: # noqa: BLE001 - API must degrade gracefully
+ logger.warning("fetch_drift_status failed: %s", exc)
+ return {"available": False, "latest": None, "history": []}
+
+
def fetch_recent_anomalies(dsn: str, *, since_seconds: int = 120, limit: int = 100) -> list[dict]:
"""Read recent anomalies for the API. Never raises on DB trouble."""
try:
diff --git a/kernel_ai/ml/train.py b/kernel_ai/ml/train.py
index 213ea1d..193392b 100644
--- a/kernel_ai/ml/train.py
+++ b/kernel_ai/ml/train.py
@@ -113,11 +113,22 @@ def train(
# MLflow tracking + registry (best-effort: training must not hard-fail if
# MLflow has a hiccup, the artifact is already saved above).
+ #
+ # MLflow's sqlite store unconditionally mkdir's its default artifact root
+ # (``./mlruns``) relative to the *current working directory* at store-init
+ # time. Under systemd the cwd is the (read-only-to-www-data) project root,
+ # so we run the whole MLflow block from inside the writable data dir.
+ data_dir = os.path.dirname(cfg.model_path)
+ prev_cwd = os.getcwd()
try:
+ os.chdir(data_dir)
import mlflow
import mlflow.sklearn
mlflow.set_tracking_uri(cfg.mlflow_uri)
+ # Ensure the experiment stores artifacts in the service-writable data dir.
+ if mlflow.get_experiment_by_name(cfg.mlflow_experiment) is None:
+ mlflow.create_experiment(cfg.mlflow_experiment, artifact_location=cfg.mlflow_artifact_uri)
mlflow.set_experiment(cfg.mlflow_experiment)
with mlflow.start_run() as run:
mlflow.log_params(
@@ -139,6 +150,22 @@ def train(
logger.info("logged MLflow run %s (experiment=%s)", run.info.run_id, cfg.mlflow_experiment)
except Exception as exc: # noqa: BLE001 - tracking is optional
logger.warning("MLflow logging skipped: %s", exc)
+ finally:
+ os.chdir(prev_cwd)
+
+ # Stage 4: rebuild the syscall-sequence (STIDE) profile from the n-grams the
+ # worker has accumulated. Best-effort: a missing/young vocabulary must not
+ # block IsolationForest training (which is already saved above).
+ if cfg.enable_stage4:
+ try:
+ from kernel_ai.ml.sequence import build_profile
+
+ seq_meta = build_profile(cfg)
+ metrics["seq_vocab_kept"] = float(seq_meta.get("vocab_kept", 0))
+ except SystemExit as exc:
+ logger.info("STIDE profile not rebuilt: %s", exc)
+ except Exception as exc: # noqa: BLE001 - sequence profile is optional
+ logger.warning("STIDE profile build failed: %s", exc)
return metrics
diff --git a/kernel_ai/ml/worker.py b/kernel_ai/ml/worker.py
index fd132fa..4fb0388 100644
--- a/kernel_ai/ml/worker.py
+++ b/kernel_ai/ml/worker.py
@@ -92,6 +92,35 @@ def _build_isoforest_anomaly(score: float, scores: dict[str, Score], cfg: MLConf
}
+def _build_sequence_anomaly(mismatch: float, misses: int, window_len: int,
+ top_unseen: list[str], cfg: MLConfig) -> dict:
+ """Build a mutation from a STIDE syscall-sequence verdict (Stage 4).
+
+ Unlike Stages 1-2 (which judge *magnitudes*), this fires when the recent
+ *order* of syscalls contains a high fraction of sequences never seen while
+ the normal profile was learned -- the classic signature of an intrusion.
+ """
+ severity = "high" if mismatch >= cfg.seq_mismatch_crit else "medium"
+ cause = ("; novel: " + ", ".join(top_unseen)) if top_unseen else ""
+ return {
+ "source": "stage4_sequence",
+ "feature": "syscall_seq",
+ "subsystem": "scheduler",
+ "type": "syscall_sequence",
+ "severity": severity,
+ "score": round(mismatch, 4),
+ "value": float(misses),
+ "baseline_mean": None,
+ "baseline_std": None,
+ "position": 0.22,
+ "message": (
+ f"Unusual syscall sequencing: {mismatch * 100:.0f}% of recent "
+ f"{window_len} n-grams are novel ({misses} unseen){cause}"
+ ),
+ "meta": {"stage": 4, "mismatch": round(mismatch, 4), "window": window_len},
+ }
+
+
class MLWorker:
def __init__(self, cfg: MLConfig | None = None) -> None:
self.cfg = cfg or MLConfig()
@@ -106,6 +135,22 @@ def __init__(self, cfg: MLConfig | None = None) -> None:
self._last_if_emit = 0.0
self._maybe_load_model()
+ # Stage 4 (syscall sequence / STIDE). Sampler + tracker run regardless so
+ # the n-gram vocabulary keeps growing; scoring only happens once a profile
+ # artifact exists (built by training/retrain).
+ self.seq_sampler = None
+ self.seq_tracker = None
+ self.seq_model = None
+ self._seq_model_mtime: float | None = None
+ self._last_seq_emit = 0.0
+ self._last_seq_flush = 0.0
+ if self.cfg.enable_stage4:
+ from kernel_ai.ml.sequence import NgramTracker, SyscallSampler
+
+ self.seq_sampler = SyscallSampler(max_pids=self.cfg.seq_max_pids)
+ self.seq_tracker = NgramTracker(n=self.cfg.seq_n, window=self.cfg.seq_window)
+ self._maybe_load_seq_model()
+
def _maybe_load_model(self) -> None:
"""Load / hot-reload the IsolationForest artifact if present and changed.
@@ -130,6 +175,63 @@ def _maybe_load_model(self) -> None:
except Exception as exc: # noqa: BLE001 - keep running on Stage 1 only
logger.warning("failed to load Stage 2 model: %s", exc)
+ def _maybe_load_seq_model(self) -> None:
+ """Load / hot-reload the STIDE profile artifact if present and changed."""
+ if not self.cfg.enable_stage4:
+ return
+ path = self.cfg.seq_model_path
+ try:
+ mtime = os.path.getmtime(path)
+ except OSError:
+ return # no profile yet -> sequence scoring stays dormant
+ if self._seq_model_mtime is not None and mtime <= self._seq_model_mtime:
+ return
+ try:
+ from kernel_ai.ml.sequence import StideModel
+
+ self.seq_model = StideModel.load(path)
+ self._seq_model_mtime = mtime
+ logger.info("loaded Stage 4 STIDE profile: %s (%s)", path, self.seq_model.meta)
+ except Exception as exc: # noqa: BLE001 - keep running without Stage 4
+ logger.warning("failed to load STIDE profile: %s", exc)
+
+ def _tick_sequence(self) -> dict | None:
+ """Sample syscalls, grow the n-gram vocabulary, and score the window."""
+ if self.seq_sampler is None or self.seq_tracker is None:
+ return None
+ # Burst of rapid sub-samples: parked daemons still yield X,X,X (normal),
+ # while actively-working processes reveal real syscall transitions.
+ bursts = max(1, self.cfg.seq_subsamples)
+ gap = max(0.0, self.cfg.seq_subsample_gap_ms / 1000.0)
+ for i in range(bursts):
+ samples = self.seq_sampler.sample()
+ if samples:
+ self.seq_tracker.update(samples)
+ if i < bursts - 1 and gap:
+ time.sleep(gap)
+
+ # Periodically persist newly observed n-grams so the profile can grow.
+ now = time.time()
+ if (now - self._last_seq_flush) >= self.cfg.seq_flush_sec:
+ pending = self.seq_tracker.drain_pending()
+ if pending:
+ self.store.upsert_ngram_counts(self.cfg.seq_n, pending)
+ self._last_seq_flush = now
+
+ if self.seq_model is None:
+ return None
+ window = self.seq_tracker.recent()
+ if len(window) < self.cfg.seq_min_window:
+ return None
+ mismatch, misses = self.seq_model.score_window(window)
+ if mismatch < self.cfg.seq_mismatch_warn:
+ return None
+ if (now - self._last_seq_emit) < self.cfg.seq_cooldown_sec:
+ return None
+ self._last_seq_emit = now
+ top = self.seq_model.top_unseen(window, limit=3)
+ return _build_sequence_anomaly(mismatch, misses, len(window), top, self.cfg)
+
def stop(self, *_args) -> None:
self._running = False
@@ -154,6 +256,15 @@ def _tick(self) -> int:
anomalies.append(_build_isoforest_anomaly(if_score, scores, self.cfg))
self._last_if_emit = now
+ # Stage 4 second opinion: anomalous *ordering* of syscalls.
+ if self.cfg.enable_stage4:
+ try:
+ seq_anom = self._tick_sequence()
+ if seq_anom is not None:
+ anomalies.append(seq_anom)
+ except Exception as exc: # noqa: BLE001 - never let Stage 4 kill the tick
+ logger.warning("sequence scoring failed: %s", exc)
+
if self.cfg.store_features:
self.store.insert_feature_snapshot(features)
if anomalies:
@@ -188,6 +299,7 @@ def run(self) -> None:
self.store.prune(self.cfg.retain_features_hours, self.cfg.retain_anomalies_hours)
# Pick up a freshly retrained model without a restart.
self._maybe_load_model()
+ self._maybe_load_seq_model()
except Exception as exc: # noqa: BLE001 - keep the loop alive
logger.exception("tick failed: %s", exc)
# Reconnect on DB hiccups rather than dying.
diff --git a/static/js/kernel-dna.js b/static/js/kernel-dna.js
index b090276..718510d 100755
--- a/static/js/kernel-dna.js
+++ b/static/js/kernel-dna.js
@@ -1582,9 +1582,222 @@ class KernelDNAVisualization {
`;
this.container.appendChild(legendDiv);
+ const driftEl = this._ensureDriftIndicator();
+
await this.addProcessSelector();
const selectorEl = this.container.querySelector('.dna-process-selector');
- this._applyStaggeredReveal([titleDiv, devLabel, legendDiv, selectorEl]);
+ this._applyStaggeredReveal([titleDiv, devLabel, legendDiv, driftEl, selectorEl]);
+ }
+
+ /**
+ * Model-freshness indicator: shows whether the ML baseline is "fresh" or has
+ * "drifted", current flag-rate vs the expected contamination, feature drift,
+ * model artifact age, and a small flag-rate history sparkline. Reads from the
+ * read-only /api/ml-drift endpoint that the drift monitor / retrain job feeds.
+ */
+ _ensureDriftStyles() {
+ if (this._driftStylesInjected || typeof document === 'undefined') return;
+ const style = document.createElement('style');
+ style.id = 'kernel-dna-drift-styles';
+ style.textContent = `
+ .dna-drift-indicator {
+ position: absolute;
+ bottom: 26px;
+ left: 20px;
+ z-index: 1001;
+ width: 196px;
+ padding: 11px 13px 12px;
+ font-family: 'Share Tech Mono', monospace;
+ color: #c8ccd4;
+ cursor: default;
+ background:
+ linear-gradient(180deg, rgba(18,26,32,0.78), rgba(10,13,16,0.72));
+ border: 1px solid rgba(103, 200, 224, 0.28);
+ box-shadow: 0 0 18px rgba(103,200,224,0.08),
+ inset 0 0 24px rgba(103,200,224,0.04);
+ backdrop-filter: blur(3px);
+ overflow: hidden;
+ /* clipped "cut corner" -> instrument-panel silhouette */
+ clip-path: polygon(0 0, calc(100% - 12px) 0, 100% 12px, 100% 100%, 12px 100%, 0 calc(100% - 12px));
+ }
+ /* corner tick brackets */
+ .dna-drift-indicator::before,
+ .dna-drift-indicator::after {
+ content: '';
+ position: absolute;
+ width: 9px; height: 9px;
+ border: 1px solid rgba(103,200,224,0.65);
+ pointer-events: none;
+ }
+ .dna-drift-indicator::before { top: 4px; left: 4px; border-width: 1px 0 0 1px; }
+ .dna-drift-indicator::after { bottom: 4px; right: 4px; border-width: 0 1px 1px 0; }
+ /* slow scanline sweep */
+ .dna-drift-scan {
+ position: absolute; left: 0; right: 0; top: 0; height: 28px;
+ background: linear-gradient(180deg, rgba(103,200,224,0.10), rgba(103,200,224,0));
+ pointer-events: none;
+ animation: dna-drift-scan 4.6s linear infinite;
+ }
+ @keyframes dna-drift-scan {
+ 0% { transform: translateY(-30px); opacity: 0; }
+ 12% { opacity: 1; }
+ 88% { opacity: 1; }
+ 100% { transform: translateY(150px); opacity: 0; }
+ }
+ .dna-drift-head { display:flex; align-items:center; gap:7px; margin-bottom:8px; }
+ .dna-drift-led {
+ width:7px; height:7px; border-radius:50%;
+ background:#9aa3ad; box-shadow:0 0 6px rgba(154,163,173,0.8);
+ animation: dna-drift-led 1.8s ease-in-out infinite;
+ }
+ @keyframes dna-drift-led { 0%,100%{opacity:0.45;} 50%{opacity:1;} }
+ .dna-drift-title {
+ font-size:10px; letter-spacing:1.2px; opacity:0.82; flex:1;
+ text-shadow: 0 0 8px rgba(103,200,224,0.25);
+ }
+ .dna-drift-pill {
+ font-size:8.5px; letter-spacing:0.6px; padding:2px 7px;
+ border:1px solid rgba(120,128,138,0.5); color:#9aa3ad;
+ clip-path: polygon(0 0, calc(100% - 5px) 0, 100% 5px, 100% 100%, 5px 100%, 0 calc(100% - 5px));
+ }
+ .dna-drift-metrics { font-size:10px; line-height:1.6; opacity:0.9; }
+ .dna-drift-metrics .k { opacity:0.5; }
+ `;
+ document.head.appendChild(style);
+ this._driftStylesInjected = true;
+ }
+
+ _ensureDriftIndicator() {
+ let panel = this.container.querySelector('.dna-drift-indicator');
+ if (panel) return panel;
+ this._ensureDriftStyles();
+
+ panel = document.createElement('div');
+ panel.className = 'dna-drift-indicator';
+ window.setSafeHtml(panel, `
+
+
+
+ ML MODEL
+ …
+
+
+
+
flags – / exp –
+
feat drift –
+
model –
+