oikos/oikos/drift.py

#!/usr/bin/env python3
"""oikos/drift.py — drift detectors (Week 3 reliability layer).

Each detector returns a list of finding dicts:
    {kind, severity, entity, evidence, likely_cause, recommended_action,
     verification}
matching the Signal schema in oikos/signal.py. `run_all()` runs every
detector and raises a Signal for each finding, skipping ones that already
have an open Signal for the same (entity, kind) — see
oikos/signal.py open_signal_for(). SSH-based detectors degrade to a
"probe-unreachable" info finding instead of a false drift signal when the
target can't be reached (no ssh, no network) — drift means "state
disagrees," not "couldn't check."

Two detectors are fully local (no SSH, always safe to run from anywhere
with the repo checked out): SOPS-recipient-vs-inventory and lifecycle
consistency. Two need live access to hubris (pct list, caddy backend
config) and gracefully no-op when unreachable.

Not yet implemented — needs data this repo doesn't structure yet:
  - DNS records vs inventory services (no Technitium API wiring here)
  - Generic tracked-config-repo cleanliness for every service (needs the
    `mutation_path`/local-checkout-path field per service; only caddy's
    path is documented today, so that's the one config-cleanliness check
    implemented below)
"""

from __future__ import annotations

import re
import subprocess
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO))
from oikos import relations as oikos_relations  # noqa: E402
from oikos import signal as oikos_signal  # noqa: E402

import yaml  # noqa: E402

SOPS_FILE = REPO / ".sops.yaml"
INVENTORY = REPO / "inventory.yaml"

_AGE_RE = re.compile(r"age1[a-z0-9]+")


def _load_inventory() -> dict:
    return yaml.safe_load(INVENTORY.read_text())


def _sops_rules() -> list[dict]:
    """Parse .sops.yaml's creation_rules: [{path_regex, recipients: [...]}]."""
    doc = yaml.safe_load(SOPS_FILE.read_text())
    return [
        {"path_regex": r.get("path_regex", ""),
         "recipients": _AGE_RE.findall(r.get("age", "") or "")}
        for r in doc.get("creation_rules", [])
    ]


def check_sops_recipients(inv: dict | None = None) -> list[dict]:
    """Two directions of drift:
    1. A recipient pubkey in .sops.yaml that matches no host's age_pubkey
       in inventory — likely a removed/revoked host whose secrets weren't
       fully cleaned up (or a stale rule).
    2. An enrolled host (non-empty age_pubkey) missing from the
       secrets/hello.yaml rule — every enrolled client should be able to
       decrypt the bootstrap canary; absence usually means enrollment
       stalled partway.
    """
    inv = inv or _load_inventory()
    hosts = inv.get("hosts", {})
    known_pubkeys = {e["age_pubkey"]: name for name, e in hosts.items()
                     if e.get("age_pubkey")}
    rules = _sops_rules()
    findings = []

    all_recipients = {pk for r in rules for pk in r["recipients"]}
    for pk in all_recipients:
        if pk not in known_pubkeys:
            findings.append({
                "kind": "sops-orphan-recipient", "severity": "warning",
                "entity": "repo:Homelab-Docs",
                "evidence": f"age recipient {pk[:20]}... appears in .sops.yaml but "
                            f"matches no host's age_pubkey in inventory.yaml",
                "likely_cause": "host removed without full secret revocation, or a stale rule",
                "recommended_action": {"runbook": "lifecycle-destroy-node", "risk": "destructive"},
                "verification": "grep <pubkey> .sops.yaml; check inventory.yaml + archaeology",
            })

    hello_rule = next((r for r in rules if "hello" in r["path_regex"]), None)
    if hello_rule:
        for pubkey, name in known_pubkeys.items():
            if pubkey not in hello_rule["recipients"]:
                findings.append({
                    "kind": "sops-missing-recipient", "severity": "warning",
                    "entity": f"host:{name}",
                    "evidence": f"{name}'s age_pubkey is not a recipient of secrets/hello.yaml",
                    "likely_cause": "client-add --finalize-pubkey ran incompletely",
                    "recommended_action": {"runbook": "client-enrollment", "risk": "config_mutation"},
                    "verification": f"homelab client add {name} --finalize-pubkey <key>",
                })
    return findings


def check_lifecycle_consistency(inv: dict | None = None) -> list[dict]:
    """state: vs reality, using the ontology graph (oikos/relations.py) —
    no SSH needed, this only reasons over inventory.yaml itself."""
    inv = inv or _load_inventory()
    hosts = inv.get("hosts", {})
    archaeology = inv.get("archaeology", {})
    findings = []

    for name, e in hosts.items():
        state = e.get("state", "active")
        if state == "deprecated":
            rel = oikos_relations.relations(f"host:{name}", inv)
            if rel["affected_by"]:
                findings.append({
                    "kind": "lifecycle-inconsistent", "severity": "warning",
                    "entity": f"host:{name}",
                    "evidence": f"deprecated but still has inbound edges: {', '.join(rel['affected_by'])}",
                    "likely_cause": "deprecation started before dependents were migrated off",
                    "recommended_action": {"runbook": "lifecycle-deprecate-node", "risk": "config_mutation"},
                    "verification": f"homelab node {name} relations",
                })
        elif state == "destroyed":
            findings.append({
                "kind": "lifecycle-inconsistent", "severity": "critical",
                "entity": f"host:{name}",
                "evidence": "state is 'destroyed' but the host still has a live hosts.<name>: entry "
                            "(destroyed nodes belong in the archaeology: section, not hosts:)",
                "likely_cause": "lifecycle-destroy-node runbook step 6 (move to archaeology) was skipped",
                "recommended_action": {"runbook": "lifecycle-destroy-node", "risk": "destructive"},
                "verification": "grep -A3 '^  " + name + ":' inventory.yaml",
            })

    # A pve_id collision between an active host and an archaeology entry on
    # the same Proxmox node usually means the ID was reused without the old
    # entry's data being fully accounted for.
    active_by_id = {(e.get("host"), e.get("pve_id")): name
                    for name, e in hosts.items() if e.get("pve_id")}
    for aname, ae in archaeology.items():
        key = (None, ae.get("pve_id"))  # archaeology doesn't record which node
        for (host, pve_id), name in active_by_id.items():
            if pve_id == ae.get("pve_id") and name != aname:
                findings.append({
                    "kind": "lifecycle-pve-id-reuse", "severity": "info",
                    "entity": f"host:{name}",
                    "evidence": f"pve_id {pve_id} is both active ({name}) and in archaeology ({aname}, "
                                f"destroyed {ae.get('destroyed', '?')})",
                    "likely_cause": "normal ID reuse after destroy — informational only",
                    "recommended_action": None,
                    "verification": None,
                })
    return findings


def _ssh_hubris(cmd: list[str]) -> subprocess.CompletedProcess | None:
    """Best-effort ssh to hubris. Returns None (not a CompletedProcess with
    nonzero rc) if ssh itself can't run/connect at all, so callers can
    distinguish "unreachable" from "reachable but the command failed."""
    try:
        return subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
                               "root@192.168.8.77", " ".join(cmd)],
                              capture_output=True, text=True, timeout=15)
    except (subprocess.TimeoutExpired, OSError):
        return None


def check_pct_list(inv: dict | None = None) -> list[dict]:
    """Inventory LXCs/VMs on hubris vs live `pct list` + `qm list`."""
    inv = inv or _load_inventory()
    hosts = inv.get("hosts", {})
    proc = _ssh_hubris(["pct", "list"])
    if proc is None or proc.returncode != 0:
        return [{
            "kind": "probe-unreachable", "severity": "info", "entity": "host:hubris",
            "evidence": "could not reach hubris to run `pct list`",
            "likely_cause": "no network path from this client to hubris, or ssh key not authorized",
            "recommended_action": None, "verification": "homelab ssh hubris -- pct list",
        }]
    live_ids = set()
    for line in proc.stdout.splitlines()[1:]:
        parts = line.split()
        if parts and parts[0].isdigit():
            live_ids.add(int(parts[0]))

    inventory_ids = {e["pve_id"] for name, e in hosts.items()
                    if e.get("kind") == "lxc" and e.get("host") == "hubris" and e.get("pve_id")}

    findings = []
    for missing in sorted(inventory_ids - live_ids):
        name = next(n for n, e in hosts.items() if e.get("pve_id") == missing)
        findings.append({
            "kind": "inventory-vs-live", "severity": "critical", "entity": f"host:{name}",
            "evidence": f"pve_id {missing} ({name}) is in inventory.yaml but absent from "
                        f"`pct list` on hubris",
            "likely_cause": "destroyed outside the lifecycle-destroy-node runbook, or migrated "
                            "without updating inventory",
            "recommended_action": {"runbook": "lifecycle-destroy-node", "risk": "destructive"},
            "verification": "homelab ssh hubris -- pct list",
        })
    for extra in sorted(live_ids - inventory_ids):
        findings.append({
            "kind": "inventory-vs-live", "severity": "warning", "entity": "host:hubris",
            "evidence": f"pve_id {extra} exists on hubris (`pct list`) but has no inventory.yaml entry",
            "likely_cause": "created outside the provision-node runbook",
            "recommended_action": {"runbook": "lifecycle-provision-node", "risk": "config_mutation"},
            "verification": "homelab ssh hubris -- pct config " + str(extra),
        })
    return findings


def check_caddy_backends(inv: dict | None = None) -> list[dict]:
    """Caddy's /etc/caddy (a git checkout of dtoro/caddy-conf, per
    containers/121-caddy.md) vs inventory service backend IPs. Best-effort
    grep for reverse_proxy targets; skips services whose Caddyfile snippet
    doesn't use a bare IP (e.g. references a Caddy snippet/import)."""
    inv = inv or _load_inventory()
    hosts = inv.get("hosts", {})
    services = inv.get("services", {})
    caddy_ip = hosts.get("caddy", {}).get("lan_ip")
    if not caddy_ip:
        return []
    try:
        proc = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", f"root@{caddy_ip}",
             "grep -rhoE 'reverse_proxy[^{]*' /etc/caddy/ 2>/dev/null"],
            capture_output=True, text=True, timeout=15,
        )
    except (subprocess.TimeoutExpired, OSError):
        proc = None
    if proc is None or proc.returncode != 0:
        return [{
            "kind": "probe-unreachable", "severity": "info", "entity": "host:caddy",
            "evidence": "could not reach caddy to inspect /etc/caddy",
            "likely_cause": "no network path from this client to caddy, or ssh key not authorized",
            "recommended_action": None, "verification": "homelab ssh caddy -- grep -r reverse_proxy /etc/caddy",
        }]

    live_ips = set(re.findall(r"\d+\.\d+\.\d+\.\d+", proc.stdout))
    findings = []
    for svc, e in services.items():
        if not isinstance(e, dict):
            continue
        backend = e.get("backend")
        if not backend or not (e.get("url") or e.get("public_host")):
            continue
        backend_ip = hosts.get(backend, {}).get("lan_ip")
        if backend_ip and live_ips and backend_ip not in live_ips:
            findings.append({
                "kind": "caddy-backend-mismatch", "severity": "warning", "entity": f"service:{svc}",
                "evidence": f"inventory backend IP {backend_ip} ({backend}) not found in any "
                            f"reverse_proxy directive on caddy",
                "likely_cause": "backend migrated (lan_ip changed) without updating the Caddyfile, "
                                "or the service uses a snippet/import this grep can't resolve",
                "recommended_action": {"runbook": "config-change-deploy", "risk": "config_mutation"},
                "verification": f"homelab service {svc} health",
            })
    return findings


DETECTORS = [check_sops_recipients, check_lifecycle_consistency,
             check_pct_list, check_caddy_backends]


def run_all(inv: dict | None = None, *, raise_signals: bool = False) -> list[dict]:
    inv = inv or _load_inventory()
    findings: list[dict] = []
    for detector in DETECTORS:
        findings.extend(detector(inv))
    if raise_signals:
        for f in findings:
            if oikos_signal.open_signal_for(f["entity"], f["kind"]) is not None:
                continue  # already open, don't duplicate
            oikos_signal.raise_signal(
                f["kind"], f["severity"], f["entity"], f["evidence"],
                likely_cause=f.get("likely_cause"),
                recommended_action=f.get("recommended_action"),
                verification=f.get("verification"),
            )
    return findings


def main() -> int:
    import argparse
    import json
    p = argparse.ArgumentParser(description="oikos drift detectors")
    p.add_argument("--raise-signals", action="store_true",
                   help="raise a Signal for each new finding (default: print only)")
    args = p.parse_args()
    findings = run_all(raise_signals=args.raise_signals)
    for f in findings:
        print(json.dumps(f))
    print(f"\n{len(findings)} finding(s)", file=sys.stderr)
    return 1 if any(f["severity"] == "critical" for f in findings) else 0


if __name__ == "__main__":
    sys.exit(main())