From 2a990a5492fa1e01089ced31961a24f25469843c Mon Sep 17 00:00:00 2001 From: Aleksei Fedorov Date: Thu, 16 Jul 2026 12:50:38 +0000 Subject: [PATCH] SIEM service --- kernel_ai/services/siem.py | 268 +++++++++++++++++++++++++++++++++++++ 1 file changed, 268 insertions(+) create mode 100644 kernel_ai/services/siem.py diff --git a/kernel_ai/services/siem.py b/kernel_ai/services/siem.py new file mode 100644 index 0000000..e7407c8 --- /dev/null +++ b/kernel_ai/services/siem.py @@ -0,0 +1,268 @@ +"""Read-only bridge to the Elastic SIEM on the sibling server. + +The app server only *ships* logs to Elastic (Filebeat push). This module adds a +strictly read-only reverse channel: it queries the security detection alerts +index and normalizes matches into a compact shape the frontend can render as +"attack anomalies" on Kernel DNA. + +Design notes: + * No new PyPI dependency: uses urllib + ssl from the stdlib. + * Credentials/endpoint come from /etc/kernel-ai/elastic.env (API key scoped + read-only to .alerts-security.alerts-*). Falls back to os.environ. + * Short TTL cache so repeated page polls don't hammer Elastic. + * Fails soft: on any error returns {"available": False, ...} instead of + raising, so the visualization degrades gracefully. +""" + +from __future__ import annotations + +import json +import os +import ssl +import threading +import time +import urllib.request +from datetime import datetime, timezone + +_ENV_PATH = "/etc/kernel-ai/elastic.env" +_CACHE_TTL_SEC = 45.0 + +_cache_lock = threading.Lock() +_cache: dict = {} # key -> (expires_at, payload) +_config_cache: dict | None = None + +# MITRE ATT&CK tactic -> position along the helix (0..1), so scars cluster by +# attack stage. Unknown tactics fall back to a stable hash spread. +_TACTIC_POSITION = { + "reconnaissance": 0.06, + "resource development": 0.14, + "initial access": 0.30, + "execution": 0.46, + "persistence": 0.58, + "privilege escalation": 0.66, + "defense evasion": 0.72, + "credential access": 0.80, + "discovery": 0.20, + "lateral movement": 0.86, + "collection": 0.90, + "command and control": 0.94, + "exfiltration": 0.97, + "impact": 0.99, +} + + +def _load_config() -> dict: + """Read Elastic connection settings from the env file + environment.""" + global _config_cache + if _config_cache is not None: + return _config_cache + + cfg: dict = {} + try: + with open(_ENV_PATH, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, val = line.partition("=") + cfg[key.strip()] = val.strip().strip('"').strip("'") + except OSError: + pass + + def _get(name: str, default: str = "") -> str: + return os.environ.get(name, cfg.get(name, default)) + + resolved = { + "url": _get("KERNEL_AI_ES_URL").rstrip("/"), + "api_key": _get("KERNEL_AI_ES_API_KEY"), + "ca": _get("KERNEL_AI_ES_CA"), + "index": _get("KERNEL_AI_ES_ALERTS_INDEX", ".alerts-security.alerts-*"), + "tag": _get("KERNEL_AI_ES_TAG", "kernel-ai"), + } + _config_cache = resolved + return resolved + + +def _ssl_context(ca_path: str) -> ssl.SSLContext: + if ca_path and os.path.exists(ca_path): + return ssl.create_default_context(cafile=ca_path) + # No CA on disk: fall back to system trust (still verified). + return ssl.create_default_context() + + +def _es_search(cfg: dict, body: dict, timeout: float = 6.0) -> dict: + url = f"{cfg['url']}/{cfg['index']}/_search" + data = json.dumps(body).encode("utf-8") + req = urllib.request.Request(url, data=data, method="POST") + req.add_header("Content-Type", "application/json") + req.add_header("Authorization", f"ApiKey {cfg['api_key']}") + ctx = _ssl_context(cfg["ca"]) + with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: + return json.loads(resp.read().decode("utf-8")) + + +def _flat(src: dict, key: str): + """Alert docs store dotted keys flat (e.g. 'kibana.alert.severity').""" + return src.get(key) + + +def _extract_mitre(src: dict) -> dict: + threat = _flat(src, "kibana.alert.rule.threat") + tactic_name = tactic_id = tech_id = tech_name = None + if isinstance(threat, list) and threat: + first = threat[0] or {} + tactic = first.get("tactic") or {} + tactic_name = tactic.get("name") + tactic_id = tactic.get("id") + techs = first.get("technique") or [] + if techs: + tech_id = (techs[0] or {}).get("id") + tech_name = (techs[0] or {}).get("name") + return { + "tactic": tactic_name, + "tactic_id": tactic_id, + "technique": tech_id, + "technique_name": tech_name, + } + + +def _position_for(mitre: dict, rule: str, ip: str) -> float: + tactic = (mitre.get("tactic") or "").strip().lower() + if tactic in _TACTIC_POSITION: + return _TACTIC_POSITION[tactic] + # Stable spread for anything unmapped. + seed = f"{rule}|{ip}" + h = 0 + for ch in seed: + h = (h * 131 + ord(ch)) & 0xFFFFFFFF + return round((h % 1000) / 1000.0, 3) + + +def _severity_norm(sev) -> str: + s = str(sev or "").lower() + if s in ("critical", "high", "medium", "low"): + return s + return "medium" + + +def _normalize_hit(hit: dict) -> dict: + src = hit.get("_source", {}) or {} + mitre = _extract_mitre(src) + rule = _flat(src, "kibana.alert.rule.name") or "web attack" + ip = _flat(src, "source.ip") or "" + sev = _severity_norm(_flat(src, "kibana.alert.severity")) + query = _flat(src, "url.query") + if isinstance(query, str) and len(query) > 140: + query = query[:139] + "\u2026" + return { + "id": hit.get("_id"), + "time": _flat(src, "@timestamp"), + "rule": rule, + "severity": sev, + "risk": _flat(src, "kibana.alert.risk_score"), + "source_ip": ip, + "url_path": _flat(src, "url.path"), + "url_query": query, + "method": _flat(src, "http.request.method"), + "status": _flat(src, "http.response.status_code"), + "dataset": _flat(src, "event.dataset"), + "tactic": mitre["tactic"], + "tactic_id": mitre["tactic_id"], + "technique": mitre["technique"], + "technique_name": mitre["technique_name"], + "position": _position_for(mitre, rule, ip), + } + + +def _empty(hours: int, reason: str = "") -> dict: + return { + "available": False, + "timestamp": datetime.now(timezone.utc).isoformat(), + "window_hours": hours, + "total": 0, + "count": 0, + "by_severity": {}, + "by_tactic": {}, + "alerts": [], + "reason": reason, + } + + +def get_siem_alerts(hours: int = 24, limit: int = 120) -> dict: + """Return recent SIEM detection alerts, normalized and cached. + + Fails soft: returns available=False on any misconfiguration or error. + """ + hours = max(1, min(int(hours), 24 * 30)) + limit = max(1, min(int(limit), 500)) + cache_key = (hours, limit) + + now = time.monotonic() + with _cache_lock: + entry = _cache.get(cache_key) + if entry and entry[0] > now: + return entry[1] + + cfg = _load_config() + if not cfg["url"] or not cfg["api_key"]: + return _empty(hours, "elastic not configured") + + body = { + "size": limit, + "track_total_hits": True, + "sort": [{"@timestamp": {"order": "desc"}}], + "query": { + "bool": { + "filter": [ + {"range": {"@timestamp": {"gte": f"now-{hours}h"}}}, + {"term": {"kibana.alert.rule.tags": cfg["tag"]}}, + ] + } + }, + "_source": [ + "@timestamp", + "kibana.alert.rule.name", + "kibana.alert.severity", + "kibana.alert.risk_score", + "kibana.alert.rule.threat", + "source.ip", + "url.path", + "url.query", + "http.request.method", + "http.response.status_code", + "event.dataset", + ], + } + + try: + res = _es_search(cfg, body) + except Exception as exc: # noqa: BLE001 - fail soft, never break the page + return _empty(hours, f"query failed: {type(exc).__name__}") + + hits = (res.get("hits", {}) or {}).get("hits", []) or [] + total_obj = (res.get("hits", {}) or {}).get("total", {}) or {} + total = total_obj.get("value", len(hits)) if isinstance(total_obj, dict) else len(hits) + + alerts = [_normalize_hit(h) for h in hits] + + by_sev: dict = {} + by_tactic: dict = {} + for a in alerts: + by_sev[a["severity"]] = by_sev.get(a["severity"], 0) + 1 + t = a["tactic"] or "unmapped" + by_tactic[t] = by_tactic.get(t, 0) + 1 + + payload = { + "available": True, + "timestamp": datetime.now(timezone.utc).isoformat(), + "window_hours": hours, + "total": total, + "count": len(alerts), + "by_severity": by_sev, + "by_tactic": by_tactic, + "alerts": alerts, + } + + with _cache_lock: + _cache[cache_key] = (now + _CACHE_TTL_SEC, payload) + return payload