11d91a40b4
Surface integrity and watchdog state in the read-only dashboard via a new /api/integrity endpoint and integrity_report(): watchdog/heartbeat verdict, FIM baseline and package-DB anchor freshness, pacman keyring and SigLevel posture, and Sentinel's own self-integrity footprint. The endpoint summarizes existing anchors and heartbeats only — it runs no live FIM or package verification from the request path, keeping the dashboard read-only. Add the enodia.integrity.v1 schema (schemas.py + docs/SCHEMAS.md) with a contract test, a new "Integrity" dashboard tab and summary metric, and web tests for the report shape, schema contract, endpoint, and console wiring. Docs updated; completes the v1.0 "dashboard to local console" roadmap item. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
542 lines
18 KiB
Python
542 lines
18 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Read-only HTTPS web dashboard, served by the stdlib http.server (zero deps).
|
|
|
|
Exposes a small JSON API over the log directory plus a single self-contained
|
|
page. It is deliberately read-only — no actions, no writes — because it serves
|
|
sensitive forensic data. TLS is mandatory, using configured certificates or an
|
|
auto-generated self-signed certificate under ``log_dir``. By default it binds
|
|
the host's Tailscale address and requires a bearer token; on loopback the token
|
|
is optional.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import ssl
|
|
import subprocess
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, unquote, urlparse
|
|
|
|
from . import __version__
|
|
from . import schemas
|
|
from .config import Config
|
|
|
|
_STATIC = Path(__file__).parent / "static"
|
|
_ALERT_NAME = re.compile(r"^alert-\d{8}-\d{6}\.(json|log)$")
|
|
|
|
|
|
# --- network helpers -------------------------------------------------------
|
|
|
|
def detect_tailscale_ip() -> str | None:
|
|
"""Best-effort Tailscale IPv4 of this host."""
|
|
try:
|
|
out = subprocess.run(["tailscale", "ip", "-4"], capture_output=True,
|
|
text=True, timeout=4).stdout.strip().splitlines()
|
|
if out and out[0]:
|
|
return out[0].strip()
|
|
except (OSError, subprocess.SubprocessError):
|
|
pass
|
|
try:
|
|
out = subprocess.run(["ip", "-o", "-4", "addr", "show", "tailscale0"],
|
|
capture_output=True, text=True, timeout=4).stdout
|
|
m = re.search(r"inet (\d+\.\d+\.\d+\.\d+)", out)
|
|
if m:
|
|
return m.group(1)
|
|
except (OSError, subprocess.SubprocessError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def resolve_bind(cfg: Config) -> str:
|
|
if cfg.web_bind:
|
|
return cfg.web_bind
|
|
return detect_tailscale_ip() or "127.0.0.1"
|
|
|
|
|
|
def is_loopback(addr: str) -> bool:
|
|
return addr.startswith("127.") or addr in ("::1", "localhost")
|
|
|
|
|
|
def ensure_token(cfg: Config) -> str:
|
|
if cfg.web_token:
|
|
return cfg.web_token
|
|
path = cfg.log_dir / ".web_token"
|
|
try:
|
|
if path.is_file():
|
|
tok = path.read_text().strip()
|
|
if tok:
|
|
return tok
|
|
except OSError:
|
|
pass
|
|
tok = secrets.token_urlsafe(24)
|
|
try:
|
|
cfg.log_dir.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(tok)
|
|
path.chmod(0o600)
|
|
except OSError:
|
|
pass
|
|
return tok
|
|
|
|
|
|
def tls_paths(cfg: Config) -> tuple[Path, Path]:
|
|
cert = Path(cfg.web_tls_cert) if cfg.web_tls_cert else cfg.log_dir / "web-selfsigned.crt"
|
|
key = Path(cfg.web_tls_key) if cfg.web_tls_key else cfg.log_dir / "web-selfsigned.key"
|
|
return cert, key
|
|
|
|
|
|
def _san_for(bind: str) -> str:
|
|
names = ["DNS:localhost", "IP:127.0.0.1"]
|
|
host = os.uname().nodename
|
|
if host:
|
|
names.append(f"DNS:{host}")
|
|
try:
|
|
ipaddress.ip_address(bind)
|
|
names.append(f"IP:{bind}")
|
|
except ValueError:
|
|
if bind and bind != "localhost":
|
|
names.append(f"DNS:{bind}")
|
|
return ",".join(dict.fromkeys(names))
|
|
|
|
|
|
def ensure_tls_cert(cfg: Config, bind: str) -> tuple[Path, Path]:
|
|
"""Return usable cert/key paths, generating a self-signed pair if absent."""
|
|
cert, key = tls_paths(cfg)
|
|
if cert.is_file() and key.is_file():
|
|
return cert, key
|
|
cert.parent.mkdir(parents=True, exist_ok=True)
|
|
key.parent.mkdir(parents=True, exist_ok=True)
|
|
subj = f"/CN={bind if bind else 'localhost'}"
|
|
cmd = [
|
|
"openssl", "req", "-x509", "-newkey", "rsa:3072", "-nodes",
|
|
"-sha256", "-days", "365",
|
|
"-keyout", str(key), "-out", str(cert),
|
|
"-subj", subj,
|
|
"-addext", f"subjectAltName={_san_for(bind)}",
|
|
]
|
|
try:
|
|
subprocess.run(cmd, capture_output=True, text=True, timeout=20, check=True)
|
|
key.chmod(0o600)
|
|
cert.chmod(0o644)
|
|
except (OSError, subprocess.SubprocessError) as exc:
|
|
raise RuntimeError(
|
|
"TLS is mandatory for the web dashboard, but no certificate/key "
|
|
"were available and self-signed generation with openssl failed"
|
|
) from exc
|
|
return cert, key
|
|
|
|
|
|
# --- data layer (pure, testable) ------------------------------------------
|
|
|
|
def list_alerts(cfg: Config, limit: int = 200) -> list[dict]:
|
|
out = []
|
|
for p in sorted(cfg.log_dir.glob("alert-*.json"), reverse=True)[:limit]:
|
|
try:
|
|
d = json.loads(p.read_text())
|
|
except (OSError, ValueError):
|
|
continue
|
|
out.append({
|
|
"name": p.with_suffix(".log").name,
|
|
"time": d.get("time", ""),
|
|
"host": d.get("host", ""),
|
|
"severity": d.get("severity", ""),
|
|
"signatures": sorted({a.get("signature", "") for a in d.get("alerts", [])}),
|
|
"sids": sorted({a.get("sid", 0) for a in d.get("alerts", []) if a.get("sid")}),
|
|
"count": len(d.get("alerts", [])),
|
|
})
|
|
return out
|
|
|
|
|
|
def get_alert(cfg: Config, name: str) -> dict | None:
|
|
if not _ALERT_NAME.match(name):
|
|
return None
|
|
base = cfg.log_dir / name
|
|
jpath = base.with_suffix(".json")
|
|
try:
|
|
data = {"json": json.loads(jpath.read_text())} if jpath.is_file() else {}
|
|
lpath = base.with_suffix(".log")
|
|
if lpath.is_file():
|
|
data["text"] = lpath.read_text()
|
|
return data or None
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def tail_events(cfg: Config, limit: int = 100) -> list[str]:
|
|
try:
|
|
lines = cfg.events_log.read_text().splitlines()
|
|
except OSError:
|
|
return []
|
|
return lines[-limit:]
|
|
|
|
|
|
def daemon_status(cfg: Config) -> dict:
|
|
pid_alive = False
|
|
pidfile = cfg.log_dir / "sentinel.pid"
|
|
try:
|
|
pid = int(pidfile.read_text().strip())
|
|
os.kill(pid, 0)
|
|
pid_alive = True
|
|
except (OSError, ValueError):
|
|
pid_alive = False
|
|
alerts = list_alerts(cfg, limit=10000)
|
|
counts: dict[str, int] = {}
|
|
for a in alerts:
|
|
counts[a["severity"]] = counts.get(a["severity"], 0) + 1
|
|
ebpf_exec = "unknown"
|
|
ebpf_syscall = "unknown"
|
|
for line in reversed(tail_events(cfg, 200)):
|
|
if ebpf_exec == "unknown" and "eBPF exec monitor:" in line:
|
|
ebpf_exec = line.split("eBPF exec monitor:", 1)[1].strip()
|
|
if ebpf_syscall == "unknown" and "eBPF syscall monitor:" in line:
|
|
ebpf_syscall = line.split("eBPF syscall monitor:", 1)[1].strip()
|
|
if ebpf_exec != "unknown" and ebpf_syscall != "unknown":
|
|
break
|
|
from .selfprotect import heartbeat_age
|
|
age = heartbeat_age(cfg)
|
|
return {
|
|
"schema": schemas.STATUS_V1,
|
|
"version": __version__,
|
|
"running": pid_alive,
|
|
"total_alerts": len(alerts),
|
|
"counts": counts,
|
|
"last_alert": alerts[0]["time"] if alerts else None,
|
|
"ebpf": ebpf_exec,
|
|
"ebpf_exec": ebpf_exec,
|
|
"ebpf_syscall": ebpf_syscall,
|
|
"host": os.uname().nodename,
|
|
"heartbeat_age": age,
|
|
"heartbeat_stale": (age is not None and age > cfg.heartbeat_max_age),
|
|
}
|
|
|
|
|
|
def list_incidents(cfg: Config) -> list[dict]:
|
|
from . import incident
|
|
|
|
return sorted(incident.load_index(cfg).values(),
|
|
key=lambda i: i.get("last_ts", 0.0), reverse=True)
|
|
|
|
|
|
def get_incident(cfg: Config, incident_id: str) -> dict | None:
|
|
from . import incident
|
|
|
|
inc = incident.load_index(cfg).get(incident_id)
|
|
if inc is None:
|
|
return None
|
|
snapshots = []
|
|
timeline = []
|
|
for name in inc.get("snapshots", []):
|
|
path = (cfg.log_dir / name).with_suffix(".json")
|
|
try:
|
|
report = json.loads(path.read_text())
|
|
except (OSError, ValueError):
|
|
timeline.append({"snapshot": name, "time": "?", "missing": True})
|
|
continue
|
|
snapshots.append(report)
|
|
timeline.append({
|
|
"snapshot": name,
|
|
"time": report.get("time", "?"),
|
|
"severity": report.get("severity", "?"),
|
|
"signatures": list(dict.fromkeys(
|
|
a.get("signature", "") for a in report.get("alerts", []))),
|
|
"pids": [p.get("pid") for p in report.get("processes", [])
|
|
if isinstance(p.get("pid"), int)],
|
|
})
|
|
timeline.sort(key=lambda r: r.get("time", ""))
|
|
return {
|
|
"schema": schemas.INCIDENT_VIEW_V1,
|
|
"incident": inc,
|
|
"timeline": timeline,
|
|
"snapshots": snapshots,
|
|
}
|
|
|
|
|
|
def response_plan(cfg: Config, incident_id: str) -> dict | None:
|
|
from . import respond
|
|
|
|
bundle = respond.load_bundle(cfg, incident_id)
|
|
if bundle is None:
|
|
return None
|
|
return respond.build_plan(bundle, cfg)
|
|
|
|
|
|
def posture_report(cfg: Config, runner=None) -> dict:
|
|
"""Return advisory host posture findings for the management console."""
|
|
if runner is None:
|
|
from . import posture
|
|
runner = posture.run
|
|
findings = sorted(runner(cfg), key=lambda a: (-a.severity, a.signature))
|
|
counts: dict[str, int] = {}
|
|
for f in findings:
|
|
sev = str(f.severity)
|
|
counts[sev] = counts.get(sev, 0) + 1
|
|
return {
|
|
"count": len(findings),
|
|
"counts": counts,
|
|
"findings": [f.to_dict() for f in findings],
|
|
}
|
|
|
|
|
|
def rules_catalog(cfg: Config) -> dict:
|
|
"""Return active event-rule metadata for the management console."""
|
|
from . import ruleops
|
|
|
|
rules = ruleops.list_rules(cfg)
|
|
by_event: dict[str, int] = {}
|
|
by_severity: dict[str, int] = {}
|
|
by_origin: dict[str, int] = {}
|
|
for r in rules:
|
|
by_event[r["event"]] = by_event.get(r["event"], 0) + 1
|
|
by_severity[r["severity"]] = by_severity.get(r["severity"], 0) + 1
|
|
by_origin[r["origin"]] = by_origin.get(r["origin"], 0) + 1
|
|
return {
|
|
"count": len(rules),
|
|
"by_event": by_event,
|
|
"by_severity": by_severity,
|
|
"by_origin": by_origin,
|
|
"rules": rules,
|
|
}
|
|
|
|
|
|
def _json_file_state(path: Path, now: float) -> dict:
|
|
try:
|
|
st = path.stat()
|
|
except OSError:
|
|
return {
|
|
"path": str(path),
|
|
"exists": False,
|
|
"status": "missing",
|
|
"mtime": None,
|
|
"age": None,
|
|
"count": None,
|
|
}
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
except (OSError, ValueError):
|
|
return {
|
|
"path": str(path),
|
|
"exists": True,
|
|
"status": "unreadable",
|
|
"mtime": st.st_mtime,
|
|
"age": max(0.0, now - st.st_mtime),
|
|
"count": None,
|
|
}
|
|
count = len(data) if isinstance(data, dict) else None
|
|
return {
|
|
"path": str(path),
|
|
"exists": True,
|
|
"status": "ok",
|
|
"mtime": st.st_mtime,
|
|
"age": max(0.0, now - st.st_mtime),
|
|
"count": count,
|
|
}
|
|
|
|
|
|
def integrity_report(cfg: Config, status: dict | None = None,
|
|
now: float | None = None) -> dict:
|
|
"""Return dashboard-readable integrity and watchdog state.
|
|
|
|
This is intentionally a summary of existing anchors and heartbeats. It does
|
|
not run live FIM or package verification work from the web request path.
|
|
"""
|
|
from . import pkgdb
|
|
from .selfprotect import SELF_PATHS, heartbeat_path, watchdog_verdict
|
|
|
|
ts = time.time() if now is None else now
|
|
status = status or daemon_status(cfg)
|
|
watchdog_ok, watchdog_message = watchdog_verdict(status, cfg.heartbeat_max_age)
|
|
anchor = pkgdb.load_anchor(cfg)
|
|
anchor_path = pkgdb._anchor_path(cfg)
|
|
anchor_time = anchor.get("time") if isinstance(anchor, dict) else None
|
|
sig_alert = pkgdb.siglevel_alert()
|
|
watched = []
|
|
for raw in SELF_PATHS:
|
|
path = Path(raw)
|
|
watched.append({"path": raw, "exists": path.exists()})
|
|
present = sum(1 for item in watched if item["exists"])
|
|
fim_state = _json_file_state(cfg.fim_baseline, ts)
|
|
pkg_state = {
|
|
"path": str(anchor_path),
|
|
"exists": bool(anchor),
|
|
"status": "ok" if anchor else "missing",
|
|
"time": anchor_time,
|
|
"age": max(0.0, ts - anchor_time) if isinstance(anchor_time, (int, float)) else None,
|
|
"fingerprint_prefix": (
|
|
str(anchor.get("fingerprint", ""))[:16]
|
|
if isinstance(anchor, dict) else ""
|
|
),
|
|
}
|
|
keyring_present = pkgdb.keyring_present()
|
|
checks = {
|
|
"watchdog": "ok" if watchdog_ok else "review",
|
|
"fim_baseline": fim_state["status"],
|
|
"pkgdb_anchor": pkg_state["status"],
|
|
"pacman_siglevel": "review" if sig_alert else "ok",
|
|
"pacman_keyring": "ok" if keyring_present else "missing",
|
|
}
|
|
overall = "ok"
|
|
if any(v in ("review", "missing", "unreadable") for v in checks.values()):
|
|
overall = "review"
|
|
if not watchdog_ok:
|
|
overall = "critical"
|
|
return {
|
|
"schema": schemas.INTEGRITY_V1,
|
|
"generated_at": ts,
|
|
"status": overall,
|
|
"checks": checks,
|
|
"watchdog": {
|
|
"ok": watchdog_ok,
|
|
"message": watchdog_message,
|
|
"heartbeat_path": str(heartbeat_path(cfg)),
|
|
"heartbeat_age": status.get("heartbeat_age"),
|
|
"heartbeat_stale": status.get("heartbeat_stale", False),
|
|
"max_age": cfg.heartbeat_max_age,
|
|
"daemon_running": status.get("running", False),
|
|
},
|
|
"anchors": {
|
|
"fim_baseline": fim_state,
|
|
"pkgdb": pkg_state,
|
|
"pacman": {
|
|
"keyring_present": keyring_present,
|
|
"siglevel_status": "review" if sig_alert else "ok",
|
|
"siglevel_alert": sig_alert.to_dict() if sig_alert else None,
|
|
},
|
|
},
|
|
"sentinel_footprint": {
|
|
"configured": len(watched),
|
|
"present": present,
|
|
"missing": len(watched) - present,
|
|
"paths": watched,
|
|
},
|
|
"read_only": True,
|
|
}
|
|
|
|
|
|
# --- HTTP layer ------------------------------------------------------------
|
|
|
|
class _Handler(BaseHTTPRequestHandler):
|
|
server_version = "EnodiaSentinel"
|
|
|
|
def log_message(self, *a): # quiet by default
|
|
pass
|
|
|
|
def _authorized(self) -> bool:
|
|
token = self.server.token # type: ignore[attr-defined]
|
|
if not token:
|
|
return True
|
|
supplied = ""
|
|
auth = self.headers.get("Authorization", "")
|
|
if auth.startswith("Bearer "):
|
|
supplied = auth[7:]
|
|
if not supplied:
|
|
supplied = parse_qs(urlparse(self.path).query).get("token", [""])[0]
|
|
return hmac.compare_digest(supplied, token)
|
|
|
|
def _send_body(self, body: bytes, ctype: str, status=200, send_body=True):
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
if send_body:
|
|
self.wfile.write(body)
|
|
|
|
def _json(self, obj, status=200, send_body=True):
|
|
body = json.dumps(obj).encode("utf-8")
|
|
self._send_body(body, "application/json", status, send_body)
|
|
|
|
def _json_or_not_found(self, obj, send_body=True):
|
|
self._json(obj if obj else {"error": "not found"},
|
|
200 if obj else 404, send_body=send_body)
|
|
|
|
def do_GET(self):
|
|
self._handle_get(send_body=True)
|
|
|
|
def do_HEAD(self):
|
|
self._handle_get(send_body=False)
|
|
|
|
def _handle_get(self, send_body=True):
|
|
path = urlparse(self.path).path
|
|
if not self._authorized():
|
|
self._json({"error": "unauthorized"}, 401, send_body=send_body)
|
|
return
|
|
cfg: Config = self.server.cfg # type: ignore[attr-defined]
|
|
|
|
if path in ("/", "/index.html", "/dashboard.html"):
|
|
self._serve_static("dashboard.html", "text/html; charset=utf-8",
|
|
send_body=send_body)
|
|
elif path == "/api/status":
|
|
self._json(daemon_status(cfg), send_body=send_body)
|
|
elif path == "/api/alerts":
|
|
self._json(list_alerts(cfg), send_body=send_body)
|
|
elif path.startswith("/api/alerts/"):
|
|
alert = get_alert(cfg, path.rsplit("/", 1)[-1])
|
|
self._json_or_not_found(alert, send_body=send_body)
|
|
elif path == "/api/events":
|
|
self._json({"events": tail_events(cfg)}, send_body=send_body)
|
|
elif path == "/api/posture":
|
|
self._json(posture_report(cfg), send_body=send_body)
|
|
elif path == "/api/rules":
|
|
self._json(rules_catalog(cfg), send_body=send_body)
|
|
elif path == "/api/integrity":
|
|
self._json(integrity_report(cfg), send_body=send_body)
|
|
elif path == "/api/incidents":
|
|
self._json(list_incidents(cfg), send_body=send_body)
|
|
elif path.startswith("/api/incidents/"):
|
|
iid = unquote(path.rsplit("/", 1)[-1])
|
|
inc = get_incident(cfg, iid)
|
|
self._json_or_not_found(inc, send_body=send_body)
|
|
elif path.startswith("/api/respond/plan/"):
|
|
iid = unquote(path.rsplit("/", 1)[-1])
|
|
plan = response_plan(cfg, iid)
|
|
self._json_or_not_found(plan, send_body=send_body)
|
|
else:
|
|
self._json({"error": "not found"}, 404, send_body=send_body)
|
|
|
|
def _serve_static(self, name, ctype, send_body=True):
|
|
try:
|
|
body = (_STATIC / name).read_bytes()
|
|
except OSError:
|
|
self._json({"error": "missing asset"}, 500, send_body=send_body)
|
|
return
|
|
self._send_body(body, ctype, send_body=send_body)
|
|
|
|
|
|
def build_server(cfg: Config) -> tuple[ThreadingHTTPServer, str, str]:
|
|
"""Construct the HTTPS server. Returns (httpd, bind, token)."""
|
|
bind = resolve_bind(cfg)
|
|
token = cfg.web_token
|
|
if not is_loopback(bind) and not token:
|
|
token = ensure_token(cfg)
|
|
cert, key = ensure_tls_cert(cfg, bind)
|
|
httpd = ThreadingHTTPServer((bind, cfg.web_port), _Handler)
|
|
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
ctx.load_cert_chain(certfile=str(cert), keyfile=str(key))
|
|
httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)
|
|
httpd.cfg = cfg # type: ignore[attr-defined]
|
|
httpd.token = token # type: ignore[attr-defined]
|
|
httpd.tls_cert = cert # type: ignore[attr-defined]
|
|
return httpd, bind, token
|
|
|
|
|
|
def serve(cfg: Config) -> None:
|
|
httpd, bind, token = build_server(cfg)
|
|
suffix = f"/?token={token}" if token else "/"
|
|
url = f"https://{bind}:{cfg.web_port}{suffix}"
|
|
msg = (f"{time.strftime('%FT%T%z')} dashboard serving at {url} "
|
|
f"(TLS cert: {getattr(httpd, 'tls_cert', '')})")
|
|
print(msg, flush=True)
|
|
try:
|
|
with open(cfg.events_log, "a") as fh:
|
|
fh.write(msg + "\n")
|
|
except OSError:
|
|
pass
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
httpd.shutdown()
|