Spaces:
Running
Running
| """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"<span class='srcbadge {cls}'>{label}</span>" | |
| 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 "<div class='ph'>Inject an attack to inspect token surprise.</div>" | |
| 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"<span class='tok' title='surprise {nll:.1f}' " | |
| f"style='color:{color};{bg}'>{text}</span>" | |
| ) | |
| return "<div class='heat'>" + "".join(spans) + "</div>" | |
| 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"<span class='chip'>{score:.2f}</span>" if flagged | |
| else f"<span class='chip ok'>{score:.2f}</span>" | |
| ) | |
| tag = source_badge(r.get("source")) | |
| rows.append(f"<div class='{cls}'>{chip}<code>{html.escape(r['raw'])}</code>{tag}</div>") | |
| return "<div class='stream'>" + "".join(rows) + "</div>" | |
| 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"<polygon points='{' '.join(pts)}' fill='none' stroke='#2a2e33' stroke-width='1'/>") | |
| 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"<text x='{lx:.1f}' y='{ly:.1f}' fill='#8a9099' font-size='10' " | |
| f"font-family='JetBrains Mono, monospace' text-anchor='middle' " | |
| f"dominant-baseline='middle'>{_AXIS_LABEL[name]}</text>" | |
| ) | |
| poly = (f"<polygon points='{' '.join(value_pts)}' fill='rgba(226,35,26,0.35)' " | |
| f"stroke='{ACCENT}' stroke-width='1.5'/>") | |
| return ( | |
| f"<svg viewBox='0 0 {size} {size}' width='{size}' height='{size}'>" | |
| + "".join(grid) + poly + "".join(labels) + "</svg>" | |
| ) | |
| def analyst_html(explanation: dict[str, Any] | None, source: str | None = None) -> str: | |
| """Render the AI analyst verdict card.""" | |
| if not explanation: | |
| return "<div class='ph'>The analyst is idle. No anomaly in the stream.</div>" | |
| sev = str(explanation.get("severity", "unknown")).lower() | |
| fp = explanation.get("likely_false_positive") | |
| return ( | |
| f"<div class='verdict'>" | |
| f"<div class='sevline'><span class='sev sev-{sev}'>{sev.upper()}</span>" | |
| f"{source_badge(source)}</div>" | |
| f"<div class='vsummary'>{html.escape(str(explanation.get('summary','')))}</div>" | |
| f"<div class='vrow'><span>WHY</span>{html.escape(str(explanation.get('why','')))}</div>" | |
| f"<div class='vrow'><span>ACTION</span>{html.escape(str(explanation.get('next_action','')))}</div>" | |
| f"<div class='vrow'><span>FALSE POSITIVE</span>{'likely' if fp else 'no'}</div>" | |
| f"</div>" | |
| ) | |
| 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"<div class='livebar'><span class='{dot}'></span>" | |
| f"<b>{state}</b><span class='lsep'>·</span>connected to live source" | |
| f"<span class='lsep'>·</span>{clock}" | |
| f"<span class='lsep'>·</span>last event {last_ago}" | |
| f"<span class='lsep'>·</span>{n_events} live events</div>" | |
| ) | |
| def _drow(label: str, value: Any) -> str: | |
| return f"<div class='vrow'><span>{label}</span>{html.escape(str(value))}</div>" 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 = ["<div class='detail'>"] | |
| head = f"<div class='dhead'>{source_badge(record.get('source'))}" | |
| if sev: | |
| head += f"<span class='sev sev-{sev}'>{sev.upper()}</span>" | |
| head += f"<span class='dscore'>risk {record.get('global_score', 0)}</span></div>" | |
| out.append(head) | |
| out.append(f"<div class='draw'>{html.escape(record.get('raw', ''))}</div>") | |
| if explanation: | |
| out.append("<div class='dsec'>AI ANALYSIS</div>") | |
| 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("<div class='dsec'>WAZUH ALERT</div>") | |
| 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("<div class='dsec'>FULL LOG</div>" | |
| f"<pre class='dpre'>{html.escape(detail['full_log'])}</pre>") | |
| if detail.get("raw_json"): | |
| out.append("<div class='dsec'>RAW ALERT (JSON)</div>" | |
| f"<pre class='dpre'>{html.escape(detail['raw_json'])}</pre>") | |
| else: | |
| if record.get("reasons"): | |
| out.append("<div class='dsec'>WHY FLAGGED</div>") | |
| out += [f"<div class='vrow'>· {html.escape(r)}</div>" for r in record["reasons"]] | |
| axes = {k: v for k, v in (record.get("axes") or {}).items() if v} | |
| if axes: | |
| out.append("<div class='dsec'>RISK AXES</div>") | |
| out.append("<div class='vrow'>" + html.escape(", ".join(f"{k}:{v}" for k, v in axes.items())) + "</div>") | |
| if record.get("tokens"): | |
| out.append("<div class='dsec'>PERPLEXITY</div>" + heatmap_html(record["tokens"])) | |
| out.append("</div>") | |
| 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"<div class='metric'><div class='mval'>{v}</div><div class='mlbl'>{k}</div></div>" | |
| for k, v in cells | |
| ) | |
| return f"<div class='metrics'>{inner}</div>" | |