"""Pure render helpers for the TinySOC dark-console UI (HTML/SVG strings). No plotting dependency: the risk radar is hand-built SVG so it stays light on a free CPU Space and fully controllable for the dark theme. """ from __future__ import annotations import html import math from typing import Any ACCENT = "#E2231A" # Montreal red DIM = "#5a5f66" AMBER = "#e0a020" _AXES = ("time", "user", "process", "host", "ip") _AXIS_LABEL = {"time": "TIME", "user": "USER", "process": "PROC", "host": "HOST", "ip": "IP"} # Provenance of an event: label + CSS class (color). Honesty: an analyst must # always know whether a finding came from a test, an uploaded file, or live data. _SOURCE = { "demo": ("SAMPLE", "src-demo"), "injected": ("INJECTED · DEMO", "src-inject"), "uploaded": ("YOUR LOG", "src-upload"), "live": ("LIVE FEED · DEMO", "src-inject"), "live-wazuh": ("LIVE WAZUH", "src-live"), } def source_badge(source: str | None) -> str: label, cls = _SOURCE.get(source or "demo", _SOURCE["demo"]) return f"{label}" def _token_color(nll: float) -> tuple[str, float]: """Map a token NLL to (color, background-alpha).""" if nll >= 9: return ACCENT, 0.85 if nll >= 6: return ACCENT, 0.5 if nll >= 3: return AMBER, 0.35 return DIM, 0.0 def heatmap_html(tokens: list[tuple[str, float]] | None) -> str: """Render perplexity tokens as colored spans; brighter red = more surprising.""" if not tokens: return "
Inject an attack to inspect token surprise.
" spans = [] for tok, nll in tokens: color, alpha = _token_color(nll) text = html.escape(tok).replace(" ", " ") bg = f"background:rgba(226,35,26,{alpha});" if alpha else "" spans.append( f"{text}" ) return "
" + "".join(spans) + "
" def stream_html(records: list[dict[str, Any]]) -> str: """Render the log stream; flagged lines get a red rail and score chip.""" rows = [] for r in records: flagged = r.get("flagged") cls = "row flag" if flagged else "row" score = r.get("global_score", 0.0) chip = ( f"{score:.2f}" if flagged else f"{score:.2f}" ) tag = source_badge(r.get("source")) rows.append(f"
{chip}{html.escape(r['raw'])}{tag}
") return "
" + "".join(rows) + "
" def radar_svg(axes: dict[str, float] | None, size: int = 230) -> str: """Hand-built pentagon radar of the 5 risk axes (values 0..1).""" axes = axes or {} cx = cy = size / 2 radius = size * 0.36 n = len(_AXES) grid = [] for ring in (0.33, 0.66, 1.0): pts = [] for i in range(n): ang = -math.pi / 2 + 2 * math.pi * i / n pts.append(f"{cx + radius*ring*math.cos(ang):.1f},{cy + radius*ring*math.sin(ang):.1f}") grid.append(f"") labels, value_pts = [], [] for i, name in enumerate(_AXES): ang = -math.pi / 2 + 2 * math.pi * i / n val = max(0.0, min(1.0, float(axes.get(name, 0.0)))) value_pts.append(f"{cx + radius*val*math.cos(ang):.1f},{cy + radius*val*math.sin(ang):.1f}") lx, ly = cx + (radius + 16) * math.cos(ang), cy + (radius + 16) * math.sin(ang) labels.append( f"{_AXIS_LABEL[name]}" ) poly = (f"") return ( f"" + "".join(grid) + poly + "".join(labels) + "" ) def analyst_html(explanation: dict[str, Any] | None, source: str | None = None) -> str: """Render the AI analyst verdict card.""" if not explanation: return "
The analyst is idle. No anomaly in the stream.
" sev = str(explanation.get("severity", "unknown")).lower() fp = explanation.get("likely_false_positive") return ( f"
" f"
{sev.upper()}" f"{source_badge(source)}
" f"
{html.escape(str(explanation.get('summary','')))}
" f"
WHY{html.escape(str(explanation.get('why','')))}
" f"
ACTION{html.escape(str(explanation.get('next_action','')))}
" f"
FALSE POSITIVE{'likely' if fp else 'no'}
" f"
" ) def live_status_html(clock: str, last_ago: str, n_events: int, connected: bool = True) -> str: """Live connection status bar: pulsing dot + wall clock + last-event age.""" dot = "live-dot" if connected else "" state = "LIVE" if connected else "IDLE" return ( f"
" f"{state}·connected to live source" f"·{clock}" f"·last event {last_ago}" f"·{n_events} live events
" ) def _drow(label: str, value: Any) -> str: return f"
{label}{html.escape(str(value))}
" if value else "" def detail_html(record: dict[str, Any], explanation: dict[str, Any] | None = None) -> str: """Full 'detail page' for one alert: provenance, LLM analysis, all fields.""" detail = record.get("detail") or {} sev = str((explanation or {}).get("severity", "")).lower() out = ["
"] head = f"
{source_badge(record.get('source'))}" if sev: head += f"{sev.upper()}" head += f"risk {record.get('global_score', 0)}
" out.append(head) out.append(f"
{html.escape(record.get('raw', ''))}
") if explanation: out.append("
AI ANALYSIS
") out.append(_drow("SUMMARY", explanation.get("summary"))) out.append(_drow("WHY", explanation.get("why"))) out.append(_drow("NEXT ACTION", explanation.get("next_action"))) out.append(_drow("FALSE POSITIVE", "likely" if explanation.get("likely_false_positive") else "no")) if detail: out.append("
WAZUH ALERT
") out.append(_drow("TIME", detail.get("timestamp"))) out.append(_drow("HOST", detail.get("host"))) out.append(_drow("USER", detail.get("user"))) out.append(_drow("SRC IP", detail.get("src_ip"))) out.append(_drow("RULE ID", detail.get("rule_id"))) out.append(_drow("LEVEL", detail.get("level"))) out.append(_drow("DESCRIPTION", detail.get("description"))) out.append(_drow("GROUPS", ", ".join(detail.get("groups") or []))) mitre = detail.get("mitre") or {} if mitre.get("id"): out.append(_drow("MITRE", " ".join(mitre.get("id", [])) + " " + " ".join(mitre.get("technique", [])))) if detail.get("full_log"): out.append("
FULL LOG
" f"
{html.escape(detail['full_log'])}
") if detail.get("raw_json"): out.append("
RAW ALERT (JSON)
" f"
{html.escape(detail['raw_json'])}
") else: if record.get("reasons"): out.append("
WHY FLAGGED
") out += [f"
· {html.escape(r)}
" for r in record["reasons"]] axes = {k: v for k, v in (record.get("axes") or {}).items() if v} if axes: out.append("
RISK AXES
") out.append("
" + html.escape(", ".join(f"{k}:{v}" for k, v in axes.items())) + "
") if record.get("tokens"): out.append("
PERPLEXITY
" + heatmap_html(record["tokens"])) out.append("
") return "".join(out) def metrics_html(processed: int, anomalies: int, max_ppl: float, risk: float) -> str: cells = [ ("LINES", str(processed)), ("ANOMALIES", str(anomalies)), ("MAX PERPLEXITY", f"{max_ppl:.1f}"), ("RISK", f"{int(risk*100)}"), ] inner = "".join( f"
{v}
{k}
" for k, v in cells ) return f"
{inner}
"