Add dashboard integrity/watchdog console

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>
This commit is contained in:
Luna 2026-06-18 17:39:53 -07:00
parent e0e5cd62d9
commit 11d91a40b4
12 changed files with 337 additions and 29 deletions

View file

@ -303,6 +303,120 @@ def rules_catalog(cfg: Config) -> dict:
}
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):
@ -368,6 +482,8 @@ class _Handler(BaseHTTPRequestHandler):
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/"):