#!/usr/bin/env python3
"""sandbroker-check-auditor - confirm the AI auditor is correctly configured.

The AI auditor (experimental, see sandbrokerd.py "AI auditor") reviews confirm-tier
secret-USE requests and may auto-approve the clearly-benign ones so a human
Windows Hello is reserved for anything doubtful. This tool lets an operator
verify, WITHOUT running a real secret request, that the auditor is wired up:

  * the Anthropic key file is present, owner-only, and non-empty
  * the global kill-switch (etc/auditor.disabled) is not set
  * (unless --offline) a minimal live ping authenticates and the model string
    is valid/reachable at the Messages API
  * which vaults have opted in with "auditor": true

It NEVER reads, prints, or logs the key VALUE. The live ping passes the key to
the HTTP request as the x-api-key header, but the value is never displayed and
the on-disk check reports only "present, <mode>, <N> bytes".

Because the key and vault registry live under the daemon's private tree, on the
production box run this as the daemon user:

    sudo -u sandbroker SANDBROKER_BASE=/opt/sandbroker sandbroker-check-auditor
    sudo -u sandbroker SANDBROKER_BASE=/opt/sandbroker sandbroker-check-auditor --offline

Exit status:
    0  all critical checks passed (auditor is usable, or --offline local checks OK)
    1  a critical check failed (key missing / bad perms / empty, or ping failed)
    2  usage error

This module is import-safe: the daemon's `--check-auditor` flag calls
run_checks()/render() here so there is ONE implementation (see
docs/integration-notes/check-auditor.md).
"""
from __future__ import annotations

import argparse
import json
import os
import socket
import stat
import sys
import urllib.error
import urllib.request


# --------------------------------------------------------------------------
# Import the auditor constants from sandbrokerd.py rather than redefining them, so
# a path or model change there is picked up automatically. The daemon module has
# no package name (bare sandbroker/sandbrokerd.py), so load it by file path. BASE is
# read from SANDBROKER_BASE at sandbroker import time, so any --base override must be
# applied to the environment BEFORE this import runs (see main()).
# --------------------------------------------------------------------------

def _load_sandbroker(base: str | None):
    """Import the sibling sandbroker/sandbrokerd.py as a module and return it.

    Reused constants: AUDITOR_KEY_FILE, AUDITOR_DISABLED_PATH, AUDITOR_MODEL,
    AUDITOR_ENDPOINT, AUDITOR_API_TIMEOUT, VAULTS_PATH, load_vault_registry.
    """
    if base:
        os.environ["SANDBROKER_BASE"] = base
    from importlib.machinery import SourceFileLoader
    here = os.path.dirname(os.path.abspath(__file__))
    mod_path = os.path.join(os.path.dirname(here), "sandbroker", "sandbrokerd.py")
    if not os.path.exists(mod_path):
        # Fallback: allow SANDBROKER_MODULE to point at an installed copy.
        mod_path = os.environ.get("SANDBROKER_MODULE", mod_path)
    return SourceFileLoader("sandbroker_daemon", mod_path).load_module()


# --------------------------------------------------------------------------
# Individual checks. Each returns a small dict; none of them ever place the key
# value into their result.
# --------------------------------------------------------------------------

def check_key_file(path: str) -> dict:
    """Key present, owner-only (no group/other bits), and non-empty.

    Reports the octal mode and byte count only -- never the contents. The daemon
    installs this file 0400 (owner read-only); we accept any owner-only mode
    (0400 or 0600) and reject anything readable by group or other.
    """
    res = {"name": "key_file", "path": path, "critical": True, "ok": False}
    try:
        st = os.stat(path)
    except FileNotFoundError:
        res["detail"] = "missing"
        return res
    except OSError as exc:
        res["detail"] = "unreadable: %s" % exc.__class__.__name__
        return res
    mode = stat.S_IMODE(st.st_mode)
    res["mode"] = "%04o" % mode
    res["bytes"] = st.st_size
    owner_only = (mode & 0o077) == 0            # no group/other permission bits
    owner_read = bool(mode & stat.S_IRUSR)
    if not owner_only:
        res["detail"] = "perms too open (%04o); must be owner-only (0400/0600)" % mode
        return res
    if not owner_read:
        res["detail"] = "owner cannot read (%04o)" % mode
        return res
    if st.st_size == 0:
        res["detail"] = "present but empty"
        return res
    res["ok"] = True
    res["detail"] = "present, %04o, %d bytes" % (mode, st.st_size)
    return res


def check_disabled(path: str) -> dict:
    """Global kill-switch. Not a critical failure -- a valid operational state --
    but surfaced prominently because it forces every vault back to human review."""
    disabled = os.path.exists(path)
    return {
        "name": "kill_switch",
        "path": path,
        "critical": False,
        "ok": not disabled,
        "disabled": disabled,
        "detail": "AUDITOR GLOBALLY DISABLED (etc/auditor.disabled present)"
                  if disabled else "not disabled",
    }


def check_vaults(sd) -> dict:
    """Which vaults opted in with auditor: true. Informational (not critical)."""
    res = {"name": "opted_in_vaults", "critical": False, "ok": True,
           "path": sd.VAULTS_PATH, "vaults": []}
    try:
        reg = sd.load_vault_registry()
    except Exception as exc:  # registry unreadable / malformed
        res["ok"] = False
        res["detail"] = "vault registry unreadable: %s" % exc.__class__.__name__
        return res
    opted = sorted(a for a, spec in reg.get("aliases", {}).items()
                   if spec.get("auditor"))
    res["vaults"] = opted
    res["detail"] = (", ".join(opted) if opted
                     else "none (no vault has \"auditor\": true)")
    return res


def check_ping(sd, key_ok: bool) -> dict:
    """Minimal live request to AUDITOR_ENDPOINT with the configured model.

    A tiny max_tokens "reply OK" message with NO tools and NO system prompt. It
    confirms the key authenticates and the model string is valid/reachable. The
    key is read only to populate the x-api-key header; it is NEVER logged or
    returned. HTTP status and a short reason are reported.
    """
    res = {"name": "live_ping", "critical": True, "ok": False,
           "endpoint": sd.AUDITOR_ENDPOINT, "model": sd.AUDITOR_MODEL}
    if not key_ok:
        res["detail"] = "skipped (key file check failed)"
        res["status"] = None
        return res
    try:
        key = sd._auditor_key()            # reads the file; value stays in memory
    except OSError as exc:
        res["detail"] = "could not read key: %s" % exc.__class__.__name__
        res["status"] = None
        return res

    body = json.dumps({
        "model": sd.AUDITOR_MODEL,
        "max_tokens": 8,
        "messages": [{"role": "user", "content": "reply OK"}],
    }).encode()
    req = urllib.request.Request(sd.AUDITOR_ENDPOINT, data=body, headers={
        "content-type": "application/json",
        "anthropic-version": "2023-06-01",
        "x-api-key": key,                  # used, never displayed
    })
    try:
        with urllib.request.urlopen(req, timeout=sd.AUDITOR_API_TIMEOUT) as resp:
            status = resp.getcode()
            resp.read(1)                   # drain a byte; body is never inspected
        res["status"] = status
        res["ok"] = 200 <= status < 300
        res["detail"] = "HTTP %d, model %r reachable" % (status, sd.AUDITOR_MODEL)
        return res
    except urllib.error.HTTPError as exc:
        res["status"] = exc.code
        reason = _api_error_reason(exc)
        if exc.code in (401, 403):
            res["detail"] = "HTTP %d auth failed (key rejected): %s" % (exc.code, reason)
        elif exc.code == 404:
            res["detail"] = "HTTP 404 model %r not found: %s" % (sd.AUDITOR_MODEL, reason)
        elif exc.code == 400:
            res["detail"] = "HTTP 400 bad request (model %r?): %s" % (sd.AUDITOR_MODEL, reason)
        else:
            res["detail"] = "HTTP %d: %s" % (exc.code, reason)
        return res
    except (urllib.error.URLError, socket.timeout, TimeoutError) as exc:
        res["status"] = None
        why = getattr(exc, "reason", exc)
        res["detail"] = "endpoint unreachable: %s" % why
        return res
    finally:
        del key                            # drop the reference promptly


def _api_error_reason(exc: urllib.error.HTTPError) -> str:
    """Short, key-free reason from an Anthropic error body. The API returns its
    own {"error": {"type", "message"}}; that text is Anthropic's, not the key,
    but we still truncate it hard."""
    try:
        payload = json.loads(exc.read().decode("utf-8", "replace"))
        err = payload.get("error", {})
        msg = err.get("message") or err.get("type") or ""
    except Exception:
        msg = exc.reason if getattr(exc, "reason", None) else ""
    msg = " ".join(str(msg).split())
    return (msg[:120] + "...") if len(msg) > 120 else (msg or "no detail")


# --------------------------------------------------------------------------
# Orchestration
# --------------------------------------------------------------------------

def run_checks(offline: bool = False, base: str | None = None, sd=None) -> dict:
    """Run every check and return a structured result. Shared by this CLI and
    the daemon's --check-auditor flag. Does not print anything.

    `sd` is the loaded sandbroker module. The standalone CLI leaves it None and we
    load sandbrokerd.py by path; the daemon passes itself (already imported) so it
    is not loaded a second time -- see docs/integration-notes/check-auditor.md."""
    if sd is None:
        sd = _load_sandbroker(base)
    checks = []
    key = check_key_file(sd.AUDITOR_KEY_FILE)
    checks.append(key)
    checks.append(check_disabled(sd.AUDITOR_DISABLED_PATH))
    checks.append(check_vaults(sd))
    if offline:
        checks.append({"name": "live_ping", "critical": True, "ok": True,
                       "skipped": True, "status": None,
                       "endpoint": sd.AUDITOR_ENDPOINT, "model": sd.AUDITOR_MODEL,
                       "detail": "skipped (--offline)"})
    else:
        checks.append(check_ping(sd, key_ok=key["ok"]))

    failed = [c for c in checks if c.get("critical") and not c.get("ok")]
    return {
        "ok": not failed,
        "offline": offline,
        "base": sd.BASE,
        "model": sd.AUDITOR_MODEL,
        "endpoint": sd.AUDITOR_ENDPOINT,
        "checks": checks,
    }


def render(result: dict) -> str:
    """Human-readable summary. Colour only when stdout is a TTY."""
    tty = sys.stdout.isatty()
    green = "\033[32m" if tty else ""
    red = "\033[31m" if tty else ""
    yellow = "\033[33m" if tty else ""
    dim = "\033[2m" if tty else ""
    off = "\033[0m" if tty else ""

    def mark(c):
        if c.get("skipped"):
            return "%s-%s" % (dim, off)
        if c.get("ok"):
            return "%sOK%s" % (green, off)
        return ("%sWARN%s" % (yellow, off)) if not c.get("critical") \
            else "%sFAIL%s" % (red, off)

    lines = []
    lines.append("sandbroker auditor check  (base=%s)" % result["base"])
    lines.append("  model:    %s" % result["model"])
    lines.append("  endpoint: %s" % result["endpoint"])
    lines.append("  mode:     %s" % ("offline (local checks only)"
                                     if result["offline"] else "live ping"))
    lines.append("")
    label = {
        "key_file": "key file",
        "kill_switch": "kill-switch",
        "opted_in_vaults": "opted-in vaults",
        "live_ping": "live ping",
    }
    for c in result["checks"]:
        lines.append("  [%s] %-16s %s"
                     % (mark(c), label.get(c["name"], c["name"]), c.get("detail", "")))
    lines.append("")
    verdict = ("%sPASS%s - auditor configuration looks good" % (green, off)) \
        if result["ok"] else ("%sFAIL%s - auditor is not usable as configured" % (red, off))
    lines.append("  %s" % verdict)
    return "\n".join(lines)


def main(argv=None) -> int:
    argv = sys.argv[1:] if argv is None else argv
    ap = argparse.ArgumentParser(
        prog="sandbroker-check-auditor",
        description="Confirm the sandbroker AI auditor is correctly configured "
                    "(key, kill-switch, model reachability, opted-in vaults). "
                    "Never reads or prints the key value.")
    ap.add_argument("--offline", action="store_true",
                    help="skip the network ping; run only local checks")
    ap.add_argument("--json", action="store_true",
                    help="emit a machine-readable JSON result")
    ap.add_argument("--base",
                    help="override SANDBROKER_BASE (default: env or /opt/sandbroker)")
    args = ap.parse_args(argv)

    try:
        result = run_checks(offline=args.offline, base=args.base)
    except Exception as exc:
        sys.stderr.write("sandbroker-check-auditor: %s: %s\n"
                         % (exc.__class__.__name__, exc))
        return 1

    if args.json:
        print(json.dumps(result, indent=2, sort_keys=True))
    else:
        print(render(result))
    return 0 if result["ok"] else 1


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