ML collectors, attribution, rules, sequence deep

kernel_ai/ml/attribution/__init__.py

@@ -0,0 +1,5 @@
+"""Stage 7 — ATT&CK / Sigma-lite attribution for ML anomalies."""
+
+from kernel_ai.ml.attribution.enrich import enrich_anomalies
+
+__all__ = ["enrich_anomalies"]

kernel_ai/ml/attribution/attack_map.py

@@ -0,0 +1,170 @@
+"""MITRE ATT&CK technique catalogue + heuristic mapping from anomaly shape.
+
+v1 is intentionally rule-based and explainable. A supervised classifier can
+plug in later (``classifier.py``) without changing the mutation contract.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Technique:
+    mitre: str
+    family: str
+    name: str
+    color: str  # UI accent hint
+    cves: tuple[str, ...] = ()
+
+
+# Compact catalogue of families we can currently speak about from Stages 1–5.
+TECHNIQUES: dict[str, Technique] = {
+    "T1059": Technique("T1059", "execution", "Command and Scripting Interpreter", "#e8a54b"),
+    "T1204": Technique("T1204", "execution", "User Execution", "#e8a54b"),
+    "T1498": Technique("T1498", "impact", "Network Denial of Service", "#e0564e"),
+    "T1499": Technique("T1499", "impact", "Endpoint Denial of Service", "#e0564e"),
+    "T1496": Technique("T1496", "impact", "Resource Hijacking", "#c9a6ff"),
+    "T1071": Technique("T1071", "command_and_control", "Application Layer Protocol", "#67c8e0"),
+    "T1046": Technique("T1046", "discovery", "Network Service Discovery", "#8ff0d2"),
+    "T1068": Technique("T1068", "privilege_escalation", "Exploitation for Privilege Escalation", "#e0564e"),
+    "T1548": Technique("T1548", "privilege_escalation", "Abuse Elevation Control Mechanism", "#e0564e"),
+    "T1083": Technique("T1083", "discovery", "File and Directory Discovery", "#b8c7da"),
+    "T1106": Technique("T1106", "execution", "Native API", "#f0c48a"),
+    "T1055": Technique("T1055", "defense_evasion", "Process Injection", "#c9a6ff"),
+}
+
+
+# child_comm tokens that suggest scripting / shells / C2 helpers
+_EXEC_CHILDREN = frozenset(
+    {
+        "bash", "sh", "dash", "zsh", "python", "python3", "perl", "ruby", "php",
+        "node", "nc", "ncat", "netcat", "socat", "curl", "wget", "busybox",
+    }
+)
+_SCAN_HINTS = frozenset({"nmap", "masscan", "zmap", "nikto"})
+_MINER_HINTS = frozenset({"xmrig", "minerd", "cpuminer", "ethminer"})
+
+
+def _technique_payload(tech: Technique, *, confidence: float, source: str, why: str) -> dict:
+    return {
+        "family": tech.family,
+        "mitre": tech.mitre,
+        "name": tech.name,
+        "label_confidence": round(max(0.0, min(1.0, confidence)), 3),
+        "cve": list(tech.cves),
+        "source": source,
+        "color": tech.color,
+        "why": why,
+    }
+
+
+def map_anomaly(anomaly: dict) -> dict | None:
+    """Return an ``attack`` dict or None if we should leave it unattributed."""
+    source = str(anomaly.get("source") or "")
+    feature = str(anomaly.get("feature") or "")
+    atype = str(anomaly.get("type") or "")
+    message = str(anomaly.get("message") or "").lower()
+    meta = anomaly.get("meta") or {}
+    kind = str(meta.get("kind") or "")
+
+    # --- Stage 5 lineage / privesc ---
+    if source == "stage5_process" or feature.startswith("lineage:") or atype.startswith("lineage:"):
+        child = str(meta.get("comm") or "")
+        parent = str(meta.get("parent_comm") or "")
+        edge = f"{parent}→{child}".lower()
+        child_l = child.lower()
+        if child_l in _MINER_HINTS or any(h in edge for h in _MINER_HINTS):
+            return _technique_payload(
+                TECHNIQUES["T1496"],
+                confidence=0.72,
+                source="heuristic",
+                why=f"lineage suggests miner binary ({parent}→{child})",
+            )
+        if child_l in _SCAN_HINTS:
+            return _technique_payload(
+                TECHNIQUES["T1046"],
+                confidence=0.7,
+                source="heuristic",
+                why=f"lineage suggests scanner ({parent}→{child})",
+            )
+        if child_l in _EXEC_CHILDREN or kind == "lineage":
+            conf = 0.66 if child_l in _EXEC_CHILDREN else 0.45
+            return _technique_payload(
+                TECHNIQUES["T1059"],
+                confidence=conf,
+                source="heuristic",
+                why=f"unusual process lineage {parent}→{child}",
+            )
+
+    if kind == "privesc" or "euid_root" in atype or feature.startswith("privesc:"):
+        return _technique_payload(
+            TECHNIQUES["T1548"],
+            confidence=0.75,
+            source="heuristic",
+            why="effective uid 0 with non-root real uid",
+        )
+
+    # --- Stage 4 / Stage 8 sequence ---
+    if (
+        source in ("stage4_sequence", "stage8_sequence")
+        or feature in ("syscall_seq", "syscall_seq_deep")
+        or atype in ("syscall_sequence", "syscall_sequence_deep")
+    ):
+        why = (
+            "low-likelihood syscall order (deep sequence model)"
+            if source == "stage8_sequence" or feature == "syscall_seq_deep"
+            else "novel syscall sequencing (STIDE mismatch)"
+        )
+        conf = 0.58 if source == "stage8_sequence" else 0.55
+        return _technique_payload(
+            TECHNIQUES["T1106"],
+            confidence=conf,
+            source="heuristic",
+            why=why,
+        )
+
+    # --- Stage 1 / 2 host features ---
+    feat = feature.lower()
+    if any(k in feat for k in ("tcp_retrans", "net_softirq", "tcp_inseg", "tcp_outseg")):
+        return _technique_payload(
+            TECHNIQUES["T1498"],
+            confidence=0.4,
+            source="heuristic",
+            why=f"network-path pressure on {feature}",
+        )
+    if any(k in feat for k in ("ctxt_per_sec", "procs_running", "proc_count", "run_queue", "load1", "cpu_busy")):
+        # Resource pressure — could be DoS or miner; stay conservative.
+        if "miner" in message or "xmrig" in message:
+            tech = TECHNIQUES["T1496"]
+            conf = 0.6
+            why = "host CPU/sched pressure with miner hint"
+        else:
+            tech = TECHNIQUES["T1499"]
+            conf = 0.38
+            why = f"host sched/CPU pressure on {feature}"
+        return _technique_payload(tech, confidence=conf, source="heuristic", why=why)
+    if any(k in feat for k in ("pgfault", "pgmajfault", "pgscan", "swap_io", "psi_mem")):
+        return _technique_payload(
+            TECHNIQUES["T1499"],
+            confidence=0.36,
+            source="heuristic",
+            why=f"memory pressure on {feature}",
+        )
+    if "hardirq" in feat or "block_softirq" in feat:
+        return _technique_payload(
+            TECHNIQUES["T1499"],
+            confidence=0.34,
+            source="heuristic",
+            why=f"IRQ/block pressure on {feature}",
+        )
+
+    if source == "stage2_isoforest":
+        return _technique_payload(
+            TECHNIQUES["T1499"],
+            confidence=0.3,
+            source="heuristic",
+            why="IsolationForest unusual host state (family-level guess)",
+        )
+
+    return None

kernel_ai/ml/attribution/classifier.py

@@ -0,0 +1,12 @@
+"""Placeholder for a future supervised ATT&CK classifier (GB / sklearn).
+
+v1 Stage 7 uses heuristics + Sigma-lite only. This module exists so the
+roadmap file layout stays stable when polygon-labeled training lands.
+"""
+
+from __future__ import annotations
+
+
+def predict_attack(_anomaly: dict) -> dict | None:
+    """Return attack dict or None. Untrained stub always returns None."""
+    return None

kernel_ai/ml/attribution/enrich.py

@@ -0,0 +1,61 @@
+"""Enrich ML anomaly records with Stage 7 ``attack`` attribution."""
+
+from __future__ import annotations
+
+import logging
+
+from kernel_ai.ml.attribution import classifier
+from kernel_ai.ml.attribution.attack_map import map_anomaly
+from kernel_ai.ml.attribution.sigma_engine import load_rules, match_anomaly
+
+logger = logging.getLogger("kernel_ai.ml.attribution")
+
+# Below this confidence we keep the anomaly but mark family as unknown-ish
+# by omitting attack (UI stays uncolored). Tunable via enrich() arg.
+_DEFAULT_MIN_CONF = 0.35
+
+
+def enrich_anomaly(
+    anomaly: dict,
+    *,
+    rules: list[dict] | None = None,
+    min_confidence: float = _DEFAULT_MIN_CONF,
+) -> dict:
+    """Attach ``attack`` + mirror into ``meta.attack`` (back-compat for DNA)."""
+    if not isinstance(anomaly, dict):
+        return anomaly
+    if anomaly.get("attack"):
+        return anomaly
+
+    # Precedence: Sigma-lite (high precision) > heuristic map > ML classifier stub.
+    attack = match_anomaly(anomaly, rules=rules)
+    if attack is None:
+        attack = map_anomaly(anomaly)
+    if attack is None:
+        attack = classifier.predict_attack(anomaly)
+    if attack is None:
+        return anomaly
+    if float(attack.get("label_confidence") or 0.0) < min_confidence:
+        return anomaly
+
+    anomaly = dict(anomaly)
+    anomaly["attack"] = attack
+    meta = dict(anomaly.get("meta") or {})
+    meta["attack"] = attack
+    anomaly["meta"] = meta
+    # Optional: prefix message once for operators curling the API.
+    mitre = attack.get("mitre")
+    if mitre and mitre not in str(anomaly.get("message") or ""):
+        anomaly["message"] = f"[{mitre}] {anomaly.get('message') or ''}".strip()
+    return anomaly
+
+
+def enrich_anomalies(anomalies: list[dict], *, min_confidence: float = _DEFAULT_MIN_CONF) -> list[dict]:
+    if not anomalies:
+        return []
+    try:
+        rules = load_rules()
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("sigma rules load failed: %s", exc)
+        rules = []
+    return [enrich_anomaly(a, rules=rules, min_confidence=min_confidence) for a in anomalies]

kernel_ai/ml/attribution/sigma_engine.py

@@ -0,0 +1,89 @@
+"""Minimal Sigma-lite matcher (JSON rules, no PyYAML dependency).
+
+Full Sigma is Stage 7 aspirational; this covers a few high-precision patterns
+we can express from ML anomaly records + Stage 5 meta (comm/lineage).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+
+logger = logging.getLogger("kernel_ai.ml.attribution.sigma")
+
+_RULES_DIR = Path(__file__).resolve().parents[1] / "rules" / "sigma"
+
+
+def load_rules(directory: Path | None = None) -> list[dict]:
+    root = directory or _RULES_DIR
+    rules: list[dict] = []
+    if not root.is_dir():
+        return rules
+    for path in sorted(root.glob("*.json")):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            logger.warning("skip sigma rule %s: %s", path.name, exc)
+            continue
+        if isinstance(data, dict):
+            data.setdefault("id", path.stem)
+            rules.append(data)
+        elif isinstance(data, list):
+            for i, item in enumerate(data):
+                if isinstance(item, dict):
+                    item.setdefault("id", f"{path.stem}_{i}")
+                    rules.append(item)
+    return rules
+
+
+def match_anomaly(anomaly: dict, rules: list[dict] | None = None) -> dict | None:
+    """Return attack dict from the first matching rule, else None."""
+    rules = rules if rules is not None else load_rules()
+    meta = anomaly.get("meta") or {}
+    hay = {
+        "source": str(anomaly.get("source") or ""),
+        "feature": str(anomaly.get("feature") or ""),
+        "type": str(anomaly.get("type") or ""),
+        "message": str(anomaly.get("message") or ""),
+        "comm": str(meta.get("comm") or ""),
+        "parent_comm": str(meta.get("parent_comm") or ""),
+        "kind": str(meta.get("kind") or ""),
+    }
+    edge = f"{hay['parent_comm']}->{hay['comm']}".lower()
+
+    for rule in rules:
+        when = rule.get("when") or {}
+        ok = True
+        for key, expected in when.items():
+            if key == "child_in":
+                ok = hay["comm"].lower() in {str(x).lower() for x in expected}
+            elif key == "parent_in":
+                ok = hay["parent_comm"].lower() in {str(x).lower() for x in expected}
+            elif key == "edge_contains":
+                ok = any(str(x).lower() in edge for x in expected)
+            elif key == "feature_contains":
+                ok = any(str(x).lower() in hay["feature"].lower() for x in expected)
+            elif key == "source_in":
+                ok = hay["source"] in set(expected)
+            elif key == "kind_in":
+                ok = hay["kind"] in set(expected)
+            else:
+                ok = True
+            if not ok:
+                break
+        if not ok:
+            continue
+        attack = rule.get("attack") or {}
+        return {
+            "family": attack.get("family", "unknown"),
+            "mitre": attack.get("mitre"),
+            "name": attack.get("name") or rule.get("title") or rule.get("id"),
+            "label_confidence": float(attack.get("confidence", 0.85)),
+            "cve": list(attack.get("cve") or []),
+            "source": "sigma",
+            "color": attack.get("color", "#e8a54b"),
+            "why": rule.get("description") or rule.get("title") or rule.get("id"),
+            "rule_id": rule.get("id"),
+        }
+    return None

kernel_ai/ml/collectors/__init__.py

@@ -0,0 +1,11 @@
+"""Syscall event sources for Stage 4 / Stage 6.
+
+Privileged collection lives outside the ML worker (see
+``docs/ML_STAGE6_L2_COLLECTOR.md``). The worker only *drains* a normalized
+event stream and feeds :class:`kernel_ai.ml.sequence.NgramTracker`.
+"""
+
+from kernel_ai.ml.collectors.base import ALLOWED_SYSCALLS, SyscallEvent
+from kernel_ai.ml.collectors.socket_source import SocketSyscallSource
+
+__all__ = ["ALLOWED_SYSCALLS", "SyscallEvent", "SocketSyscallSource"]