"""SQLite persistence for audit domains and checks.""" from __future__ import annotations import json import sqlite3 from datetime import datetime, timezone from typing import Any from app.collectors.base import CHECK_LABELS ONBOARD_DOMAIN_EVENTS = frozenset({"account.created", "onboarding.completed"}) TENANT_ONBOARD = 1 STATUS_RANK = {"pass": 0, "skip": 1, "warn": 2, "error": 3, "fail": 4} def _now() -> str: return datetime.now(timezone.utc).isoformat() def _parse_payload(raw: str | None) -> dict: if not raw: return {} try: return json.loads(raw) except json.JSONDecodeError: return {} def init_audit_schema(conn: sqlite3.Connection) -> None: conn.executescript(""" CREATE TABLE IF NOT EXISTS audit_domains ( id INTEGER PRIMARY KEY, tenant_id INTEGER NOT NULL, domain TEXT NOT NULL, source TEXT NOT NULL DEFAULT 'onboarding', created_at TEXT NOT NULL, UNIQUE(tenant_id, domain) ); CREATE TABLE IF NOT EXISTS audit_checks ( id INTEGER PRIMARY KEY, tenant_id INTEGER NOT NULL, domain TEXT NOT NULL, check_id TEXT NOT NULL, status TEXT NOT NULL, message TEXT, evidence TEXT, checked_at TEXT NOT NULL, UNIQUE(tenant_id, domain, check_id) ); """) def sync_domains_from_webhooks(conn: sqlite3.Connection) -> int: rows = conn.execute( """ SELECT event_type, payload FROM webhook_events WHERE source = 'vm112-onboard' ORDER BY id DESC LIMIT 500 """ ).fetchall() added = 0 now = _now() seen: set[tuple[int, str]] = set() for row in rows: if row["event_type"] not in ONBOARD_DOMAIN_EVENTS: continue payload = _parse_payload(row["payload"]) domain = (payload.get("domain") or "").strip().lower() if not domain or len(domain) < 3: continue key = (TENANT_ONBOARD, domain) if key in seen: continue seen.add(key) cur = conn.execute( """ INSERT OR IGNORE INTO audit_domains (tenant_id, domain, source, created_at) VALUES (?, ?, 'onboarding', ?) """, (TENANT_ONBOARD, domain, now), ) if cur.rowcount: added += 1 conn.commit() return added def list_audit_domains(conn: sqlite3.Connection, tenant_id: int | None = None) -> list[dict]: if tenant_id: rows = conn.execute( "SELECT tenant_id, domain, source, created_at FROM audit_domains WHERE tenant_id = ? ORDER BY domain", (tenant_id,), ).fetchall() else: rows = conn.execute( "SELECT tenant_id, domain, source, created_at FROM audit_domains ORDER BY tenant_id, domain" ).fetchall() return [dict(r) for r in rows] def upsert_check( conn: sqlite3.Connection, tenant_id: int, domain: str, check_id: str, status: str, message: str, evidence: dict | None, checked_at: str | None = None, ) -> None: conn.execute( """ INSERT INTO audit_checks (tenant_id, domain, check_id, status, message, evidence, checked_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(tenant_id, domain, check_id) DO UPDATE SET status = excluded.status, message = excluded.message, evidence = excluded.evidence, checked_at = excluded.checked_at """, ( tenant_id, domain.lower(), check_id, status, message, json.dumps(evidence or {}), checked_at or _now(), ), ) def get_checks(conn: sqlite3.Connection, tenant_id: int, domain: str) -> list[dict]: rows = conn.execute( """ SELECT check_id, status, message, evidence, checked_at FROM audit_checks WHERE tenant_id = ? AND domain = ? ORDER BY check_id """, (tenant_id, domain.lower()), ).fetchall() out = [] for row in rows: item = dict(row) item["label"] = CHECK_LABELS.get(item["check_id"], item["check_id"]) item["evidence"] = _parse_payload(item.get("evidence")) out.append(item) return out def aggregate_score(checks: list[dict]) -> dict[str, Any]: total = len(CHECK_LABELS) counts = {"pass": 0, "warn": 0, "fail": 0, "error": 0, "skip": 0} worst = "pass" for c in checks: st = c.get("status") or "skip" counts[st] = counts.get(st, 0) + 1 if STATUS_RANK.get(st, 0) > STATUS_RANK.get(worst, 0): worst = st if worst in ("fail", "error"): overall = "critical" elif worst == "warn": overall = "degraded" elif checks: overall = "healthy" else: overall = "unknown" return { "pass": counts.get("pass", 0), "warn": counts.get("warn", 0), "fail": counts.get("fail", 0), "error": counts.get("error", 0), "skip": counts.get("skip", 0), "total": total, "overall_status": overall, } def tenant_overview(conn: sqlite3.Connection, tenant_id: int, name: str, ip: str) -> dict: domains = list_audit_domains(conn, tenant_id) if not domains: return { "tenant_id": tenant_id, "name": name, "ip": ip, "status": "unknown", "score": {"pass": 0, "warn": 0, "fail": 0, "total": 8}, "domains_count": 0, "last_audit_at": None, "top_issues": [], } all_checks: list[dict] = [] last_audit = None top_issues: list[dict] = [] domain_scores: list[dict] = [] for d in domains: checks = get_checks(conn, tenant_id, d["domain"]) if not checks: continue all_checks.extend(checks) domain_scores.append(aggregate_score(checks)) for c in checks: if c["checked_at"] and (not last_audit or c["checked_at"] > last_audit): last_audit = c["checked_at"] if c["status"] in ("fail", "error", "warn"): top_issues.append({ "domain": d["domain"], "check_id": c["check_id"], "status": c["status"], "message": c.get("message"), }) if domain_scores: worst = max(domain_scores, key=lambda s: STATUS_RANK.get(s["overall_status"], 0)) score = worst else: score = aggregate_score(all_checks) return { "tenant_id": tenant_id, "name": name, "ip": ip, "status": score["overall_status"], "score": { "pass": score["pass"], "warn": score["warn"], "fail": score["fail"] + score["error"], "total": score["total"], }, "domains_count": len(domains), "last_audit_at": last_audit, "top_issues": top_issues[:5], } def build_overview(conn: sqlite3.Connection) -> dict: tenants = conn.execute("SELECT id, name, ip FROM tenants ORDER BY id").fetchall() return { "generated_at": _now(), "tenants": [tenant_overview(conn, t["id"], t["name"], t["ip"]) for t in tenants], } def scorecard(conn: sqlite3.Connection, tenant_id: int, domain: str) -> dict: domain = domain.lower().strip() checks = get_checks(conn, tenant_id, domain) score = aggregate_score(checks) return { "tenant_id": tenant_id, "domain": domain, "checked_at": max((c["checked_at"] for c in checks), default=None), "overall_status": score["overall_status"], "checks": checks, }