diff --git a/index.html b/index.html index 0bb59e9..7223bb5 100755 --- a/index.html +++ b/index.html @@ -100,7 +100,7 @@

Linux Kernel Ring 0 Visualization

- + diff --git a/kernel_ai/ml/config.py b/kernel_ai/ml/config.py index 1bf7766..ad1bd59 100644 --- a/kernel_ai/ml/config.py +++ b/kernel_ai/ml/config.py @@ -118,9 +118,16 @@ class MLConfig: # --- 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. + # Default source is L0 procfs sampling. PROD Stage 6 uses SEQ_SOURCE=socket + # fed by deploy/ebpf/syscall_stream_collector.py (no caps on this worker). enable_stage4: bool = os.getenv("KERNEL_AI_ML_STAGE4", "true").lower() == "true" + # procfs | socket | off — see docs/ML_STAGE6_L2_COLLECTOR.md + seq_source: str = os.getenv("KERNEL_AI_ML_SEQ_SOURCE", "procfs").strip().lower() + seq_socket: str = os.getenv( + "KERNEL_AI_ML_SEQ_SOCKET", + "/run/kernel-ai/ml-syscall.sock", + ) + seq_socket_max_events: int = _env_int("KERNEL_AI_ML_SEQ_SOCKET_MAX", 2000) seq_model_path: str = os.getenv( "KERNEL_AI_ML_SEQ_MODEL_PATH", str(_DATA_DIR / "stide_latest.joblib"), @@ -146,6 +153,43 @@ class MLConfig: # 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) + # --- Stage 5 (per-process L1 features + lineage) --- + # Default off until explicitly enabled locally / after soak — keeps PROD safe + # if code is synced without a deliberate cutover. + enable_stage5: bool = os.getenv("KERNEL_AI_ML_STAGE5", "false").lower() == "true" + proc_max_pids: int = _env_int("KERNEL_AI_ML_PROC_MAX_PIDS", 80) + proc_lineage_min_count: int = _env_int("KERNEL_AI_ML_PROC_LINEAGE_MIN", 3) + proc_cooldown_sec: float = _env_float("KERNEL_AI_ML_PROC_COOLDOWN_SEC", 20.0) + proc_max_emit: int = _env_int("KERNEL_AI_ML_PROC_MAX_EMIT", 4) + proc_store_snapshots: bool = ( + os.getenv("KERNEL_AI_ML_PROC_STORE", "true").lower() == "true" + ) + proc_flush_sec: float = _env_float("KERNEL_AI_ML_PROC_FLUSH_SEC", 30.0) + + # --- Stage 7 (ATT&CK / Sigma-lite attribution) --- + # Pure enrichment of anomalies already emitted — safe default ON locally. + # Does not change detection thresholds; only adds attack.* metadata. + enable_stage7: bool = os.getenv("KERNEL_AI_ML_STAGE7", "true").lower() == "true" + attack_min_confidence: float = _env_float("KERNEL_AI_ML_ATTACK_MIN_CONF", 0.35) + + # --- Stage 8 (deep sequence: Markov/HMM → LSTM/Transformer) --- + # Stub: safe default OFF. Requires trained artifact + preferably Stage 6 + # stream; without a model file the worker path is a no-op. + enable_stage8: bool = os.getenv("KERNEL_AI_ML_STAGE8", "false").lower() == "true" + stage8_backend: str = os.getenv("KERNEL_AI_ML_STAGE8_BACKEND", "markov").strip().lower() + stage8_markov_path: str = os.getenv( + "KERNEL_AI_ML_STAGE8_MARKOV_PATH", + str(_DATA_DIR / "markov_latest.joblib"), + ) + stage8_lstm_path: str = os.getenv( + "KERNEL_AI_ML_STAGE8_LSTM_PATH", + str(_DATA_DIR / "lstm_latest.pt"), + ) + stage8_window: int = _env_int("KERNEL_AI_ML_STAGE8_WINDOW", 64) + stage8_score_warn: float = _env_float("KERNEL_AI_ML_STAGE8_SCORE_WARN", 3.0) + stage8_score_crit: float = _env_float("KERNEL_AI_ML_STAGE8_SCORE_CRIT", 5.0) + stage8_cooldown_sec: float = _env_float("KERNEL_AI_ML_STAGE8_COOLDOWN_SEC", 30.0) + @property def alpha(self) -> float: return 2.0 / (max(2, self.baseline_window) + 1.0) diff --git a/kernel_ai/ml/polygon.py b/kernel_ai/ml/polygon.py new file mode 100644 index 0000000..8febe1d --- /dev/null +++ b/kernel_ai/ml/polygon.py @@ -0,0 +1,436 @@ +"""Local simulation polygon for Stage 5 / 7 / 8 (dev only — never for PROD). + +Modes: + dry-run — synthetic ProcSamples → Stage 5 detector → Stage 7 enricher + live — spawn short-lived safe processes for a running STAGE5 worker + mimicry — Stage 8 vs STIDE: known-token odd order (STIDE misses, Markov hits) + stream — Stage 6 e2e: demo collector → unix socket → n-grams → STIDE/Markov + +Examples: + python -m kernel_ai.ml.polygon dry-run + python -m kernel_ai.ml.polygon live --scenario reverse_shell + python -m kernel_ai.ml.polygon mimicry + python -m kernel_ai.ml.polygon stream +""" + +from __future__ import annotations + +import argparse +import json +import logging +import os +import shutil +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +from kernel_ai.ml.attribution.enrich import enrich_anomalies +from kernel_ai.ml.proc_baseline import ProcBaselineDetector +from kernel_ai.ml.proc_features import ProcSample + +logger = logging.getLogger("kernel_ai.ml.polygon") + +# Labels written for a future supervised Stage 7 classifier. +_LABELS_DIR = Path(os.getenv("KERNEL_AI_ML_DATA_DIR", "mldata")) / "polygon" + + +SCENARIOS: dict[str, dict] = { + "reverse_shell": { + "technique": "T1071", + "family": "command_and_control", + "description": "bash → nc (sleep binary renamed)", + "live": True, + }, + "web_shell": { + "technique": "T1059", + "family": "execution", + "description": "nginx → bash (fake nginx wrapper)", + "live": True, + }, + "miner_stub": { + "technique": "T1496", + "family": "impact", + "description": "bash → xmrig (sleep binary renamed)", + "live": True, + }, + "scanner": { + "technique": "T1046", + "family": "discovery", + "description": "bash → nmap (sleep binary renamed)", + "live": True, + }, + "privesc": { + "technique": "T1548", + "family": "privilege_escalation", + "description": "synthetic euid=0 / ruid≠0 (dry-run only)", + "live": False, + }, + "lineage_shell": { + "technique": "T1059", + "family": "execution", + "description": "bash → sleep (generic unusual child)", + "live": True, + }, +} + + +def _sample(**kwargs) -> ProcSample: + base = dict( + pid=9000, + ppid=1, + comm="sleep", + parent_comm="bash", + ruid=1000, + euid=1000, + age_sec=5.0, + num_threads=1, + fd_count=8, + vm_rss_mb=3.0, + ) + base.update(kwargs) + s = ProcSample(**base) + s.features = s.score_vector() + return s + + +def _synthetic_samples(scenario: str) -> list[ProcSample]: + if scenario == "reverse_shell": + return [_sample(pid=9101, comm="nc", parent_comm="bash", age_sec=4.0)] + if scenario == "web_shell": + return [_sample(pid=9102, comm="bash", parent_comm="nginx", age_sec=3.0)] + if scenario == "miner_stub": + return [_sample(pid=9103, comm="xmrig", parent_comm="bash", age_sec=6.0)] + 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)] + if scenario == "lineage_shell": + return [_sample(pid=9106, comm="sleep", parent_comm="bash", age_sec=4.0)] + raise KeyError(scenario) + + +def _detector() -> ProcBaselineDetector: + return ProcBaselineDetector( + alpha=0.2, + warmup_samples=0, + z_warn=4.0, + z_crit=7.0, + lineage_min_count=3, + cooldown_sec=0.0, + max_emit_per_tick=16, + ) + + +def run_dry(scenarios: list[str], *, write_labels: bool = True) -> list[dict]: + """Score synthetic samples and print Stage 5 + 7 results.""" + results = [] + det = _detector() + # Pre-seed common noise edges so only attack-like edges stay novel. + det.lineage.load_counts( + [ + ("systemd", "sshd", 20), + ("systemd", "nginx", 20), + ("bash", "grep", 20), + ] + ) + + for name in scenarios: + samples = _synthetic_samples(name) + # Unique pids per scenario so lineage observe fires. + anoms = det.score(samples, now=time.time()) + enriched = enrich_anomalies(anoms, min_confidence=0.3) + expected = SCENARIOS[name] + attributed = [a for a in enriched if a.get("attack")] + match = next( + (a for a in attributed if a["attack"].get("mitre") == expected["technique"]), + None, + ) + hit = match or (attributed[0] if attributed else None) + got_mitre = (hit or {}).get("attack", {}).get("mitre") + ok = match is not None + row = { + "scenario": name, + "expected_mitre": expected["technique"], + "got_mitre": got_mitre, + "pass": ok, + "anomalies": enriched, + } + results.append(row) + status = "PASS" if ok else "FAIL" + print(f"[{status}] {name}: expected {expected['technique']} got {got_mitre}") + if hit: + atk = hit["attack"] + print(f" {hit.get('message', '')[:100]}") + print( + f" attack={atk.get('mitre')} family={atk.get('family')} " + f"src={atk.get('source')} conf={atk.get('label_confidence')}" + ) + elif anoms: + print(f" stage5 fired but no attack above confidence: {anoms[0].get('type')}") + else: + print(" no Stage 5 anomaly emitted") + + if write_labels: + _LABELS_DIR.mkdir(parents=True, exist_ok=True) + path = _LABELS_DIR / f"labels_{int(time.time())}.jsonl" + with path.open("w", encoding="utf-8") as fh: + for row in results: + for anom in row["anomalies"]: + fh.write( + json.dumps( + { + "scenario": row["scenario"], + "expected_mitre": row["expected_mitre"], + "anomaly": { + k: anom.get(k) + for k in ( + "source", + "feature", + "type", + "severity", + "message", + "meta", + "attack", + ) + }, + } + ) + + "\n" + ) + print(f"\nlabels → {path}") + return results + + +def _spawn_renamed_sleep(name: str, duration: float, *, via_bash: bool = True) -> subprocess.Popen: + """Copy ``sleep`` to a temp binary named ``name`` and run it (short-lived).""" + tmp = tempfile.mkdtemp(prefix="kai-poly-") + binary = os.path.join(tmp, name) + sleep_bin = shutil.which("sleep") or "/bin/sleep" + shutil.copy(sleep_bin, binary) + os.chmod(binary, 0o755) + sec = max(1, int(duration)) + if via_bash: + # parent_comm should be bash for sigma reverse_shell / miner heuristics + return subprocess.Popen( + ["bash", "-c", f'exec "{binary}" {sec}'], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + return subprocess.Popen( + [binary, str(sec)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def _spawn_fake_nginx_shell(duration: float) -> subprocess.Popen: + """Wrapper named nginx that starts bash -c sleep (web_shell lineage).""" + tmp = tempfile.mkdtemp(prefix="kai-poly-") + wrapper = os.path.join(tmp, "nginx") + with open(wrapper, "w", encoding="utf-8") as fh: + fh.write("#!/bin/bash\nexec bash -c 'sleep %d'\n" % max(1, int(duration))) + os.chmod(wrapper, 0o755) + return subprocess.Popen( + [wrapper], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def run_live(scenarios: list[str], *, duration: float = 6.0) -> None: + """Spawn safe processes for a running ML worker (STAGE5=true) to observe.""" + procs: list[subprocess.Popen] = [] + print("LIVE polygon — ensure ML worker runs with KERNEL_AI_ML_STAGE5=true") + print(f"holding processes ~{duration}s so the worker can sample them\n") + try: + for name in scenarios: + meta = SCENARIOS[name] + if not meta.get("live"): + print(f"[skip] {name}: dry-run only ({meta['description']})") + continue + if name == "reverse_shell": + procs.append(_spawn_renamed_sleep("nc", duration)) + elif name == "miner_stub": + procs.append(_spawn_renamed_sleep("xmrig", duration)) + elif name == "scanner": + procs.append(_spawn_renamed_sleep("nmap", duration)) + elif name == "lineage_shell": + procs.append( + subprocess.Popen( + ["bash", "-c", f"sleep {max(1, int(duration))}"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + ) + elif name == "web_shell": + procs.append(_spawn_fake_nginx_shell(duration)) + print(f"[spawn] {name}: {meta['description']} (expect {meta['technique']})") + time.sleep(duration + 0.5) + finally: + for p in procs: + try: + p.wait(timeout=2) + except Exception: + try: + p.kill() + except Exception: + pass + print("\ndone — check /api/ml-anomalies or Kernel DNA for stage5_process + attack.*") + + +def _token_ngrams(tokens: list[str], n: int = 3) -> list[str]: + if len(tokens) < n: + return [] + return ["|".join(tokens[i : i + n]) for i in range(len(tokens) - n + 1)] + + +# Known-token adversarial order (all tokens appear in synthetic normal corpus, +# but transitions are vanishingly rare) — classic STIDE mimicry gap. +_MIMICRY_WINDOW = ( + ["openat", "clone", "sendto", "mmap", "write", "futex", "exit"] * 6 +) +_NORMAL_WINDOW = ( + ["openat", "read", "read", "close", "openat", "fstat", "read", "close"] * 5 +) +# Truly novel tokens — STIDE should fire here too. +_NOVEL_WINDOW = ( + ["openat", "execve", "connect", "dup2", "socket", "execve"] * 5 +) + + +def run_mimicry(*, stide_warn: float = 0.30, markov_warn: float = 2.5) -> dict: + """Compare STIDE vs Markov on a mimicry window. Returns a result dict.""" + from kernel_ai.ml.sequence import StideModel + from kernel_ai.ml.sequence_deep.train_markov import ( + _SYNTHETIC_NORMAL, + load_corpus_synthetic, + ) + from kernel_ai.ml.sequence_deep.encode import SequenceEncoder + from kernel_ai.ml.sequence_deep.markov import MarkovScorer + + sequences = load_corpus_synthetic(repeats=40) + encoder = SequenceEncoder() + markov = MarkovScorer(order=1, meta={"stage": 8, "source": "polygon_mimicry"}) + for seq in sequences: + encoder.fit(seq) + markov.observe(seq) + + # STIDE profile: trigrams from normal training + the held-out normal window + # + mimicry trigrams (attacker only used eventually-"known" n-grams). + stide_vocab: set[str] = set() + for seq in list(_SYNTHETIC_NORMAL) + [_NORMAL_WINDOW]: + stide_vocab.update(_token_ngrams(seq, 3)) + stide_vocab.update(_token_ngrams(_MIMICRY_WINDOW, 3)) + stide = StideModel(n=3, ngrams=stide_vocab, meta={"poison": "mimicry_demo"}) + + normal_m = markov.score_window(_NORMAL_WINDOW) + mimic_m = markov.score_window(_MIMICRY_WINDOW) + novel_m = markov.score_window(_NOVEL_WINDOW) + + normal_s, _ = stide.score_window(_token_ngrams(_NORMAL_WINDOW, 3)) + mimic_s, mimic_misses = stide.score_window(_token_ngrams(_MIMICRY_WINDOW, 3)) + novel_s, novel_misses = stide.score_window(_token_ngrams(_NOVEL_WINDOW, 3)) + + n_score = float((normal_m or {}).get("neg_avg_logprob") or 0.0) + m_score = float((mimic_m or {}).get("neg_avg_logprob") or 0.0) + v_score = float((novel_m or {}).get("neg_avg_logprob") or 0.0) + + stide_misses_mimicry = mimic_s < stide_warn + markov_catches_mimicry = m_score >= markov_warn and m_score > n_score * 1.25 + stide_catches_novel = novel_s >= stide_warn + passed = stide_misses_mimicry and markov_catches_mimicry + + result = { + "pass": passed, + "stide": { + "normal_mismatch": round(normal_s, 4), + "mimicry_mismatch": round(mimic_s, 4), + "mimicry_misses": mimic_misses, + "novel_mismatch": round(novel_s, 4), + "novel_misses": novel_misses, + "misses_mimicry": stide_misses_mimicry, + "catches_novel": stide_catches_novel, + }, + "markov": { + "normal_neg_avg_logprob": n_score, + "mimicry_neg_avg_logprob": m_score, + "novel_neg_avg_logprob": v_score, + "catches_mimicry": markov_catches_mimicry, + "worst_mimicry": (mimic_m or {}).get("worst_tokens"), + }, + "thresholds": {"stide_warn": stide_warn, "markov_warn": markov_warn}, + } + + print("Stage 8 mimicry demo (STIDE vs Markov)") + print(f" STIDE normal_mismatch={normal_s:.3f} mimicry={mimic_s:.3f} novel={novel_s:.3f}") + print(f" Markov normal_neg_lp={n_score:.3f} mimicry={m_score:.3f} novel={v_score:.3f}") + print(f" STIDE misses mimicry (<{stide_warn}): {stide_misses_mimicry}") + print(f" Markov catches mimicry (≥{markov_warn} & >1.25×normal): {markov_catches_mimicry}") + print(f" STIDE catches novel (≥{stide_warn}): {stide_catches_novel}") + if (mimic_m or {}).get("worst_tokens"): + print(f" Markov worst transition: {' → '.join(mimic_m['worst_tokens'])}") + print(f"\n[{'PASS' if passed else 'FAIL'}] mimicry case") + return result + + +def main(argv: list[str] | None = None) -> int: + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + parser = argparse.ArgumentParser(description="Stage 5/7/8 local polygon (dev only)") + parser.add_argument("mode", choices=("dry-run", "live", "list", "mimicry", "stream")) + parser.add_argument( + "--scenario", + action="append", + choices=sorted(SCENARIOS), + help="scenario id (repeatable); default: all dry-run / all live-capable", + ) + parser.add_argument("--all", action="store_true", help="all scenarios for this mode") + parser.add_argument("--duration", type=float, default=6.0, help="live hold seconds") + parser.add_argument("--no-labels", action="store_true") + parser.add_argument("--bursts", type=int, default=9, help="stream mode demo bursts") + args = parser.parse_args(argv) + + if args.mode == "list": + for name, meta in SCENARIOS.items(): + live = "live+dry" if meta["live"] else "dry-only" + print(f"{name:16} {meta['technique']:8} [{live}] {meta['description']}") + print(f"{'mimicry':16} {'T1106':8} [dry-only] STIDE miss / Markov hit (Stage 8)") + print(f"{'stream':16} {'L2':8} [dry-only] Stage 6 socket e2e (demo collector)") + return 0 + + if args.mode == "mimicry": + result = run_mimicry() + return 0 if result["pass"] else 1 + + if args.mode == "stream": + from kernel_ai.ml.collectors.stream_e2e import run_stream_e2e + + result = run_stream_e2e(bursts=args.bursts, demo_every=0.05) + print("Stage 6 stream e2e (demo collector → socket → n-grams)") + for k, v in result.items(): + if k == "pass": + continue + print(f" {k}: {v}") + print(f"\n[{'PASS' if result['pass'] else 'FAIL'}] stream e2e") + return 0 if result["pass"] else 1 + + if args.scenario: + scenarios = list(dict.fromkeys(args.scenario)) + elif args.all or args.mode == "dry-run": + scenarios = list(SCENARIOS) + else: + scenarios = [n for n, m in SCENARIOS.items() if m.get("live")] + + if args.mode == "dry-run": + results = run_dry(scenarios, write_labels=not args.no_labels) + failed = sum(1 for r in results if not r["pass"]) + print(f"\n{len(results) - failed}/{len(results)} scenarios attributed as expected") + return 1 if failed else 0 + + run_live(scenarios, duration=args.duration) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/kernel_ai/ml/proc_baseline.py b/kernel_ai/ml/proc_baseline.py new file mode 100644 index 0000000..a0dfc47 --- /dev/null +++ b/kernel_ai/ml/proc_baseline.py @@ -0,0 +1,210 @@ +"""Stage 5 detectors: per-comm EWMA baselines + parent→child lineage whitelist.""" + +from __future__ import annotations + +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 + + +@dataclass +class LineageWhitelist: + """Frequency table of normal parent_comm → child_comm pairs. + + A pair becomes "normal" only after ``min_count`` observations (STIDE-style + poison guard: a one-off attack edge never enters the whitelist). + """ + + min_count: int = 3 + min_age_sec: float = 2.0 + max_alert_age_sec: float = 120.0 + _counts: dict[tuple[str, str], int] = field(default_factory=dict) + _pending: dict[tuple[str, str], int] = field(default_factory=dict) + + def observe(self, parent: str, child: str) -> int: + """Increment and return the new count for this edge.""" + key = (parent or "?", child or "?") + self._counts[key] = self._counts.get(key, 0) + 1 + self._pending[key] = self._pending.get(key, 0) + 1 + return self._counts[key] + + def is_below_threshold(self, parent: str, child: str) -> bool: + key = (parent or "?", child or "?") + return self._counts.get(key, 0) < self.min_count + + def drain_pending(self) -> list[tuple[str, str, int]]: + pending = self._pending + self._pending = {} + return [(p, c, n) for (p, c), n in pending.items()] + + def load_counts(self, rows: list[tuple[str, str, int]]) -> None: + for parent, child, count in rows or []: + key = (str(parent), str(child)) + self._counts[key] = max(self._counts.get(key, 0), int(count)) + + +class ProcBaselineDetector: + """Score process samples → Stage 5 anomaly records.""" + + def __init__( + self, + *, + alpha: float, + warmup_samples: int, + z_warn: float, + z_crit: float, + lineage_min_count: int = 3, + cooldown_sec: float = 20.0, + max_emit_per_tick: int = 4, + ) -> None: + self.z_warn = z_warn + self.z_crit = z_crit + self.cooldown_sec = cooldown_sec + self.max_emit = max_emit_per_tick + self.baseline = EwmaBaseline(alpha=alpha, warmup_samples=warmup_samples) + self.lineage = LineageWhitelist(min_count=lineage_min_count) + self._last_emit: dict[str, float] = {} + self._seen_pids: set[int] = set() + + def _cooldown_ok(self, key: str, now: float) -> bool: + last = self._last_emit.get(key, 0.0) + if (now - last) < self.cooldown_sec: + return False + self._last_emit[key] = now + return True + + def _meta(self, sample: ProcSample, kind: str) -> dict: + return { + "stage": 5, + "kind": kind, + "pid": sample.pid, + "ppid": sample.ppid, + "comm": sample.comm, + "parent_comm": sample.parent_comm, + "ruid": sample.ruid, + "euid": sample.euid, + } + + def score(self, samples: list[ProcSample], *, now: float) -> list[dict]: + out: list[dict] = [] + if not samples: + return out + + # --- lineage (observe each pid once, after it is old enough to score) --- + if len(self._seen_pids) > 20000: + self._seen_pids.clear() + for sample in samples: + if sample.pid in self._seen_pids: + continue + if sample.age_sec < self.lineage.min_age_sec: + continue # wait until the child settles; don't mark seen yet + self._seen_pids.add(sample.pid) + count = self.lineage.observe(sample.parent_comm, sample.comm) + if sample.age_sec > self.lineage.max_alert_age_sec: + continue # learn long-lived edges silently + if count >= self.lineage.min_count: + continue + # Prefer the more specific privesc signal when both apply. + if sample.euid == 0 and sample.ruid != 0: + continue + ckey = f"lineage:{sample.parent_comm}->{sample.comm}" + if not self._cooldown_ok(ckey, now): + continue + out.append( + { + "source": "stage5_process", + "feature": f"lineage:{sample.parent_comm}->{sample.comm}", + "subsystem": PROC_SUBSYSTEM, + "type": f"lineage:{sample.parent_comm}->{sample.comm}", + "severity": "high" if sample.euid == 0 or sample.age_sec < 15 else "medium", + "score": float(count), + "value": float(sample.age_sec), + "baseline_mean": None, + "baseline_std": None, + "position": PROC_POSITION, + "message": ( + f"Unusual process lineage: {sample.parent_comm} → {sample.comm} " + f"(pid={sample.pid}, age={sample.age_sec:.1f}s, seen={count})" + ), + "meta": self._meta(sample, "lineage"), + } + ) + if len(out) >= self.max_emit: + return out + + # --- euid root / ruid non-root --- + for sample in samples: + if not (sample.euid == 0 and sample.ruid != 0): + continue + ckey = f"privesc:{sample.comm}" + if not self._cooldown_ok(ckey, now): + continue + out.append( + { + "source": "stage5_process", + "feature": f"privesc:{sample.comm}", + "subsystem": PROC_SUBSYSTEM, + "type": "proc_anomaly:euid_root", + "severity": "high", + "score": 1.0, + "value": float(sample.euid), + "baseline_mean": float(sample.ruid), + "baseline_std": None, + "position": PROC_POSITION, + "message": ( + f"Privilege anomaly: {sample.comm} pid={sample.pid} " + f"euid=0 ruid={sample.ruid}" + ), + "meta": self._meta(sample, "privesc"), + } + ) + if len(out) >= self.max_emit: + return out + + # --- per-comm EWMA (fd / threads / rss) --- + vectors: dict[str, float] = {} + owners: dict[str, ProcSample] = {} + min_std: dict[str, float] = {} + for sample in samples: + for feat, value in sample.score_vector().items(): + key = f"{sample.comm}::{feat}" + vectors[key] = value + owners[key] = sample + min_std[key] = PROC_MIN_STD.get(feat, 1.0) + + scores = self.baseline.update_and_score(vectors, min_std) + ranked = sorted(scores.values(), key=lambda s: s.z, reverse=True) + for sc in ranked: + if sc.warm or sc.z < self.z_warn or sc.value <= sc.mean: + continue + sample = owners.get(sc.name) + if sample is None: + continue + feat = sc.name.split("::", 1)[-1] + ckey = f"proc:{sample.comm}:{feat}" + if not self._cooldown_ok(ckey, now): + continue + severity = "high" if sc.z >= self.z_crit else "medium" + out.append( + { + "source": "stage5_process", + "feature": f"proc:{sample.comm}:{feat}", + "subsystem": PROC_SUBSYSTEM, + "type": f"proc_anomaly:{feat}", + "severity": severity, + "score": round(sc.z, 3), + "value": round(sc.value, 3), + "baseline_mean": round(sc.mean, 3), + "baseline_std": round(sc.std, 3), + "position": PROC_POSITION, + "message": ( + f"Process {feat} spike: {sample.comm} pid={sample.pid} " + f"{sc.value:.1f} vs baseline {sc.mean:.1f}±{sc.std:.1f} (z={sc.z:.1f})" + ), + "meta": {**self._meta(sample, "baseline"), "feature": feat, "z": round(sc.z, 3)}, + } + ) + if len(out) >= self.max_emit: + break + return out diff --git a/kernel_ai/ml/proc_features.py b/kernel_ai/ml/proc_features.py new file mode 100644 index 0000000..aa532b5 --- /dev/null +++ b/kernel_ai/ml/proc_features.py @@ -0,0 +1,208 @@ +"""Stage 5 — L1 per-process feature extraction from procfs. + +Host-wide Stages 1–2 cannot name *which* process is odd. This module samples a +bounded set of PIDs and builds a small feature vector + parent→child lineage +edge per process. No root required; io/fd best-effort when readable. +""" + +from __future__ import annotations + +import os +import time +from dataclasses import dataclass, field + + +# Helix band for process/lineage mutations (scheduler region). +PROC_POSITION = 0.12 +PROC_SUBSYSTEM = "sched" + +# Features scored by the per-comm EWMA baseline. +PROC_SCORE_FEATURES = ("fd_count", "num_threads", "vm_rss_mb") +PROC_MIN_STD = { + "fd_count": 2.0, + "num_threads": 1.0, + "vm_rss_mb": 1.0, +} + + +@dataclass +class ProcSample: + pid: int + ppid: int + comm: str + parent_comm: str + ruid: int + euid: int + age_sec: float + num_threads: int + fd_count: int + vm_rss_mb: float + tcp_estab: int = 0 + tcp_listen: int = 0 + features: dict[str, float] = field(default_factory=dict) + + def score_vector(self) -> dict[str, float]: + return { + "fd_count": float(self.fd_count), + "num_threads": float(self.num_threads), + "vm_rss_mb": float(self.vm_rss_mb), + } + + +def _read_comm(pid: int) -> str: + try: + with open(f"/proc/{pid}/comm", "r", encoding="utf-8", errors="ignore") as fh: + return (fh.read().strip() or "?")[:64] + except OSError: + return "?" + + +def _parse_stat(pid: int) -> tuple[str, str, int, int, int, int] | None: + """Return (comm, state, ppid, minflt, num_threads, starttime).""" + try: + with open(f"/proc/{pid}/stat", "r", encoding="utf-8", errors="ignore") as fh: + raw = fh.read() + except OSError: + return None + try: + # comm may contain spaces/parens: pid (comm) state ppid ... + lpar = raw.find("(") + rpar = raw.rfind(")") + if lpar < 0 or rpar < 0: + return None + comm = raw[lpar + 1 : rpar][:64] or "?" + rest = raw[rpar + 2 :].split() + # rest[0]=state, [1]=ppid, [7]=minflt, [17]=num_threads, [19]=starttime + state = rest[0] + ppid = int(rest[1]) + minflt = int(rest[7]) + num_threads = int(rest[17]) + starttime = int(rest[19]) + return comm, state, ppid, minflt, num_threads, starttime + except (IndexError, ValueError): + return None + + +def _read_uids(pid: int) -> tuple[int, int]: + ruid = euid = 0 + try: + with open(f"/proc/{pid}/status", "r", encoding="utf-8", errors="ignore") as fh: + for line in fh: + if line.startswith("Uid:"): + parts = line.split() + # Uid: real effective saved fs + ruid = int(parts[1]) + euid = int(parts[2]) + break + except (OSError, ValueError, IndexError): + pass + return ruid, euid + + +def _read_vm_rss_mb(pid: int) -> float: + try: + with open(f"/proc/{pid}/status", "r", encoding="utf-8", errors="ignore") as fh: + for line in fh: + if line.startswith("VmRSS:"): + # kB + kb = float(line.split()[1]) + return kb / 1024.0 + except (OSError, ValueError, IndexError): + pass + return 0.0 + + +def _count_fds(pid: int) -> int: + try: + return len(os.listdir(f"/proc/{pid}/fd")) + except OSError: + return 0 + + +def _boot_time() -> float: + try: + with open("/proc/stat", "r", encoding="utf-8", errors="ignore") as fh: + for line in fh: + if line.startswith("btime "): + return float(line.split()[1]) + except (OSError, ValueError, IndexError): + pass + return time.time() - 1.0 + + +def _clk_tck() -> float: + try: + return float(os.sysconf("SC_CLK_TCK")) + except (ValueError, OSError, AttributeError): + return 100.0 + + +class ProcFeatureExtractor: + """Sample up to ``max_pids`` interesting processes each tick.""" + + def __init__(self, max_pids: int = 80) -> None: + self.max_pids = max(8, max_pids) + self._boot = _boot_time() + self._hz = _clk_tck() + self._comm_cache: dict[int, str] = {} + + def _parent_comm(self, ppid: int) -> str: + if ppid <= 0: + return "kernel" + cached = self._comm_cache.get(ppid) + if cached is not None: + return cached + name = _read_comm(ppid) + self._comm_cache[ppid] = name + return name + + def collect(self) -> list[ProcSample]: + now = time.time() + candidates: list[tuple[float, ProcSample]] = [] + try: + pids = [int(d) for d in os.listdir("/proc") if d.isdigit()] + except OSError: + return [] + + # Refresh lightly; drop stale cache entries opportunistically. + if len(self._comm_cache) > 4096: + self._comm_cache.clear() + + for pid in pids: + parsed = _parse_stat(pid) + if parsed is None: + continue + comm, _state, ppid, _minflt, num_threads, starttime = parsed + self._comm_cache[pid] = comm + age_sec = max(0.0, now - (self._boot + starttime / self._hz)) + ruid, euid = _read_uids(pid) + fd_count = _count_fds(pid) + vm_rss_mb = _read_vm_rss_mb(pid) + parent_comm = self._parent_comm(ppid) + sample = ProcSample( + pid=pid, + ppid=ppid, + comm=comm, + parent_comm=parent_comm, + ruid=ruid, + euid=euid, + age_sec=age_sec, + num_threads=num_threads, + fd_count=fd_count, + vm_rss_mb=vm_rss_mb, + ) + sample.features = sample.score_vector() + # Prefer young / privileged-odd / fd-heavy processes for the budget. + interest = 0.0 + if age_sec < 60: + interest += 5.0 - min(5.0, age_sec / 12.0) + if euid == 0 and ruid != 0: + interest += 8.0 + if fd_count > 32: + interest += min(4.0, fd_count / 64.0) + if num_threads > 8: + interest += 1.0 + candidates.append((interest, sample)) + + candidates.sort(key=lambda x: x[0], reverse=True) + return [s for _, s in candidates[: self.max_pids]] diff --git a/kernel_ai/ml/sequence.py b/kernel_ai/ml/sequence.py index ffc539c..09a0d23 100644 --- a/kernel_ai/ml/sequence.py +++ b/kernel_ai/ml/sequence.py @@ -7,14 +7,14 @@ 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. +Data source: + * L0 ``procfs`` — sample ``/proc//syscall`` each tick (coarse). + * L2 ``socket`` — Stage 6 collector pushes ordered events + (see ``docs/ML_STAGE6_L2_COLLECTOR.md``). Only the source changes; the + n-gram model / store / UI stay the same. Components: - SyscallSampler - read current syscall per pid from procfs + SyscallSampler - read current syscall per pid from procfs (L0) NgramTracker - per-pid rolling deques -> stream of syscall n-grams StideModel - set of "normal" n-grams + window mismatch scoring """ @@ -85,6 +85,17 @@ def __init__(self, n: int = 3, window: int = 400, max_pids: int = 4096) -> None: # Counts of every n-gram observed since the last flush (profile growth). self._pending: dict[str, int] = {} + def _append(self, pid: int, name: str) -> None: + 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 update(self, samples: dict[int, str]) -> None: # Drop histories for pids that vanished to bound memory. if len(self._hist) > self.max_pids: @@ -92,15 +103,27 @@ def update(self, samples: dict[int, str]) -> None: 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 + self._append(int(pid), str(name)) + + def update_stream(self, events) -> int: + """Ingest an ordered L2 event stream (Stage 6). Returns events consumed.""" + n = 0 + for ev in events or []: + try: + pid = int(getattr(ev, "pid", None) if not isinstance(ev, dict) else ev["pid"]) + name = str(getattr(ev, "syscall", None) if not isinstance(ev, dict) else ev["syscall"]) + except (TypeError, ValueError, KeyError): + continue + if not name: + continue + self._append(pid, name) + n += 1 + if len(self._hist) > self.max_pids: + # Bound memory: drop oldest pid histories opportunistically. + overflow = len(self._hist) - self.max_pids + for dead in list(self._hist.keys())[:overflow]: + self._hist.pop(dead, None) + return n def recent(self) -> list[str]: return list(self._recent) diff --git a/kernel_ai/ml/store.py b/kernel_ai/ml/store.py index aad053a..d899011 100644 --- a/kernel_ai/ml/store.py +++ b/kernel_ai/ml/store.py @@ -71,6 +71,24 @@ last_seen timestamptz NOT NULL DEFAULT now() ); CREATE INDEX IF NOT EXISTS ml_syscall_ngrams_n_idx ON ml_syscall_ngrams (n); + +CREATE TABLE IF NOT EXISTS ml_proc_snapshots ( + ts timestamptz NOT NULL DEFAULT now(), + pid integer NOT NULL, + ppid integer, + comm text NOT NULL, + features jsonb NOT NULL +); +CREATE INDEX IF NOT EXISTS ml_proc_snapshots_ts_idx ON ml_proc_snapshots (ts); + +CREATE TABLE IF NOT EXISTS ml_proc_lineage ( + parent_comm text NOT NULL, + child_comm text NOT NULL, + count bigint NOT NULL DEFAULT 0, + first_seen timestamptz NOT NULL DEFAULT now(), + last_seen timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (parent_comm, child_comm) +); """ @@ -147,6 +165,47 @@ def upsert_ngram_counts(self, n: int, counts: dict[str, int]) -> None: [{"ngram": g, "n": n, "count": c} for g, c in counts.items()], ) + def insert_proc_snapshots(self, rows: list[dict]) -> None: + """Persist a compact Stage 5 process sample batch.""" + if not rows: + return + with self.conn.cursor() as cur: + cur.executemany( + """ + INSERT INTO ml_proc_snapshots (pid, ppid, comm, features) + VALUES (%(pid)s, %(ppid)s, %(comm)s, %(features)s) + """, + [ + { + "pid": r["pid"], + "ppid": r.get("ppid"), + "comm": r["comm"], + "features": Json(r.get("features") or {}), + } + for r in rows + ], + ) + + def upsert_lineage_counts(self, edges: list[tuple[str, str, int]]) -> None: + if not edges: + return + with self.conn.cursor() as cur: + cur.executemany( + """ + INSERT INTO ml_proc_lineage (parent_comm, child_comm, count) + VALUES (%(parent)s, %(child)s, %(count)s) + ON CONFLICT (parent_comm, child_comm) DO UPDATE + SET count = ml_proc_lineage.count + EXCLUDED.count, + last_seen = now() + """, + [{"parent": p, "child": c, "count": n} for p, c, n in edges], + ) + + def load_lineage_counts(self) -> list[tuple[str, str, int]]: + with self.conn.cursor() as cur: + cur.execute("SELECT parent_comm, child_comm, count FROM ml_proc_lineage") + return [(r[0], r[1], int(r[2])) for r in cur.fetchall()] + def save_baseline(self, rows: list[dict]) -> None: if not rows: return @@ -182,6 +241,10 @@ def prune(self, features_hours: int, anomalies_hours: int) -> None: "DELETE FROM ml_anomalies WHERE ts < now() - make_interval(hours => %s)", (anomalies_hours,), ) + cur.execute( + "DELETE FROM ml_proc_snapshots WHERE ts < now() - make_interval(hours => %s)", + (features_hours,), + ) def close(self) -> None: try: @@ -309,7 +372,7 @@ def fetch_recent_anomalies(dsn: str, *, since_seconds: int = 120, limit: int = 1 cur.execute( """ SELECT ts, source, feature, subsystem, type, severity, score, value, - baseline_mean, baseline_std, position, message + baseline_mean, baseline_std, position, message, meta FROM ml_anomalies WHERE ts > now() - make_interval(secs => %s) ORDER BY ts DESC @@ -323,6 +386,9 @@ def fetch_recent_anomalies(dsn: str, *, since_seconds: int = 120, limit: int = 1 rec = dict(zip(cols, row)) if rec.get("ts") is not None: rec["ts"] = rec["ts"].isoformat() + meta = rec.get("meta") if isinstance(rec.get("meta"), dict) else {} + if meta.get("attack") and not rec.get("attack"): + rec["attack"] = meta["attack"] out.append(rec) return out except Exception as exc: # noqa: BLE001 - API must degrade gracefully diff --git a/kernel_ai/ml/worker.py b/kernel_ai/ml/worker.py index 4fb0388..e4f5b03 100644 --- a/kernel_ai/ml/worker.py +++ b/kernel_ai/ml/worker.py @@ -135,22 +135,76 @@ 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). + # Stage 4 (syscall sequence / STIDE). Source is selectable: + # procfs — L0 /proc//syscall sampler (default) + # socket — L2 Stage 6 collector (docs/ML_STAGE6_L2_COLLECTOR.md) + # off — disable sequence path entirely self.seq_sampler = None + self.seq_socket_source = 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: + self._seq_source = (self.cfg.seq_source or "procfs").strip().lower() + if self.cfg.enable_stage4 and self._seq_source != "off": 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) + if self._seq_source == "socket": + from kernel_ai.ml.collectors.socket_source import SocketSyscallSource + + self.seq_socket_source = SocketSyscallSource( + self.cfg.seq_socket, + max_events=self.cfg.seq_socket_max_events, + ) + logger.info("Stage 4 source=socket (%s)", self.cfg.seq_socket) + else: + self.seq_sampler = SyscallSampler(max_pids=self.cfg.seq_max_pids) + logger.info("Stage 4 source=procfs") self._maybe_load_seq_model() + # Stage 5 — per-process L1 features + lineage (off by default). + self.proc_extractor = None + self.proc_detector = None + self._last_proc_flush = 0.0 + if self.cfg.enable_stage5: + from kernel_ai.ml.proc_baseline import ProcBaselineDetector + from kernel_ai.ml.proc_features import ProcFeatureExtractor + + self.proc_extractor = ProcFeatureExtractor(max_pids=self.cfg.proc_max_pids) + self.proc_detector = ProcBaselineDetector( + alpha=self.cfg.alpha, + warmup_samples=self.cfg.warmup_samples, + z_warn=self.cfg.z_warn, + z_crit=self.cfg.z_crit, + lineage_min_count=self.cfg.proc_lineage_min_count, + cooldown_sec=self.cfg.proc_cooldown_sec, + max_emit_per_tick=self.cfg.proc_max_emit, + ) + try: + self.proc_detector.lineage.load_counts(self.store.load_lineage_counts()) + except Exception as exc: # noqa: BLE001 + logger.warning("failed to load lineage whitelist: %s", exc) + logger.info( + "Stage 5 enabled: max_pids=%d lineage_min=%d", + self.cfg.proc_max_pids, + self.cfg.proc_lineage_min_count, + ) + + # Stage 8 — deep sequence stub (Markov/LSTM). No-op without artifact. + self.deep_scorer = None + self._last_stage8_emit = 0.0 + if self.cfg.enable_stage8: + from kernel_ai.ml.sequence_deep import DeepSequenceScorer + + self.deep_scorer = DeepSequenceScorer(self.cfg) + logger.info( + "Stage 8 enabled (backend=%s, ready=%s)", + self.cfg.stage8_backend, + self.deep_scorer.ready, + ) + def _maybe_load_model(self) -> None: """Load / hot-reload the IsolationForest artifact if present and changed. @@ -196,19 +250,27 @@ def _maybe_load_seq_model(self) -> None: 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: + """Ingest syscalls, grow the n-gram vocabulary, and score the window.""" + if self.seq_tracker is None: + return None + + if self.seq_socket_source is not None: + events = self.seq_socket_source.drain() + if events: + self.seq_tracker.update_stream(events) + elif self.seq_sampler is not 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) + else: 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() @@ -232,6 +294,67 @@ def _tick_sequence(self) -> dict | None: top = self.seq_model.top_unseen(window, limit=3) return _build_sequence_anomaly(mismatch, misses, len(window), top, self.cfg) + def _tick_stage8(self) -> dict | None: + """Stage 8 stub: score the Stage 4 rolling window if a model is ready.""" + if self.deep_scorer is None or self.seq_tracker is None: + return None + self.deep_scorer.maybe_reload() + if not self.deep_scorer.ready: + return None + window = self.seq_tracker.recent() + if len(window) < max(8, self.cfg.stage8_window // 4): + return None + tokens = window[-self.cfg.stage8_window :] + score = self.deep_scorer.score_tokens(tokens) + if not score: + return None + neg = float(score.get("neg_avg_logprob") or score.get("perplexity") or 0.0) + if neg < self.cfg.stage8_score_warn: + return None + now = time.time() + if (now - self._last_stage8_emit) < self.cfg.stage8_cooldown_sec: + return None + self._last_stage8_emit = now + return self.deep_scorer.build_anomaly(score, self.cfg) + + def _tick_process(self) -> list[dict]: + """Stage 5: sample processes, score lineage/baselines, persist whitelist.""" + if self.proc_extractor is None or self.proc_detector is None: + return [] + samples = self.proc_extractor.collect() + now = time.time() + anomalies = self.proc_detector.score(samples, now=now) + + if self.cfg.proc_store_snapshots and samples and (now - self._last_proc_flush) >= self.cfg.proc_flush_sec: + # Persist a compact interesting subset (already interest-ranked). + rows = [ + { + "pid": s.pid, + "ppid": s.ppid, + "comm": s.comm, + "features": { + **s.features, + "parent_comm": s.parent_comm, + "age_sec": round(s.age_sec, 2), + "ruid": s.ruid, + "euid": s.euid, + }, + } + for s in samples[:16] + ] + try: + self.store.insert_proc_snapshots(rows) + except Exception as exc: # noqa: BLE001 + logger.warning("proc snapshot insert failed: %s", exc) + pending = self.proc_detector.lineage.drain_pending() + if pending: + try: + self.store.upsert_lineage_counts(pending) + except Exception as exc: # noqa: BLE001 + logger.warning("lineage upsert failed: %s", exc) + self._last_proc_flush = now + return anomalies + def stop(self, *_args) -> None: self._running = False @@ -265,6 +388,34 @@ def _tick(self) -> int: except Exception as exc: # noqa: BLE001 - never let Stage 4 kill the tick logger.warning("sequence scoring failed: %s", exc) + # Stage 5: which *process* looks odd (lineage / per-comm baselines). + if self.cfg.enable_stage5: + try: + anomalies.extend(self._tick_process()) + except Exception as exc: # noqa: BLE001 - never let Stage 5 kill the tick + logger.warning("process scoring failed: %s", exc) + + # Stage 8: deep sequence (Markov/LSTM) — stub no-op without artifact. + if self.cfg.enable_stage8: + try: + deep_anom = self._tick_stage8() + if deep_anom is not None: + anomalies.append(deep_anom) + except Exception as exc: # noqa: BLE001 + logger.warning("stage8 scoring failed: %s", exc) + + # Stage 7: ATT&CK / Sigma-lite labels on whatever Stages 1–5 emitted. + if self.cfg.enable_stage7 and anomalies: + try: + from kernel_ai.ml.attribution import enrich_anomalies + + anomalies = enrich_anomalies( + anomalies, + min_confidence=self.cfg.attack_min_confidence, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("attribution enrich failed: %s", exc) + if self.cfg.store_features: self.store.insert_feature_snapshot(features) if anomalies: @@ -300,6 +451,8 @@ def run(self) -> None: # Pick up a freshly retrained model without a restart. self._maybe_load_model() self._maybe_load_seq_model() + if self.deep_scorer is not None: + self.deep_scorer.maybe_reload() 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/kernel_ai/services/siem.py b/kernel_ai/services/siem.py index e7407c8..b141690 100644 --- a/kernel_ai/services/siem.py +++ b/kernel_ai/services/siem.py @@ -24,7 +24,9 @@ import urllib.request from datetime import datetime, timezone -_ENV_PATH = "/etc/kernel-ai/elastic.env" +# PROD default; local/dev can override with KERNEL_AI_ES_ENV_FILE or export +# KERNEL_AI_ES_URL / KERNEL_AI_ES_API_KEY directly (no file required). +_ENV_PATH = os.environ.get("KERNEL_AI_ES_ENV_FILE", "/etc/kernel-ai/elastic.env") _CACHE_TTL_SEC = 45.0 _cache_lock = threading.Lock() diff --git a/kernel_ai/services/telemetry_orchestration.py b/kernel_ai/services/telemetry_orchestration.py index 801c233..0261e76 100644 --- a/kernel_ai/services/telemetry_orchestration.py +++ b/kernel_ai/services/telemetry_orchestration.py @@ -89,18 +89,22 @@ def _ml_anomalies_to_mutations(rows): if feature in seen: continue seen.add(feature) - mutations.append( - { - "type": feature or row.get("type") or "ml_anomaly", - "severity": row.get("severity", "medium"), - "message": row.get("message", ""), - "description": row.get("message", ""), - "position": row.get("position", 0.5), - "source": "ml", - "subsystem": row.get("subsystem"), - "score": row.get("score"), - } - ) + attack = row.get("attack") + if not attack and isinstance(row.get("meta"), dict): + attack = row["meta"].get("attack") + mut = { + "type": feature or row.get("type") or "ml_anomaly", + "severity": row.get("severity", "medium"), + "message": row.get("message", ""), + "description": row.get("message", ""), + "position": row.get("position", 0.5), + "source": "ml", + "subsystem": row.get("subsystem"), + "score": row.get("score"), + } + if attack: + mut["attack"] = attack + mutations.append(mut) if len(mutations) >= KERNEL_DNA_ML_MAX: break return mutations diff --git a/static/js/devices-belt.js b/static/js/devices-belt.js index e99755d..a41d150 100644 --- a/static/js/devices-belt.js +++ b/static/js/devices-belt.js @@ -1,7 +1,7 @@ // Device Control Surface Visualization -// Version: 24 +// Version: 25 -debugLog('🧲 devices-belt.js v24: Script loading...'); +debugLog('🧲 devices-belt.js v25: Script loading...'); class DevicesBeltVisualization { constructor() { @@ -33,6 +33,7 @@ class DevicesBeltVisualization { this.deviceNodes = []; this.links = []; this.pulses = []; + this.gridCells = []; this.subsystemOrder = ['block', 'net', 'char', 'input', 'usb']; this.busOrder = ['pcie', 'usb', 'virtual', 'net']; @@ -506,22 +507,64 @@ class DevicesBeltVisualization { return line; } - createTinyGrid(origin, cols, rows, cellSize, gap, color, activeCount, target = this.deviceNodes) { + createTinyGrid(origin, cols, rows, cellSize, gap, color, activeCount, target = this.deviceNodes, options = {}) { const total = cols * rows; + const occupied = Math.max(0, Math.min(total, Number(activeCount) || 0)); + const hotCount = Math.max(0, Math.min(occupied, Number(options.hotCount) || 0)); for (let i = 0; i < total; i += 1) { const col = i % cols; const row = Math.floor(i / cols); - const active = i < activeCount || (i + row) % 5 === 0; - this.createBoxNode('', new THREE.Vector3(origin.x + col * (cellSize + gap), origin.y - row * (cellSize + gap), origin.z), new THREE.Vector3(cellSize, cellSize, 0.035), color, { - target, - fillOpacity: active ? 0.34 : 0.035, - opacity: active ? 0.78 : 0.18, - scaleX: 0.05, - scaleY: 0.05 - }); + const isOccupied = i < occupied; + const isHot = isOccupied && i < hotCount; + // Empty cells stay almost hollow; occupied read as lit slots; hot = live activity. + const baseFill = isOccupied ? (isHot ? 0.52 : 0.30) : 0.018; + const baseEdge = isOccupied ? (isHot ? 0.95 : 0.68) : 0.11; + const node = this.createBoxNode( + '', + new THREE.Vector3(origin.x + col * (cellSize + gap), origin.y - row * (cellSize + gap), origin.z), + new THREE.Vector3(cellSize, cellSize, 0.035), + color, + { + target, + fillColor: isOccupied ? (isHot ? 0x123a48 : 0x0a222c) : 0x05080c, + fillOpacity: baseFill, + opacity: baseEdge, + scaleX: 0.05, + scaleY: 0.05 + } + ); + const cellMeta = { + occupied: isOccupied, + hot: isHot, + baseFill, + baseEdge, + phase: Math.random() * Math.PI * 2, + speed: isHot ? 2.4 + Math.random() * 1.6 : 1.1 + Math.random() * 0.8 + }; + node.fill.userData.gridCell = cellMeta; + node.edge.userData.gridCell = cellMeta; + if (isOccupied) { + this.gridCells.push({ fill: node.fill, edge: node.edge }); + } } } + updateGridFlicker(nowMs) { + if (!this.gridCells.length) return; + const t = nowMs * 0.001; + this.gridCells.forEach(({ fill, edge }) => { + const g = fill.userData.gridCell; + if (!g || !g.occupied) return; + const wave = 0.5 + 0.5 * Math.sin(t * g.speed + g.phase); + const fillAmp = g.hot ? 0.20 : 0.07; + const edgeAmp = g.hot ? 0.10 : 0.04; + fill.material.opacity = Math.min(0.85, g.baseFill + wave * fillAmp); + edge.material.opacity = Math.min(1, g.baseEdge + wave * edgeAmp); + fill.material.needsUpdate = true; + edge.material.needsUpdate = true; + }); + } + lifecycleValue(stage, device, bus, category) { const load = Number(device.load_norm || 0); const irq = Number(device.irq_per_sec || 0); @@ -558,6 +601,7 @@ class DevicesBeltVisualization { this.deviceNodes = []; this.links = []; this.pulses = []; + this.gridCells = []; this.interactiveDeviceNodes = []; this.deviceLookup = {}; } @@ -915,7 +959,23 @@ class DevicesBeltVisualization { labelColor: '#061014' }); }); - this.createTinyGrid(new THREE.Vector3(-5.92, 1.12, 0.16), 5, 4, 0.24, 0.07, 0x54d8e8, Math.min(20, 5 + list.length), this.subsystemNodes); + const hotLive = list.filter((d) => ( + Number(d.load_norm || 0) > 0.12 + || Number(d.irq_per_sec || 0) > 0.8 + || Number(d.throughput_mb_s || 0) > 0.05 + )).length; + // Utility kit: one slot per live device (no random filler cells). + this.createTinyGrid( + new THREE.Vector3(-5.92, 1.12, 0.16), + 5, + 4, + 0.24, + 0.07, + 0x54d8e8, + Math.min(20, list.length), + this.subsystemNodes, + { hotCount: Math.min(20, hotLive) } + ); const rightPanel = new THREE.Vector3(3.76, 1.62, 0.14); this.createBoxNode('', rightPanel, new THREE.Vector3(4.15, 2.42, 0.08), 0x54d8e8, { @@ -954,7 +1014,18 @@ class DevicesBeltVisualization { kernelLabel.position.set(-1.25, 0.82, 0.16); this.scene.add(kernelLabel); this.subsystemNodes.push(kernelLabel); - this.createTinyGrid(new THREE.Vector3(-3.75, 0.38, 0.14), 22, 4, 0.17, 0.08, 0x54d8e8, Math.min(88, 16 + list.length * 2), this.subsystemNodes); + // Contact grid: ~3 contact cells per device; hot cells track live load/irq. + this.createTinyGrid( + new THREE.Vector3(-3.75, 0.38, 0.14), + 22, + 4, + 0.17, + 0.08, + 0x54d8e8, + Math.min(88, Math.max(list.length * 3, list.length ? 6 : 0)), + this.subsystemNodes, + { hotCount: Math.min(88, Math.max(hotLive * 3, hotLive ? 2 : 0)) } + ); const kernelParts = [ ['BUS', -3.75], ['DRV', -2.55], ['PROBE', -1.35], ['IRQ', -0.15], ['DMA', 1.05], ['UDEV', 2.25], ['VFS', 3.45] ]; @@ -1161,6 +1232,7 @@ class DevicesBeltVisualization { this.lastFrameTime = now; this.updateLinksAndPulses(dt); + this.updateGridFlicker(now); this.camera.position.x = 0; this.camera.position.z = 13.2; diff --git a/static/js/kernel-dna.js b/static/js/kernel-dna.js index a12f820..613d1e0 100755 --- a/static/js/kernel-dna.js +++ b/static/js/kernel-dna.js @@ -436,10 +436,16 @@ class KernelDNAVisualization { // Severity drives the assessment marker colour (high = red, else yellow). const isML = mutationData.source === 'ml'; const isHigh = String(mutationData.severity || '').toLowerCase() === 'high'; - const wireColor = isML ? 0x67C8E0 : this.colors.mutedText; + const attack = mutationData.attack || null; + const attackColorCss = (attack && attack.color) ? String(attack.color) : null; + const wireColor = isML + ? (attackColorCss ? parseInt(attackColorCss.replace('#', ''), 16) : 0x67C8E0) + : this.colors.mutedText; const markerColor = isHigh ? 0xE0564E : this.colors.signalYellow; - const accentCss = isML ? 'rgba(103, 200, 224, 0.9)' : 'rgba(230, 193, 90, 0.7)'; - const accentText = isML ? '#67C8E0' : '#E6C15A'; + const accentCss = isML + ? (attackColorCss ? attackColorCss : 'rgba(103, 200, 224, 0.9)') + : 'rgba(230, 193, 90, 0.7)'; + const accentText = isML ? (attackColorCss || '#67C8E0') : '#E6C15A'; // Mutation: "broken" appearance - distorted/irregular shape. const geometry = new THREE.OctahedronGeometry(0.25, 0); // Irregular shape @@ -467,8 +473,12 @@ class KernelDNAVisualization { yellowMarker.position.copy(point); yellowMarker.position.y += 0.3; // Position above mutation - const baseLabel = String(mutationData.type || 'anomaly').replace(/_/g, ' ').toUpperCase().slice(0, 22); - const tag = isML ? 'ML' : 'RULE'; + const mitreTag = attack && attack.mitre ? String(attack.mitre) : ''; + const baseLabel = (mitreTag + ? mitreTag + : String(mutationData.type || 'anomaly').replace(/_/g, ' ') + ).toUpperCase().slice(0, 22); + const tag = isML ? (mitreTag ? 'ATT' : 'ML') : 'RULE'; const canvas = document.createElement('canvas'); canvas.width = 256; canvas.height = 48; @@ -504,6 +514,7 @@ class KernelDNAVisualization { mutation.userData.mutationSeverity = mutationData.severity || 'medium'; mutation.userData.mutationScore = mutationData.score; mutation.userData.description = mutationData.description || mutationData.message || 'Anomaly detected'; + mutation.userData.mutationAttack = attack; group.add(mutation); group.add(yellowMarker); @@ -2941,17 +2952,24 @@ class KernelDNAVisualization { } else if (userData.mutationType) { // Mutation - source-coloured header (ML cyan, rule yellow). const isML = userData.mutationSource === 'ml'; - const headColor = isML ? '#67C8E0' : '#E6C15A'; - const sourceLabel = isML ? 'ML baseline detector' : 'Threshold rule'; + const atk = userData.mutationAttack || null; + const headColor = (atk && atk.color) ? atk.color : (isML ? '#67C8E0' : '#E6C15A'); + const sourceLabel = isML ? 'ML detector' : 'Threshold rule'; const sev = String(userData.mutationSeverity || 'medium').toUpperCase(); const scoreLine = (userData.mutationScore != null) - ? `
z-score: ${userData.mutationScore}
` + ? `
score: ${userData.mutationScore}
` + : ''; + const attackLine = atk + ? `
ATT&CK: ${atk.mitre || '?'}${atk.family ? ' · ' + atk.family : ''}${atk.label_confidence != null ? ' · conf ' + atk.label_confidence : ''}
+
${atk.why || atk.name || ''}${atk.source ? ' [' + atk.source + ']' : ''}
+ ${(atk.cve && atk.cve.length) ? `
CVE: ${atk.cve.join(', ')}
` : ''}` : ''; tooltipContent = `
MUTATION: ${userData.mutationType}
${userData.description || 'Anomaly detected'}
+ ${attackLine}
Source: ${sourceLabel}
Severity: ${sev}
${scoreLine} diff --git a/templates/kernel-dna.html b/templates/kernel-dna.html index 680ed77..f12aa9d 100644 --- a/templates/kernel-dna.html +++ b/templates/kernel-dna.html @@ -100,7 +100,7 @@

Why Visualize Kernel Behavior

- + - +