diff --git a/index.html b/index.html index cfbe699..aa4cdff 100755 --- a/index.html +++ b/index.html @@ -98,16 +98,19 @@

Linux Kernel Ring 0 Visualization

+ - + + - + - - + + + @@ -123,7 +126,7 @@

Linux Kernel Ring 0 Visualization

console.error('❌ RightSemicircleMenuManager class NOT loaded!'); } - + diff --git a/kernel_ai/http/api_handlers/network_system.py b/kernel_ai/http/api_handlers/network_system.py index 478b9b5..9a8ca6d 100644 --- a/kernel_ai/http/api_handlers/network_system.py +++ b/kernel_ai/http/api_handlers/network_system.py @@ -32,7 +32,9 @@ def _payload(): def network_stack_realtime(): return api_json( lambda: _network_service.get_network_stack_realtime( - network_stack_prev=get_state_container(current_app).network_stack_prev + network_stack_prev=get_state_container(current_app).network_stack_prev, + prefer_local=request.args.get("local"), + prefer_remote=request.args.get("remote"), ) ) diff --git a/kernel_ai/ml/http_features.py b/kernel_ai/ml/http_features.py new file mode 100644 index 0000000..0a40337 --- /dev/null +++ b/kernel_ai/ml/http_features.py @@ -0,0 +1,152 @@ +"""Per-IP windows over HTTP events. No raw body and no home paths leave here.""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass, field +from urllib.parse import unquote_plus + +from kernel_ai.ml.http_parse import HttpEvent +from kernel_ai.ml.http_rules import ( + _CMDI, + _JNDI, + _LFI, + _SQLI, + _XSS, + _UNUSUAL_METHOD, + classify_window, +) + +WINDOW_SEC = 60 + +# Fixed order shared by train and the worker. +HTTP_FEATURE_ORDER = [ + "count", + "rate_per_sec", + "uniq_paths", + "mean_path_len", + "max_path_len", + "frac_404", + "frac_4xx", + "frac_5xx", + "frac_2xx", + "has_dotfile", + "has_traversal", + "has_sqli", + "has_xss", + "has_cmdi", + "has_jndi", + "unusual_method", + "mean_query_len", +] + +_DOTFILE = (".env", ".git", ".svn", ".htaccess", ".htpasswd", ".aws") + + +@dataclass +class HttpWindow: + ts: float + src_ip: str + window_sec: int + events: list[HttpEvent] = field(default_factory=list) + features: dict[str, float] = field(default_factory=dict) + label: str = "benign" + cls: str = "benign" + why: str = "" + + def vector(self) -> list[float]: + return [float(self.features.get(name, 0.0)) for name in HTTP_FEATURE_ORDER] + + +def _flags(events: list[HttpEvent]) -> dict[str, float]: + has_dot = 0.0 + has_trav = 0.0 + has_sqli = 0.0 + has_xss = 0.0 + has_cmdi = 0.0 + has_jndi = 0.0 + unusual = 0.0 + for event in events: + hay = f"{event.path} {unquote_plus(event.query or '')} {event.body}" + path_l = event.path.lower() + if any(token in path_l for token in _DOTFILE): + has_dot = 1.0 + if _LFI.search(hay): + has_trav = 1.0 + if _SQLI.search(hay): + has_sqli = 1.0 + if _XSS.search(hay): + has_xss = 1.0 + if _CMDI.search(hay): + has_cmdi = 1.0 + if _JNDI.search(hay): + has_jndi = 1.0 + if event.method in _UNUSUAL_METHOD: + unusual = 1.0 + return { + "has_dotfile": has_dot, + "has_traversal": has_trav, + "has_sqli": has_sqli, + "has_xss": has_xss, + "has_cmdi": has_cmdi, + "has_jndi": has_jndi, + "unusual_method": unusual, + } + + +def features_of(events: list[HttpEvent], *, window_sec: int = WINDOW_SEC) -> dict[str, float]: + n = len(events) + if n == 0: + return {name: 0.0 for name in HTTP_FEATURE_ORDER} + paths = [event.path for event in events] + lens = [len(event.path) for event in events] + qlens = [len(event.query) for event in events] + n404 = sum(1 for event in events if event.status == 404) + n4xx = sum(1 for event in events if 400 <= event.status < 500) + n5xx = sum(1 for event in events if event.status >= 500) + n2xx = sum(1 for event in events if 200 <= event.status < 300) + flags = _flags(events) + out = { + "count": float(n), + "rate_per_sec": n / max(1.0, float(window_sec)), + "uniq_paths": float(len(set(paths))), + "mean_path_len": sum(lens) / n, + "max_path_len": float(max(lens)), + "frac_404": n404 / n, + "frac_4xx": n4xx / n, + "frac_5xx": n5xx / n, + "frac_2xx": n2xx / n, + "mean_query_len": sum(qlens) / n, + } + out.update(flags) + return out + + +def build_windows( + events: list[HttpEvent], + *, + window_sec: int = WINDOW_SEC, + label: bool = True, +) -> list[HttpWindow]: + """Bucket events by src_ip and floor(ts / window). Empty IP is dropped.""" + buckets: dict[tuple[str, int], list[HttpEvent]] = defaultdict(list) + for event in events: + if not event.src_ip: + continue + slot = int(event.ts // window_sec) + buckets[(event.src_ip, slot)].append(event) + + windows: list[HttpWindow] = [] + for (src_ip, slot), rows in sorted(buckets.items(), key=lambda item: item[0][1]): + feats = features_of(rows, window_sec=window_sec) + win = HttpWindow( + ts=float(slot * window_sec), + src_ip=src_ip, + window_sec=window_sec, + events=rows, + features=feats, + ) + if label: + win.label, win.cls, win.why = classify_window(rows, feats) + windows.append(win) + return windows diff --git a/kernel_ai/ml/http_gap.py b/kernel_ai/ml/http_gap.py new file mode 100644 index 0000000..702739e --- /dev/null +++ b/kernel_ai/ml/http_gap.py @@ -0,0 +1,76 @@ +"""Stage 2 report: HTTP/SIEM attempts vs kernel mutations (hole and forest noise).""" + +from __future__ import annotations + +import argparse +import json +from typing import Any + +from kernel_ai.ml.config import MLConfig +from kernel_ai.ml.http_join import JOIN_SEC, is_web_kernel_anomaly +from kernel_ai.ml.store import fetch_anomalies_since, fetch_http_labels + + +def _ts(row: dict[str, Any]) -> float: + try: + return float(row.get("ts") or 0.0) + except (TypeError, ValueError): + return 0.0 + + +def report_gap( + attempts: list[dict[str, Any]], + kernel_anomalies: list[dict[str, Any]], + *, + slack_sec: float = JOIN_SEC, +) -> dict: + """attempts = labeled HTTP attempt windows; kernel = Stage 1/2/4/5 rows.""" + kernel = [row for row in kernel_anomalies if not str(row.get("source") or "").startswith("stage9")] + hits = 0 + missed = 0 + for attempt in attempts: + ats = _ts(attempt) + near = any(ats <= _ts(row) <= ats + slack_sec for row in kernel) + if near: + hits += 1 + else: + missed += 1 + lonely = 0 + lonely_web = 0 + for row in kernel: + kts = _ts(row) + if any(abs(kts - _ts(attempt)) <= slack_sec for attempt in attempts): + continue + lonely += 1 + if is_web_kernel_anomaly(row): + lonely_web += 1 + return { + "attempts": len(attempts), + "kernel_anomalies": len(kernel), + "attempt_with_kernel": hits, + "attempt_without_kernel": missed, + "kernel_without_attempt": lonely, + "kernel_web_without_attempt": lonely_web, + "slack_sec": slack_sec, + "note": ( + "attempt_without_kernel is the hole (wordlist, SIEM saw it, forest silent). " + "kernel_without_attempt is host-rhythm noise (dashboard polls, retrain)." + ), + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="HTTP attempts vs kernel mutations") + parser.add_argument("--hours", type=int, default=24) + parser.add_argument("--slack-sec", type=float, default=JOIN_SEC) + args = parser.parse_args(argv) + cfg = MLConfig() + labels = [r for r in fetch_http_labels(cfg.dsn, hours=args.hours) if r.get("label") == "attempt"] + kernel = fetch_anomalies_since(cfg.dsn, hours=args.hours) + payload = report_gap(labels, kernel, slack_sec=args.slack_sec) + print(json.dumps(payload, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/kernel_ai/ml/http_join.py b/kernel_ai/ml/http_join.py new file mode 100644 index 0000000..1bb6c63 --- /dev/null +++ b/kernel_ai/ml/http_join.py @@ -0,0 +1,98 @@ +"""Join an HTTP attempt to a later web-stack kernel mutation → success.""" + +from __future__ import annotations + +from typing import Any + +WEB_COMMS = {"nginx", "gunicorn", "python", "python3"} +KERNEL_SOURCES = { + "stage4_sequence", + "stage5_process", + "stage8_markov", + "stage8_lstm", +} +JOIN_SEC = 30.0 +SUCCESS_FEATURE = "http_success:exec" +SUCCESS_POSITION = 0.12 + + +def _comm(anomaly: dict[str, Any]) -> str: + meta = anomaly.get("meta") if isinstance(anomaly.get("meta"), dict) else {} + return str(meta.get("comm") or meta.get("parent_comm") or "").lower() + + +def is_web_kernel_anomaly(anomaly: dict[str, Any]) -> bool: + comm = _comm(anomaly) + if comm in WEB_COMMS: + return True + parent = "" + meta = anomaly.get("meta") if isinstance(anomaly.get("meta"), dict) else {} + parent = str(meta.get("parent_comm") or "").lower() + if parent in WEB_COMMS: + return True + feature = str(anomaly.get("feature") or "") + return any(name in feature for name in WEB_COMMS) + + +def join_success( + attempts: list[dict[str, Any]], + kernel_anomalies: list[dict[str, Any]], + *, + slack_sec: float = JOIN_SEC, +) -> list[dict[str, Any]]: + """Each web-stack kernel hit that follows an attempt becomes one success row.""" + if not attempts or not kernel_anomalies: + return [] + out: list[dict[str, Any]] = [] + seen: set[tuple[str, int]] = set() + for kernel in kernel_anomalies: + if not is_web_kernel_anomaly(kernel): + continue + kts = float(kernel.get("ts") or 0.0) + matched = None + for attempt in attempts: + ats = float(attempt.get("ts") or 0.0) + if ats <= kts <= ats + slack_sec: + matched = attempt + break + if matched is None: + continue + key = (str(matched.get("src_ip") or ""), int(kts)) + if key in seen: + continue + seen.add(key) + cls = str(matched.get("cls") or "anomaly") + src_ip = str(matched.get("src_ip") or "") + ip_tail = ".".join(src_ip.split(".")[-2:]) if src_ip else "--" + meta = dict(kernel.get("meta") or {}) + meta.update( + { + "stage": 9, + "kind": "success", + "cls": cls, + "src_ip": src_ip, + "why": f"attempt:{cls} then {kernel.get('source')}", + "kernel_source": kernel.get("source"), + } + ) + out.append( + { + "source": "stage9_success", + "feature": SUCCESS_FEATURE, + "subsystem": "sched", + "type": SUCCESS_FEATURE, + "severity": "high", + "score": 1.0, + "value": slack_sec, + "baseline_mean": None, + "baseline_std": None, + "position": SUCCESS_POSITION, + "message": ( + f"HTTP attempt ({cls}) then web-stack kernel mutation " + f"({kernel.get('source')}, ip …{ip_tail})" + ), + "meta": meta, + "ts": kts, + } + ) + return out diff --git a/kernel_ai/ml/http_label.py b/kernel_ai/ml/http_label.py new file mode 100644 index 0000000..4513acd --- /dev/null +++ b/kernel_ai/ml/http_label.py @@ -0,0 +1,95 @@ +"""Stage 0: walk local HTTP logs, label IP windows, persist (no Elasticsearch).""" + +from __future__ import annotations + +import argparse +import logging +import time + +from kernel_ai.ml.config import MLConfig +from kernel_ai.ml.http_features import WINDOW_SEC, build_windows +from kernel_ai.ml.http_parse import iter_log_lines +from kernel_ai.ml.store import PostgresStore + +logger = logging.getLogger("kernel_ai.ml.http_label") + + +def collect_events(paths: list[str], *, since_ts: float | None) -> list: + events = [] + for path in paths: + events.extend(iter_log_lines(path, since_ts=since_ts)) + return events + + +def windows_to_rows(windows) -> list[dict]: + return [ + { + "ts": win.ts, + "src_ip": win.src_ip, + "window_sec": win.window_sec, + "label": win.label, + "cls": win.cls, + "why": win.why, + "teacher": "rules", + "features": win.features, + } + for win in windows + ] + + +def label_paths( + paths: list[str], + *, + hours: float, + window_sec: int = WINDOW_SEC, + store: PostgresStore | None = None, +) -> dict: + since = time.time() - hours * 3600.0 + events = collect_events(paths, since_ts=since) + windows = build_windows(events, window_sec=window_sec, label=True) + rows = windows_to_rows(windows) + written = 0 + if store is not None and rows: + written = store.insert_http_labels(rows) + store.insert_http_windows(rows) + attempts = sum(1 for row in rows if row["label"] == "attempt") + return { + "events": len(events), + "windows": len(rows), + "attempts": attempts, + "benign": len(rows) - attempts, + "written": written, + "classes": sorted({row["cls"] for row in rows if row["label"] == "attempt"}), + } + + +def main(argv: list[str] | None = None) -> int: + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + parser = argparse.ArgumentParser(description="Label HTTP windows from local logs") + parser.add_argument("--nginx-log", action="append", default=[], help="nginx json_analytics path") + parser.add_argument("--app-log", action="append", default=[], help="kernel_ai.http JSONL path") + parser.add_argument("--hours", type=float, default=24.0) + parser.add_argument("--window-sec", type=int, default=WINDOW_SEC) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args(argv) + + cfg = MLConfig() + paths = list(args.nginx_log or []) + list(args.app_log or []) + if not paths: + paths = [p for p in (cfg.http_nginx_log, cfg.http_app_log) if p] + if not paths: + raise SystemExit("no log paths: pass --nginx-log / --app-log or set KERNEL_AI_ML_HTTP_*_LOG") + + store = None if args.dry_run else PostgresStore(cfg.dsn) + try: + stats = label_paths(paths, hours=args.hours, window_sec=args.window_sec, store=store) + finally: + if store is not None: + store.close() + logger.info("http_label %s", stats) + print(stats) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/kernel_ai/ml/http_model.py b/kernel_ai/ml/http_model.py new file mode 100644 index 0000000..0f361bb --- /dev/null +++ b/kernel_ai/ml/http_model.py @@ -0,0 +1,88 @@ +"""Supervised HTTP-attempt model. MLflow-free; the worker only loads joblib.""" + +from __future__ import annotations + +import os +import shutil +from dataclasses import dataclass, field +from typing import Any + +import joblib + +from kernel_ai.ml.http_features import HTTP_FEATURE_ORDER + + +@dataclass +class HttpAttemptModel: + feature_names: list[str] = field(default_factory=lambda: list(HTTP_FEATURE_ORDER)) + model: Any = None + meta: dict = field(default_factory=dict) + + def fit(self, matrix: list[list[float]], labels: list[int], *, random_state: int = 42) -> "HttpAttemptModel": + from sklearn.linear_model import LogisticRegression + + clf = LogisticRegression( + max_iter=400, + class_weight="balanced", + random_state=random_state, + ) + clf.fit(matrix, labels) + self.model = clf + self.meta.update( + { + "n_samples": len(matrix), + "n_features": len(self.feature_names), + "n_attempt": int(sum(labels)), + "n_benign": int(len(labels) - sum(labels)), + "algo": "logreg", + } + ) + return self + + def _vectorize(self, features: dict[str, float]) -> list[float]: + return [float(features.get(name, 0.0)) for name in self.feature_names] + + def predict_one(self, features: dict[str, float]) -> tuple[bool, float]: + """Return (is_attempt, probability of attempt).""" + if self.model is None: + return False, 0.0 + vec = [self._vectorize(features)] + proba = float(self.model.predict_proba(vec)[0][1]) + return proba >= 0.5, proba + + def predict_matrix(self, matrix: list[list[float]]) -> list[float]: + if self.model is None or not matrix: + return [] + return [float(p[1]) for p in self.model.predict_proba(matrix)] + + def save(self, path: str) -> None: + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + joblib.dump( + {"feature_names": self.feature_names, "model": self.model, "meta": self.meta}, + path, + ) + + @classmethod + def load(cls, path: str) -> "HttpAttemptModel": + blob = joblib.load(path) + return cls( + feature_names=list(blob.get("feature_names") or HTTP_FEATURE_ORDER), + model=blob.get("model"), + meta=dict(blob.get("meta") or {}), + ) + + +def promote_artifact(new_path: str, latest_path: str, prev_path: str) -> None: + """Keep the previous latest as rollback, then replace latest.""" + if os.path.exists(latest_path): + os.makedirs(os.path.dirname(prev_path) or ".", exist_ok=True) + shutil.copy2(latest_path, prev_path) + os.makedirs(os.path.dirname(latest_path) or ".", exist_ok=True) + shutil.copy2(new_path, latest_path) + + +def rollback_artifact(latest_path: str, prev_path: str) -> bool: + if not os.path.exists(prev_path): + return False + shutil.copy2(prev_path, latest_path) + return True diff --git a/kernel_ai/ml/http_parse.py b/kernel_ai/ml/http_parse.py new file mode 100644 index 0000000..7f7a98a --- /dev/null +++ b/kernel_ai/ml/http_parse.py @@ -0,0 +1,129 @@ +"""Normalize nginx json_analytics and kernel_ai.http ECS lines into one event.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any +from urllib.parse import unquote_plus + + +@dataclass(frozen=True) +class HttpEvent: + ts: float + src_ip: str + method: str + path: str + query: str + status: int + body: str + user_agent: str + dataset: str + + +def _parse_ts(raw: Any) -> float | None: + if raw is None: + return None + if isinstance(raw, (int, float)): + value = float(raw) + return value / 1000.0 if value > 1e12 else value + text = str(raw).strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + return datetime.fromisoformat(text).timestamp() + except ValueError: + return None + + +def _first_ip(*candidates: Any) -> str: + for raw in candidates: + if not raw: + continue + token = str(raw).split(",")[0].strip() + if token and token != "-": + return token + return "" + + +def _split_uri(uri: str) -> tuple[str, str]: + value = unquote_plus(str(uri or "")) + if "?" not in value: + return value or "/", "" + path, query = value.split("?", 1) + return path or "/", query + + +def parse_record(obj: dict[str, Any], *, default_ts: float | None = None) -> HttpEvent | None: + """Accept nginx json_analytics or ECS kernel_ai.http (and Filebeat wrappers).""" + src = obj + if isinstance(obj.get("event_data"), dict): + src = {**obj, **obj["event_data"]} + if isinstance(obj.get("json"), dict): + src = {**src, **obj["json"]} + + method = str(src.get("request_method") or src.get("http.request.method") or "GET").upper() + status_raw = src.get("status", src.get("http.response.status_code", 0)) + try: + status = int(status_raw) + except (TypeError, ValueError): + status = 0 + + path = str(src.get("url.path") or "") + query = str(src.get("url.query") or src.get("args") or "") + if not path: + path, uri_query = _split_uri(str(src.get("request_uri") or src.get("url.original") or "/")) + query = query or uri_query + path = unquote_plus(path) or "/" + + ts = _parse_ts(src.get("time_iso8601") or src.get("@timestamp") or src.get("ts")) + if ts is None: + ts = default_ts + if ts is None: + return None + + body = str(src.get("http.request.body.content") or src.get("kai_lower.body") or "") + dataset = str(src.get("event.dataset") or ("kernel_ai.nginx" if "remote_addr" in src else "kernel_ai.http")) + return HttpEvent( + ts=float(ts), + src_ip=_first_ip(src.get("http_x_forwarded_for"), src.get("source.ip"), src.get("remote_addr")), + method=method, + path=path[:512], + query=query[:1024], + status=status, + body=body[:2048], + user_agent=str(src.get("http_user_agent") or src.get("user_agent.original") or "")[:256], + dataset=dataset, + ) + + +def parse_line(line: str, *, default_ts: float | None = None) -> HttpEvent | None: + text = (line or "").strip() + if not text or text[0] not in "{[": + return None + try: + obj = json.loads(text) + except ValueError: + return None + if not isinstance(obj, dict): + return None + return parse_record(obj, default_ts=default_ts) + + +def iter_log_lines(path: str, *, since_ts: float | None = None): + """Yield events from a JSON-lines file. Missing file → empty.""" + try: + handle = open(path, "r", encoding="utf-8", errors="replace") + except OSError: + return + with handle: + for line in handle: + event = parse_line(line) + if event is None: + continue + if since_ts is not None and event.ts < since_ts: + continue + yield event diff --git a/kernel_ai/ml/http_polygon.py b/kernel_ai/ml/http_polygon.py new file mode 100644 index 0000000..1fa636c --- /dev/null +++ b/kernel_ai/ml/http_polygon.py @@ -0,0 +1,103 @@ +"""Dev-only HTTP polygon: wordlist attempt vs join-to-exec success.""" + +from __future__ import annotations + +import time + +from kernel_ai.ml.http_features import build_windows +from kernel_ai.ml.http_join import join_success +from kernel_ai.ml.http_model import HttpAttemptModel +from kernel_ai.ml.http_parse import HttpEvent + + +def _event(ts: float, ip: str, path: str, status: int = 404, method: str = "GET", query: str = "", body: str = "") -> HttpEvent: + return HttpEvent( + ts=ts, + src_ip=ip, + method=method, + path=path, + query=query, + status=status, + body=body, + user_agent="polygon", + dataset="kernel_ai.nginx", + ) + + +def wordlist_events(now: float | None = None) -> list[HttpEvent]: + now = now or time.time() + ip = "203.0.113.9" + paths = [ + "/.env", "/.git/config", "/admin", "/wp-admin", "/phpmyadmin", + "/etc/passwd", "/.aws/credentials", "/backup.sql", "/id_rsa", + "/api/../etc/passwd", "/hidden", "/secret", "/.htaccess", + ] + return [_event(now - 10 + i, ip, path) for i, path in enumerate(paths)] + + +def benign_events(now: float | None = None) -> list[HttpEvent]: + now = now or time.time() + ip = "198.51.100.4" + return [ + _event(now - 5, ip, "/", 200), + _event(now - 4, ip, "/api/isolation-context", 200), + _event(now - 3, ip, "/api/ml-anomalies", 200), + ] + + +def run_http_wordlist() -> dict: + now = time.time() + attack = build_windows(wordlist_events(now), label=True) + quiet = build_windows(benign_events(now), label=True) + teacher_ok = any(w.label == "attempt" for w in attack) and all(w.label == "benign" for w in quiet) + + matrix = [w.vector() for w in attack + quiet] + labels = [1 if w.label == "attempt" else 0 for w in attack + quiet] + # Duplicate so logreg can fit a tiny set. + matrix = matrix * 8 + labels = labels * 8 + model = HttpAttemptModel() + model.fit(matrix, labels) + att_p = model.predict_one(attack[0].features) if attack else (False, 0.0) + ben_p = model.predict_one(quiet[0].features) if quiet else (True, 1.0) + model_ok = att_p[0] is True and ben_p[0] is False + return { + "pass": teacher_ok and model_ok, + "teacher_attempt": teacher_ok, + "model_attempt": att_p, + "model_benign": ben_p, + "attack_cls": attack[0].cls if attack else None, + "attack_why": attack[0].why if attack else None, + } + + +def run_http_rce() -> dict: + now = time.time() + attempt = { + "ts": now - 8, + "src_ip": "203.0.113.9", + "cls": "sqli", + "source": "stage9_http", + } + kernel = { + "ts": now - 2, + "source": "stage5_process", + "feature": "lineage:nginx->bash", + "meta": {"comm": "bash", "parent_comm": "nginx", "pid": 4242}, + } + hits = join_success([attempt], [kernel], slack_sec=30) + control = join_success([], [kernel], slack_sec=30) + quiet_kernel = { + "ts": now - 2, + "source": "stage2_isoforest", + "feature": "ctxt_per_sec", + "meta": {}, + } + no_web = join_success([attempt], [quiet_kernel], slack_sec=30) + return { + "pass": bool(hits) and not control and not no_web, + "joined": len(hits), + "feature": hits[0]["feature"] if hits else None, + "control_empty": len(control) == 0, + "host_spike_ignored": len(no_web) == 0, + } diff --git a/kernel_ai/ml/http_rules.py b/kernel_ai/ml/http_rules.py new file mode 100644 index 0000000..ffca3e1 --- /dev/null +++ b/kernel_ai/ml/http_rules.py @@ -0,0 +1,122 @@ +"""Weak labels: the same ideas as Elastic kernelai-* rules, run on a local file. + +Teacher only. Inference does not call Elasticsearch. +""" + +from __future__ import annotations + +import re +from typing import Iterable +from urllib.parse import unquote_plus + +from kernel_ai.ml.http_parse import HttpEvent + +# Classes the DNA feature name uses: http_attempt: +CLS_SCANNER = "scanner" +CLS_FLOOD = "flood" +CLS_LFI = "lfi" +CLS_SQLI = "sqli" +CLS_XSS = "xss" +CLS_CMDI = "cmdi" +CLS_SENSITIVE = "sensitive" +CLS_JNDI = "jndi" +CLS_METHOD = "method" +CLS_ANOMALY = "anomaly" + +_SQLI = re.compile( + r"(union\s+select|'[\s]*or[\s]*'|or\s+1=1|sleep\s*\(|information_schema|waitfor\s+delay)", + re.I, +) +_XSS = re.compile(r"(<\s*script|onerror\s*=|javascript:|<\s*img)", re.I) +_CMDI = re.compile( + r"(;|\||`|\$\()\s*(wget|curl|bash|sh|nc|python|perl|chmod)\b" + r"|/bin/(ba)?sh|cmd\.exe", + re.I, +) +_LFI = re.compile(r"(\.\./|\.\.\\|/etc/passwd|/proc/self|/windows/win\.ini)", re.I) +_JNDI = re.compile(r"\$\{jndi:", re.I) +_SENSITIVE = re.compile( + r"(/\.(env|git|svn|htpasswd|htaccess)\b" + r"|/wp-admin|/phpmyadmin|/adminer" + r"|/(id_rsa|shadow|credentials)\b" + r"|/\.aws/|\.sql(\.gz)?$)", + re.I, +) +# Classic probes that never belong on this host. One hit is enough to name +# the window; do not wait for the 404-scanner threshold. +_PROBE = re.compile( + r"(eval-stdin\.php|/vendor/phpunit|/wp-login\.php|/xmlrpc\.php" + r"|/cgi-bin/|/actuator/|/manager/html)", + re.I, +) +_UNUSUAL_METHOD = {"TRACE", "TRACK", "CONNECT", "DEBUG"} + +# Window thresholds (mirror kernelai-flood / 404-scanner orders of magnitude, +# scaled down so a one-minute window on this host still fires). +# 12 missed an 11×404 phpunit sweep; 8 still ignores a handful of broken links. +SCANNER_404 = 8 +FLOOD_COUNT = 80 + + +def _hay(event: HttpEvent) -> str: + query = unquote_plus(event.query or "").replace("+", " ") + return f"{event.path} {query} {event.body}" + + +def classify_event(event: HttpEvent) -> str | None: + """Single-request teacher class, or None if the request is unremarkable.""" + hay = _hay(event) + if _JNDI.search(hay): + return CLS_JNDI + if _LFI.search(hay): + return CLS_LFI + if _SQLI.search(hay): + return CLS_SQLI + if _CMDI.search(hay): + return CLS_CMDI + if _XSS.search(hay): + return CLS_XSS + if _SENSITIVE.search(event.path) or _SENSITIVE.search(event.query): + return CLS_SENSITIVE + if _PROBE.search(event.path): + return CLS_SCANNER + if event.method in _UNUSUAL_METHOD: + return CLS_METHOD + return None + + +def named_attempt_class(label: str, cls: str) -> str | None: + """DNA may only show a class the teacher can name. + + The logistic model treats raw ``count`` as flood-like, so a dashboard + polling /api/* at ~75 req/min scores as attempt while the teacher says + quiet. That is not a fact for the helix. + """ + if label != "attempt": + return None + if not cls or cls in {"benign", CLS_ANOMALY}: + return None + return cls + + +def classify_window(events: Iterable[HttpEvent], flags: dict[str, float]) -> tuple[str, str, str]: + """Return (label, cls, why) for one IP window. + + label is attempt or benign. cls is the DNA feature suffix. + """ + rows = list(events) + n = len(rows) + n404 = sum(1 for event in rows if event.status == 404) + per_event = [classify_event(event) for event in rows] + hit = next((cls for cls in per_event if cls), None) + + if n >= FLOOD_COUNT: + return "attempt", CLS_FLOOD, f"flood:{n} req in window" + if n404 >= SCANNER_404: + return "attempt", CLS_SCANNER, f"scanner:{n404} x404" + if hit: + return "attempt", hit, f"pattern:{hit}" + if flags.get("has_traversal") or flags.get("has_sqli") or flags.get("has_cmdi"): + cls = CLS_LFI if flags.get("has_traversal") else (CLS_SQLI if flags.get("has_sqli") else CLS_CMDI) + return "attempt", cls, f"flag:{cls}" + return "benign", "benign", "quiet" diff --git a/kernel_ai/ml/http_tail.py b/kernel_ai/ml/http_tail.py new file mode 100644 index 0000000..aa3e440 --- /dev/null +++ b/kernel_ai/ml/http_tail.py @@ -0,0 +1,47 @@ +"""Follow a JSON-lines HTTP log without loading the whole file each tick.""" + +from __future__ import annotations + +import os + +from kernel_ai.ml.http_parse import HttpEvent, parse_line + + +class HttpLogTail: + def __init__(self, path: str, *, start: str = "end") -> None: + self.path = path + self._offset = 0 + self._inode: int | None = None + # Default is follow: do not replay the whole nginx log on every restart. + if start == "end": + try: + stat = os.stat(path) + except OSError: + pass + else: + self._offset = stat.st_size + self._inode = stat.st_ino + + def read_new(self, *, now: float | None = None) -> list[HttpEvent]: + try: + stat = os.stat(self.path) + except OSError: + return [] + inode = stat.st_ino + if self._inode is not None and inode != self._inode: + self._offset = 0 + self._inode = inode + if stat.st_size < self._offset: + self._offset = 0 + out: list[HttpEvent] = [] + try: + with open(self.path, "r", encoding="utf-8", errors="replace") as handle: + handle.seek(self._offset) + for line in handle: + event = parse_line(line, default_ts=now) + if event is not None: + out.append(event) + self._offset = handle.tell() + except OSError: + return [] + return out diff --git a/kernel_ai/ml/http_train.py b/kernel_ai/ml/http_train.py new file mode 100644 index 0000000..9213331 --- /dev/null +++ b/kernel_ai/ml/http_train.py @@ -0,0 +1,161 @@ +"""Train the Stage 9 HTTP-attempt model with a champion/challenger gate.""" + +from __future__ import annotations + +import argparse +import logging +import os +import tempfile + +from kernel_ai.ml.config import MLConfig +from kernel_ai.ml.http_features import HTTP_FEATURE_ORDER +from kernel_ai.ml.http_model import HttpAttemptModel, promote_artifact, rollback_artifact +from kernel_ai.ml.store import fetch_http_labels + +logger = logging.getLogger("kernel_ai.ml.http_train") + + +def _rows_to_xy(rows: list[dict]) -> tuple[list[list[float]], list[int], list[str]]: + matrix: list[list[float]] = [] + labels: list[int] = [] + classes: list[str] = [] + for row in rows: + label = str(row.get("label") or "") + if label not in {"attempt", "benign"}: + continue + feats = row.get("features") if isinstance(row.get("features"), dict) else {} + matrix.append([float(feats.get(name, 0.0)) for name in HTTP_FEATURE_ORDER]) + labels.append(1 if label == "attempt" else 0) + classes.append(str(row.get("cls") or ("anomaly" if label == "attempt" else "benign"))) + return matrix, labels, classes + + +def _split(matrix, labels, classes, *, holdout: float = 0.25): + n = len(matrix) + cut = max(1, int(n * (1.0 - holdout))) if n > 4 else n + return ( + matrix[:cut], + labels[:cut], + classes[:cut], + matrix[cut:], + labels[cut:], + classes[cut:], + ) + + +def _precision(probs: list[float], labels: list[int], *, threshold: float = 0.5) -> float: + pred = [1 if p >= threshold else 0 for p in probs] + tp = sum(1 for p, y in zip(pred, labels) if p == 1 and y == 1) + fp = sum(1 for p, y in zip(pred, labels) if p == 1 and y == 0) + return tp / (tp + fp) if (tp + fp) else 0.0 + + +def _flag_rate(probs: list[float], *, threshold: float = 0.5) -> float: + if not probs: + return 0.0 + return sum(1 for p in probs if p >= threshold) / len(probs) + + +def train(cfg: MLConfig, *, min_samples: int, hours: int = 168, persist: bool = True) -> dict: + rows = fetch_http_labels(cfg.dsn, hours=hours, limit=50000) + matrix, labels, classes = _rows_to_xy(rows) + n = len(matrix) + n_attempt = sum(labels) + if n < min_samples or n_attempt < 3 or (n - n_attempt) < 3: + raise SystemExit( + f"Not enough labeled HTTP windows: n={n} attempt={n_attempt} " + f"(need >= {min_samples} and both classes). Run http_label first." + ) + + train_x, train_y, _, hold_x, hold_y, _ = _split(matrix, labels, classes) + if not hold_x: + hold_x, hold_y = train_x, train_y + + model = HttpAttemptModel() + model.fit(train_x, train_y) + hold_p = model.predict_matrix(hold_x) + precision = _precision(hold_p, hold_y) + flag_rate = _flag_rate(hold_p) + metrics = { + "n_samples": float(n), + "n_attempt": float(n_attempt), + "holdout_precision": precision, + "holdout_flag_rate": flag_rate, + "holdout_n": float(len(hold_y)), + } + model.meta.update(metrics) + + if persist: + if precision + 1e-9 < cfg.http_min_precision: + raise SystemExit( + f"Refusing HTTP model: holdout_precision={precision:.3f} " + f"< {cfg.http_min_precision}. Keeping previous artifact." + ) + if flag_rate > cfg.http_max_flag_rate: + raise SystemExit( + f"Refusing HTTP model: holdout_flag_rate={flag_rate:.3f} " + f"> {cfg.http_max_flag_rate}. Keeping previous artifact." + ) + if os.path.exists(cfg.http_model_path): + try: + prev = HttpAttemptModel.load(cfg.http_model_path) + prev_p = float(prev.meta.get("holdout_precision") or 0.0) + if prev_p and precision + 0.02 < prev_p: + raise SystemExit( + f"Challenger worse than champion: {precision:.3f} < {prev_p:.3f}. " + "Keeping previous artifact." + ) + except SystemExit: + raise + except Exception as exc: # noqa: BLE001 + logger.warning("could not compare champion: %s", exc) + + os.makedirs(os.path.dirname(cfg.http_model_path) or ".", exist_ok=True) + fd, tmp = tempfile.mkstemp(suffix=".joblib", dir=os.path.dirname(cfg.http_model_path) or ".") + os.close(fd) + try: + model.save(tmp) + promote_artifact(tmp, cfg.http_model_path, cfg.http_model_prev_path) + finally: + if os.path.exists(tmp): + os.remove(tmp) + logger.info("saved HTTP model -> %s %s", cfg.http_model_path, metrics) + + try: + _log_mlflow(cfg, metrics) + except Exception as exc: # noqa: BLE001 + logger.warning("HTTP MLflow log failed: %s", exc) + return metrics + + +def _log_mlflow(cfg: MLConfig, metrics: dict) -> None: + import mlflow + + os.makedirs(os.path.dirname(cfg.mlflow_uri.replace("sqlite:///", "")) or ".", exist_ok=True) + mlflow.set_tracking_uri(cfg.mlflow_uri) + mlflow.set_experiment(cfg.http_mlflow_experiment) + with mlflow.start_run(run_name="http_attempt"): + mlflow.log_metrics({k: float(v) for k, v in metrics.items()}) + mlflow.set_tag("stage", "9") + mlflow.set_tag("model", cfg.http_mlflow_model) + + +def main(argv: list[str] | None = None) -> int: + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + parser = argparse.ArgumentParser(description="Train Stage 9 HTTP-attempt model") + parser.add_argument("--min-samples", type=int, default=40) + parser.add_argument("--hours", type=int, default=168) + parser.add_argument("--rollback", action="store_true") + args = parser.parse_args(argv) + cfg = MLConfig() + if args.rollback: + ok = rollback_artifact(cfg.http_model_path, cfg.http_model_prev_path) + print("rolled back" if ok else "no previous HTTP artifact") + return 0 if ok else 1 + metrics = train(cfg, min_samples=args.min_samples, hours=args.hours) + print(metrics) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/kernel_ai/services/network.py b/kernel_ai/services/network.py index 6753f32..685f1e7 100644 --- a/kernel_ai/services/network.py +++ b/kernel_ai/services/network.py @@ -277,7 +277,59 @@ def _parse_snmp_section(section_name): return {} -def _get_ss_tcp_metrics(): +def _endpoint_key(value): + return str(value or "").strip().lower() + + +def _parse_ss_detail_metrics(line, details): + metrics = {} + parts = line.split() + peer = parts[4] if len(parts) >= 5 else "" + local = parts[3] if len(parts) >= 5 else "" + if len(parts) >= 4: + try: + metrics["tx_queue"] = int(parts[2]) + metrics["rx_queue"] = int(parts[1]) + except ValueError: + pass + metrics["ss_local"] = local + metrics["ss_remote"] = peer + + rtt_match = re.search(r"rtt:(\d+(?:\.\d+)?)/", details) + cwnd_match = re.search(r"cwnd:(\d+)", details) + retrans_match = re.search(r"retrans:(\d+)(?:/\d+)?", details) + if rtt_match: + metrics["rtt_ms"] = float(rtt_match.group(1)) + if cwnd_match: + metrics["cwnd"] = int(cwnd_match.group(1)) + if retrans_match: + metrics["retrans_now"] = int(retrans_match.group(1)) + mss_match = re.search(r"\bmss:(\d+)", details) + if mss_match: + metrics["mss"] = int(mss_match.group(1)) + + stripped = details.strip() + if stripped: + cc = stripped.split(" ", 1)[0] + if cc and re.match(r"^[a-z][a-z0-9_]+$", cc): + metrics["cc"] = cc + dr = re.search(r"delivery_rate\s+([\d.]+)([KMG]?)bps", details) + if dr: + metrics["delivery_rate_mbps"] = round(_rate_to_mbps(dr.group(1), dr.group(2)), 4) + pr = re.search(r"pacing_rate\s+([\d.]+)([KMG]?)bps", details) + if pr: + metrics["pacing_rate_mbps"] = round(_rate_to_mbps(pr.group(1), pr.group(2)), 4) + minrtt = re.search(r"minrtt:([\d.]+)", details) + if minrtt: + metrics["min_rtt_ms"] = float(minrtt.group(1)) + bbr = re.search(r"bbr:\(bw:([\d.]+)([KMG]?)bps,mrtt:([\d.]+)", details) + if bbr: + metrics["bbr_bw_mbps"] = round(_rate_to_mbps(bbr.group(1), bbr.group(2)), 4) + metrics["bbr_mrtt_ms"] = float(bbr.group(3)) + return metrics + + +def _get_ss_tcp_metrics(local=None, remote=None): ss_cmd = resolve_binary("ss") if not ss_cmd: return {} @@ -287,63 +339,28 @@ def _get_ss_tcp_metrics(): except (subprocess.TimeoutExpired, OSError): return {} + want_local = _endpoint_key(local) + want_remote = _endpoint_key(remote) fallback = {} for idx, line in enumerate(lines): if not line.strip().startswith("ESTAB"): continue - metrics = {} - parts = line.split() - # Prefer a real remote peer over loopback/internal so the BBR model - # reflects an actual path, not the ~0.02ms/4Gbps loopback socket. - peer = parts[4] if len(parts) >= 5 else "" - is_local = peer.startswith("127.") or peer.startswith("[::1]") or peer.startswith("0.0.0.0") - if len(parts) >= 4: - try: - metrics["tx_queue"] = int(parts[2]) - metrics["rx_queue"] = int(parts[1]) - except ValueError: - pass - details = lines[idx + 1] if (idx + 1) < len(lines) else "" - rtt_match = re.search(r"rtt:(\d+(?:\.\d+)?)/", details) - cwnd_match = re.search(r"cwnd:(\d+)", details) - retrans_match = re.search(r"retrans:(\d+)(?:/\d+)?", details) - if rtt_match: - metrics["rtt_ms"] = float(rtt_match.group(1)) - if cwnd_match: - metrics["cwnd"] = int(cwnd_match.group(1)) - if retrans_match: - metrics["retrans_now"] = int(retrans_match.group(1)) - mss_match = re.search(r"\bmss:(\d+)", details) - if mss_match: - metrics["mss"] = int(mss_match.group(1)) - - # Congestion-control algorithm (first token of the info line), plus the - # BBR model ingredients: delivery rate (BtlBw estimate), min RTT - # (RTprop), pacing rate, and — when BBR is the CC — its own bw/mrtt. - stripped = details.strip() - if stripped: - cc = stripped.split(" ", 1)[0] - if cc and re.match(r"^[a-z][a-z0-9_]+$", cc): - metrics["cc"] = cc - dr = re.search(r"delivery_rate\s+([\d.]+)([KMG]?)bps", details) - if dr: - metrics["delivery_rate_mbps"] = round(_rate_to_mbps(dr.group(1), dr.group(2)), 4) - pr = re.search(r"pacing_rate\s+([\d.]+)([KMG]?)bps", details) - if pr: - metrics["pacing_rate_mbps"] = round(_rate_to_mbps(pr.group(1), pr.group(2)), 4) - minrtt = re.search(r"minrtt:([\d.]+)", details) - if minrtt: - metrics["min_rtt_ms"] = float(minrtt.group(1)) - bbr = re.search(r"bbr:\(bw:([\d.]+)([KMG]?)bps,mrtt:([\d.]+)", details) - if bbr: - metrics["bbr_bw_mbps"] = round(_rate_to_mbps(bbr.group(1), bbr.group(2)), 4) - metrics["bbr_mrtt_ms"] = float(bbr.group(3)) - if metrics: - if not is_local: + metrics = _parse_ss_detail_metrics(line, details) + if not metrics: + continue + peer = str(metrics.get("ss_remote") or "") + is_loop = peer.startswith("127.") or peer.startswith("[::1]") or peer.startswith("0.0.0.0") + if want_local and want_remote: + if _endpoint_key(metrics.get("ss_local")) == want_local and _endpoint_key(peer) == want_remote: return metrics if not fallback: fallback = metrics + continue + if not is_loop: + return metrics + if not fallback: + fallback = metrics return fallback @@ -512,7 +529,7 @@ def get_ip_layer_map(limit_routes: int = 12, limit_neigh: int = 10) -> dict: } -def get_network_stack_realtime(network_stack_prev=None): +def get_network_stack_realtime(network_stack_prev=None, prefer_local=None, prefer_remote=None): network_stack_prev = _NETWORK_STACK_PREV_DEFAULT if network_stack_prev is None else network_stack_prev now = time.time() iface = _get_default_iface() @@ -520,26 +537,47 @@ def get_network_stack_realtime(network_stack_prev=None): iface_stats = pernic.get(iface) all_connections = get_active_connections() interesting = [c for c in all_connections if not c["remote"].startswith("127.0.0.1") and not c["remote"].startswith("0.0.0.0")] - flow = interesting[0] if interesting else (all_connections[0] if all_connections else None) + want_local = _endpoint_key(prefer_local) + want_remote = _endpoint_key(prefer_remote) + pinned = None + if want_remote: + for conn in all_connections: + remote_ok = _endpoint_key(conn.get("remote")) == want_remote + local_ok = (not want_local) or _endpoint_key(conn.get("local")) == want_local + if remote_ok and local_ok: + pinned = conn + break socket_example = _get_socket_example(all_connections) - if flow: + # One sock: prefer the pinned 4-tuple, else the example with fd/pid, else first flow. + raw_flow = pinned or socket_example or (interesting[0] if interesting else (all_connections[0] if all_connections else None)) + flow = None + if raw_flow: + same_example = ( + socket_example + and _endpoint_key(socket_example.get("local")) == _endpoint_key(raw_flow.get("local")) + and _endpoint_key(socket_example.get("remote")) == _endpoint_key(raw_flow.get("remote")) + ) + owner = socket_example if same_example else raw_flow flow = { - "local": flow.get("local"), - "remote": flow.get("remote"), - "type": str(flow.get("type", "TCP")).upper(), - "state_code": flow.get("state", "00"), - "state_name": _tcp_state_name(flow.get("state", "00")), - "inode": int(flow.get("inode") or 0), - "uid": int(flow.get("uid") or 0), - "fd": (socket_example or {}).get("fd"), - "pid": (socket_example or {}).get("pid"), - "process": (socket_example or {}).get("process"), + "local": raw_flow.get("local"), + "remote": raw_flow.get("remote"), + "type": str(raw_flow.get("type", "TCP")).upper(), + "state_code": raw_flow.get("state") or raw_flow.get("state_code") or "00", + "state_name": raw_flow.get("state_name") or _tcp_state_name(raw_flow.get("state", "00")), + "inode": int(raw_flow.get("inode") or 0), + "uid": int(raw_flow.get("uid") or 0), + "fd": owner.get("fd") if isinstance(owner, dict) else None, + "pid": owner.get("pid") if isinstance(owner, dict) else None, + "process": owner.get("process") if isinstance(owner, dict) else None, } tcpext = _parse_netstat_tcpext() ip_stats = _parse_snmp_section("Ip") tcp_stats = _parse_snmp_section("Tcp") - ss_metrics = _get_ss_tcp_metrics() + ss_metrics = _get_ss_tcp_metrics( + local=(flow or {}).get("local"), + remote=(flow or {}).get("remote"), + ) conntrack = _get_conntrack_stats() retrans_total = tcpext.get("RetransSegs", 0) @@ -624,6 +662,7 @@ def rate(curr, prev): "cwnd": int(ss_metrics.get("cwnd", 0)), "rtt_ms": round(float(ss_metrics.get("rtt_ms", 0.0)), 2), "tx_queue": int(ss_metrics.get("tx_queue", 0)), + "rx_queue": int(ss_metrics.get("rx_queue", 0)), "cc": ss_metrics.get("cc", "unknown"), "delivery_rate_mbps": ss_metrics.get("delivery_rate_mbps", 0.0), "min_rtt_ms": round(float(ss_metrics.get("min_rtt_ms", ss_metrics.get("rtt_ms", 0.0))), 3),