ML stages

kernel_ai/ml/config.py

@@ -55,9 +55,10 @@ class MLConfig:
     # settle so we don't fire on the cold-start transient).
     warmup_samples: int = _env_int("KERNEL_AI_ML_WARMUP", 30)
 
-    # Robust z-score thresholds. A feature whose current value sits this many
-    # standard deviations above its baseline becomes a mutation.
-    z_warn: float = _env_float("KERNEL_AI_ML_Z_WARN", 4.0)
+    # Robust z-score thresholds. Calibrated on live ring-0.sh traffic
+    # (2026-08-15): z=4 fired on dashboard poll bursts (ctxt/pgfault/retrans).
+    # 6σ keeps the real spikes (z 7–12) and drops the 4.3–5.5 band.
+    z_warn: float = _env_float("KERNEL_AI_ML_Z_WARN", 6.0)
     z_crit: float = _env_float("KERNEL_AI_ML_Z_CRIT", 7.0)
 
     # Persist the raw feature snapshot each tick (useful for later training /
@@ -94,8 +95,14 @@ class MLConfig:
     if_contamination: float = _env_float("KERNEL_AI_ML_IF_CONTAMINATION", 0.02)
     if_n_estimators: int = _env_int("KERNEL_AI_ML_IF_TREES", 200)
     # Min seconds between IsolationForest mutations (avoid per-tick spam during a
-    # sustained anomaly).
-    if_cooldown_sec: float = _env_float("KERNEL_AI_ML_IF_COOLDOWN_SEC", 15.0)
+    # sustained anomaly). 15s capped at 240/h and the live host sat on that
+    # ceiling; 45s caps Stage 2 at 80/h.
+    if_cooldown_sec: float = _env_float("KERNEL_AI_ML_IF_COOLDOWN_SEC", 45.0)
+    # IsolationForest predict() still flags ~2% by construction, but on this
+    # host the decision boundary is noisy (flag rate 0.20–0.53 at retrain).
+    # Require a minimum -decision_function before emit; live weak hits were
+    # 0.05–0.09, real-looking ones 0.11+.
+    if_min_score: float = _env_float("KERNEL_AI_ML_IF_MIN_SCORE", 0.10)
 
     # --- Stage 3 (drift + auto-retrain) ---
     # Window of recent snapshots used to measure drift (minutes).
@@ -203,6 +210,33 @@ class MLConfig:
     # above be derived from the live distribution of a given host.
     stage8_log_every_sec: float = _env_float("KERNEL_AI_ML_STAGE8_LOG_EVERY_SEC", 60.0)
 
+    # --- Stage 9 (HTTP attempt model + success join) ---
+    # Default OFF: do not emit on PROD until offline metrics exist.
+    enable_stage9: bool = os.getenv("KERNEL_AI_ML_STAGE9", "false").lower() == "true"
+    http_nginx_log: str = os.getenv(
+        "KERNEL_AI_ML_HTTP_NGINX_LOG",
+        "/var/log/nginx/kernel-ai.json.log",
+    )
+    http_app_log: str = os.getenv("KERNEL_AI_ML_HTTP_APP_LOG", "")
+    http_model_path: str = os.getenv(
+        "KERNEL_AI_ML_HTTP_MODEL_PATH",
+        str(_DATA_DIR / "http_attempt_latest.joblib"),
+    )
+    http_model_prev_path: str = os.getenv(
+        "KERNEL_AI_ML_HTTP_MODEL_PREV",
+        str(_DATA_DIR / "http_attempt_prev.joblib"),
+    )
+    http_window_sec: int = _env_int("KERNEL_AI_ML_HTTP_WINDOW_SEC", 60)
+    http_cooldown_sec: float = _env_float("KERNEL_AI_ML_HTTP_COOLDOWN_SEC", 45.0)
+    http_min_score: float = _env_float("KERNEL_AI_ML_HTTP_MIN_SCORE", 0.55)
+    http_min_precision: float = _env_float("KERNEL_AI_ML_HTTP_MIN_PRECISION", 0.60)
+    http_max_flag_rate: float = _env_float("KERNEL_AI_ML_HTTP_MAX_FLAG", 0.45)
+    http_join_sec: float = _env_float("KERNEL_AI_ML_HTTP_JOIN_SEC", 30.0)
+    http_mlflow_experiment: str = os.getenv(
+        "KERNEL_AI_ML_HTTP_EXPERIMENT", "kernel_dna_http_attempt"
+    )
+    http_mlflow_model: str = os.getenv("KERNEL_AI_ML_HTTP_MODEL", "kernel_dna_http_attempt")
+
     @property
     def alpha(self) -> float:
         return 2.0 / (max(2, self.baseline_window) + 1.0)

kernel_ai/ml/polygon.py

@@ -379,7 +379,10 @@ def run_mimicry(*, stide_warn: float = 0.30, markov_warn: float = 2.5) -> dict:
 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(
+        "mode",
+        choices=("dry-run", "live", "list", "mimicry", "stream", "http-wordlist", "http-rce"),
+    )
     parser.add_argument(
         "--scenario",
         action="append",
@@ -398,8 +401,24 @@ def main(argv: list[str] | None = None) -> int:
             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)")
+        print(f"{'http-wordlist':16} {'HTTP':8} [dry-only] Stage 9 scanner window")
+        print(f"{'http-rce':16} {'HTTP':8} [dry-only] Stage 9 attempt then exec join")
         return 0
 
+    if args.mode == "http-wordlist":
+        from kernel_ai.ml.http_polygon import run_http_wordlist
+
+        result = run_http_wordlist()
+        print(result)
+        return 0 if result["pass"] else 1
+
+    if args.mode == "http-rce":
+        from kernel_ai.ml.http_polygon import run_http_rce
+
+        result = run_http_rce()
+        print(result)
+        return 0 if result["pass"] else 1
+
     if args.mode == "mimicry":
         result = run_mimicry()
         return 0 if result["pass"] else 1

kernel_ai/ml/retrain.py

@@ -20,6 +20,7 @@
 
 import argparse
 import logging
+import os
 
 from kernel_ai.ml.config import MLConfig
 from kernel_ai.ml.drift import compute_drift
@@ -78,13 +79,31 @@
         logger.warning("Stage 8 Markov build failed: %s", exc)
 
 
+def _refresh_http(cfg: MLConfig, metrics: dict) -> None:
+    """Best-effort Stage 9 retrain. Guardrails live in http_train (keep previous)."""
+    if os.getenv("KERNEL_AI_ML_HTTP_RETRAIN", "true").lower() != "true":
+        return
+    try:
+        from kernel_ai.ml.http_train import train as train_http
+
+        http_metrics = train_http(cfg, min_samples=40, persist=True)
+        metrics["http_precision"] = float(http_metrics.get("holdout_precision") or 0)
+        metrics["http_flag_rate"] = float(http_metrics.get("holdout_flag_rate") or 0)
+        logger.info("HTTP attempt model refreshed: %s", http_metrics)
+    except SystemExit as exc:
+        logger.info("HTTP retrain skipped: %s", exc)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("HTTP retrain failed: %s", exc)
+
+
 def run(*, only_if_drift: bool, min_samples: int, stide_only: bool = False) -> int:
     cfg = MLConfig()
     metrics: dict = {}
 
     if stide_only:
         _refresh_stide(cfg, metrics)
         _refresh_markov(cfg, metrics)
+        _refresh_http(cfg, metrics)
         logger.info("stide-only retrain complete: %s", metrics)
         return 0
 
@@ -114,6 +133,7 @@
 
     _refresh_stide(cfg, metrics)
     _refresh_markov(cfg, metrics)
+    _refresh_http(cfg, metrics)
     logger.info("retrain complete: %s", metrics)
     return 0
 

kernel_ai/ml/store.py

@@ -89,11 +89,52 @@
     last_seen   timestamptz NOT NULL DEFAULT now(),
     PRIMARY KEY (parent_comm, child_comm)
 );
+
+CREATE TABLE IF NOT EXISTS ml_http_labels (
+    id         bigserial PRIMARY KEY,
+    ts         timestamptz NOT NULL,
+    src_ip     text        NOT NULL,
+    window_sec integer     NOT NULL DEFAULT 60,
+    label      text        NOT NULL,
+    cls        text,
+    why        text,
+    teacher    text        NOT NULL DEFAULT 'rules',
+    features   jsonb
+);
+CREATE INDEX IF NOT EXISTS ml_http_labels_ts_idx ON ml_http_labels (ts);
+
+CREATE TABLE IF NOT EXISTS ml_http_windows (
+    ts         timestamptz NOT NULL DEFAULT now(),
+    src_ip     text        NOT NULL,
+    window_sec integer     NOT NULL DEFAULT 60,
+    features   jsonb       NOT NULL,
+    label      text,
+    cls        text
+);
+CREATE INDEX IF NOT EXISTS ml_http_windows_ts_idx ON ml_http_windows (ts);
 """
 
 
+# libpq's default client cert is ~/.postgresql/postgresql.crt. Gunicorn runs
+# with ProtectHome=yes, so opening that path returns EACCES and the whole
+# connection dies — the DNA /ml-anomalies API then looks empty even though
+# the worker is writing. A missing file is fine; an unreadable HOME is not.
+# Point the cert paths at a location that simply does not exist.
+_NO_CLIENT_CERT = "/nonexistent/kernel-ai-pg.crt"
+_NO_CLIENT_KEY = "/nonexistent/kernel-ai-pg.key"
+
+
+def connect_kwargs(*, autocommit: bool = True) -> dict:
+    """psycopg kwargs that do not touch the service user's home directory."""
+    return {
+        "autocommit": autocommit,
+        "sslcert": _NO_CLIENT_CERT,
+        "sslkey": _NO_CLIENT_KEY,
+    }
+
+
 def connect(dsn: str, *, autocommit: bool = True) -> psycopg.Connection:
-    return psycopg.connect(dsn, autocommit=autocommit)
+    return psycopg.connect(dsn, **connect_kwargs(autocommit=autocommit))
 
 
 def ensure_schema(conn: psycopg.Connection) -> None:
@@ -245,6 +286,65 @@ def prune(self, features_hours: int, anomalies_hours: int) -> None:
                 "DELETE FROM ml_proc_snapshots WHERE ts < now() - make_interval(hours => %s)",
                 (features_hours,),
             )
+            cur.execute(
+                "DELETE FROM ml_http_labels WHERE ts < now() - make_interval(hours => %s)",
+                (anomalies_hours,),
+            )
+            cur.execute(
+                "DELETE FROM ml_http_windows WHERE ts < now() - make_interval(hours => %s)",
+                (features_hours,),
+            )
+
+    def insert_http_labels(self, rows: list[dict]) -> int:
+        if not rows:
+            return 0
+        with self.conn.cursor() as cur:
+            cur.executemany(
+                """
+                INSERT INTO ml_http_labels
+                    (ts, src_ip, window_sec, label, cls, why, teacher, features)
+                VALUES
+                    (to_timestamp(%(ts)s), %(src_ip)s, %(window_sec)s, %(label)s,
+                     %(cls)s, %(why)s, %(teacher)s, %(features)s)
+                """,
+                [
+                    {
+                        "ts": float(r["ts"]),
+                        "src_ip": r["src_ip"],
+                        "window_sec": int(r.get("window_sec") or 60),
+                        "label": r["label"],
+                        "cls": r.get("cls"),
+                        "why": r.get("why"),
+                        "teacher": r.get("teacher") or "rules",
+                        "features": Json(r.get("features") or {}),
+                    }
+                    for r in rows
+                ],
+            )
+        return len(rows)
+
+    def insert_http_windows(self, rows: list[dict]) -> None:
+        if not rows:
+            return
+        with self.conn.cursor() as cur:
+            cur.executemany(
+                """
+                INSERT INTO ml_http_windows (ts, src_ip, window_sec, features, label, cls)
+                VALUES (to_timestamp(%(ts)s), %(src_ip)s, %(window_sec)s, %(features)s,
+                        %(label)s, %(cls)s)
+                """,
+                [
+                    {
+                        "ts": float(r["ts"]),
+                        "src_ip": r["src_ip"],
+                        "window_sec": int(r.get("window_sec") or 60),
+                        "features": Json(r.get("features") or {}),
+                        "label": r.get("label"),
+                        "cls": r.get("cls"),
+                    }
+                    for r in rows
+                ],
+            )
 
     def close(self) -> None:
         try:
@@ -394,3 +494,47 @@ def fetch_recent_anomalies(dsn: str, *, since_seconds: int = 120, limit: int = 1
     except Exception as exc:  # noqa: BLE001 - API must degrade gracefully
         logger.warning("fetch_recent_anomalies failed: %s", exc)
         return []
+
+
+def fetch_http_labels(dsn: str, *, hours: int = 24, limit: int = 20000) -> list[dict]:
+    """Labeled HTTP windows for training / gap reports. Empty on DB trouble."""
+    try:
+        with connect(dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT extract(epoch from ts) AS ts, src_ip, window_sec, label, cls,
+                       why, teacher, features
+                FROM ml_http_labels
+                WHERE ts > now() - make_interval(hours => %s)
+                ORDER BY ts DESC
+                LIMIT %s
+                """,
+                (max(1, int(hours)), limit),
+            )
+            cols = [d.name for d in cur.description]
+            return [dict(zip(cols, row)) for row in cur.fetchall()]
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("fetch_http_labels failed: %s", exc)
+        return []
+
+
+def fetch_anomalies_since(dsn: str, *, hours: int = 24, limit: int = 5000) -> list[dict]:
+    """Anomalies with epoch ts for offline joins. Empty on DB trouble."""
+    try:
+        with connect(dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT extract(epoch from ts) AS ts, source, feature, type, severity,
+                       score, message, meta
+                FROM ml_anomalies
+                WHERE ts > now() - make_interval(hours => %s)
+                ORDER BY ts DESC
+                LIMIT %s
+                """,
+                (max(1, int(hours)), limit),
+            )
+            cols = [d.name for d in cur.description]
+            return [dict(zip(cols, row)) for row in cur.fetchall()]
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("fetch_anomalies_since failed: %s", exc)
+        return []