ops-mason/scripts/verify-reuse-surface-hub-writer.py

#!/usr/bin/env python3
"""Attended proof of the reuse-surface hub-writer caller.

    ./scripts/verify-reuse-surface-hub-writer.py member [--uid UID] [--assurance aal1|aal2]
    ./scripts/verify-reuse-surface-hub-writer.py non-member [--uid UID] [--assurance aal1|aal2]

Logs in on auth/netkingdom role reuse-surface-hub-writer-workload-kv-read with
``bao login -token-only ... skip_browser=true``: bao prints the sign-in address
and waits, so the default browser's Authelia session is never reused. Open the
address in a private window, or set ``OPS_MASON_LOGIN_BROWSER`` to a Chromium
or Firefox binary and the script opens it in a fresh temporary profile. The
operator's token helper is left alone and the lane token stays in memory.

Capabilities come from ``sys/capabilities-self`` called with the lane token
itself (CCR-2026-0045). No operator or root token is read or used.

The receipt follows ``net-kingdom/docs/sign-in-diagnosis.md``: ``uid``,
``openbao_error`` (verbatim, redacted of tokens, codes and emails) and
``assurance`` when known, plus policies, entity id, path capabilities and
pass/fail. ``--uid`` and ``--assurance`` are what KeyCape's diagnosis page
showed for this request. It never records a token or any KV value.
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import threading
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from ops_mason.attended_login import (
    TOKEN_RE,
    classify_login_line,
    fresh_profile_argv,
    login_args,
    openbao_error_text,
    redact,
    resolve_assurance,
    resolve_uid,
)
from ops_mason.reuse_surface_hub_writer import (
    DATA,
    METADATA,
    MOUNT,
    MUST_DENY,
    PARENT_DATA,
    PARENT_METADATA,
    POLICY,
    ROLE,
    ROLE_REQUIRES_AAL2,
    SIBLING,
    assess,
)

BAO_ADDR = os.environ.get("BAO_ADDR", "http://127.0.0.1:18200")
PATHS = [DATA, METADATA, PARENT_DATA, PARENT_METADATA, SIBLING]


def receipts_dir() -> Path:
    home = Path(os.environ.get("HOME", str(Path.home())))
    path = home / ".local/state/railiance/receipts"
    path.mkdir(parents=True, mode=0o700, exist_ok=True)
    return path


def run_bao(args: list[str], token: str, stdin: str | None = None) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env["BAO_ADDR"] = BAO_ADDR
    env["BAO_TOKEN"] = token
    env.pop("VAULT_TOKEN", None)
    return subprocess.run(
        ["bao", *args],
        input=stdin,
        text=True,
        capture_output=True,
        env=env,
        check=False,
    )


def open_fresh_profile(url: str, profile_dirs: list[str]) -> None:
    browser = os.environ.get("OPS_MASON_LOGIN_BROWSER", "").strip()
    if not browser:
        return
    resolved = shutil.which(browser) or browser
    profile = tempfile.mkdtemp(prefix="ops-mason-login-profile-")
    profile_dirs.append(profile)
    try:
        subprocess.Popen(
            fresh_profile_argv(resolved, Path(profile), url),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        sys.stderr.write("Opened in a fresh browser profile.\n")
    except OSError as exc:
        sys.stderr.write(f"Could not start {browser}: {exc}. Open the address in a private window.\n")


def login() -> tuple[str, str]:
    """Return (lane token or "", OpenBao error text or "")."""
    env = os.environ.copy()
    env["BAO_ADDR"] = BAO_ADDR
    env.pop("BAO_TOKEN", None)
    env.pop("VAULT_TOKEN", None)
    proc = subprocess.Popen(
        ["bao", *login_args(MOUNT, ROLE)],
        text=True,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=env,
    )
    kept: list[str] = []
    profile_dirs: list[str] = []

    def pump() -> None:
        assert proc.stderr is not None
        for line in proc.stderr:
            kind = classify_login_line(line)
            if kind == "url":
                sys.stderr.write("\nOpen this address in a private browser window:\n\n")
                sys.stderr.write(line.strip() + "\n\n")
                open_fresh_profile(line.strip(), profile_dirs)
            elif kind == "instruction":
                if line.strip():
                    sys.stderr.write(line.strip() + "\n")
            else:
                kept.append(line)
                sys.stderr.write(redact(line))

    reader = threading.Thread(target=pump, daemon=True)
    reader.start()
    token = (proc.stdout.read() if proc.stdout else "").strip()
    proc.wait()
    reader.join(timeout=5)
    for profile in profile_dirs:
        shutil.rmtree(profile, ignore_errors=True)
    if proc.returncode != 0 or TOKEN_RE.fullmatch(token) is None:
        return "", openbao_error_text(kept) or "login output was not a token"
    return token, ""


def lookup(token: str) -> dict:
    proc = run_bao(["token", "lookup", "-format=json"], token)
    if proc.returncode != 0:
        raise RuntimeError(redact(proc.stderr or "token lookup failed"))
    payload = json.loads(proc.stdout)
    return payload.get("data") or payload


def capabilities(lane: str) -> dict[str, list[str]]:
    """The lane token's own capabilities, from sys/capabilities-self."""
    proc = run_bao(
        ["write", "-format=json", "sys/capabilities-self", "paths=" + ",".join(PATHS)],
        lane,
    )
    if proc.returncode != 0:
        raise RuntimeError(redact(proc.stderr or "sys/capabilities-self failed").strip())
    payload = json.loads(proc.stdout)
    data = payload.get("data") or payload
    found: dict[str, list[str]] = {}
    for path in PATHS:
        value = data.get(path)
        if isinstance(value, list):
            found[path] = value
    return found


def revoke(token: str) -> bool:
    proc = run_bao(["token", "revoke", "-self"], token)
    return proc.returncode == 0


def write_receipt(name: str, lines: list[str]) -> Path:
    path = receipts_dir() / name
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    path.chmod(0o600)
    return path


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(prog="verify-reuse-surface-hub-writer.py")
    parser.add_argument("mode", choices=["member", "non-member"])
    parser.add_argument("--uid", default="", help="uid KeyCape showed for this request")
    parser.add_argument(
        "--assurance", default="", choices=["", "aal1", "aal2"],
        help="assurance KeyCape showed for this request",
    )
    return parser.parse_args(argv[1:])


def receipt_lines(report: dict, now: str, failures: list[str]) -> list[str]:
    lines = [
        f"mode={report['mode']}",
        f"at={now}",
        f"addr={BAO_ADDR}",
        f"role={ROLE}",
        f"policy={POLICY}",
        f"uid={report['uid']}",
        f"uid_source={report['uid_source']}",
        f"assurance={report['assurance']}",
        f"assurance_source={report['assurance_source']}",
        f"login={report['login']}",
        f"openbao_error={report['openbao_error']}",
        f"entity_id={report['entity_id']}",
        "policies=" + ",".join(report["policies"]),
        "identity_policies=" + ",".join(report["identity_policies"]),
    ]
    for path in PATHS:
        caps = (report["capabilities"] or {}).get(path)
        shown = ",".join(caps) if isinstance(caps, list) else "not-run"
        lines.append(f"{path}: {shown}")
    lines.append("revoked=" + ("yes" if report["revoked"] else "no" if report["revoked"] is False else "n/a"))
    lines.append("result=" + ("pass" if not failures else "fail"))
    for item in failures:
        lines.append("failure=" + redact(item))
    return [redact(line) for line in lines]


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    mode = args.mode
    try:
        resolve_uid(args.uid, "", MOUNT)
    except ValueError as exc:
        sys.stderr.write(f"{exc}\n")
        return 2
    sys.stderr.write(
        f"Sign in as the {mode} identity. bao prints the address and waits;\n"
        "open it in a private window. A platform-admin session is the wrong principal for member.\n"
    )
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    lane, login_error = login()
    report: dict = {
        "mode": mode,
        "login": "ok" if lane else "refused",
        "openbao_error": login_error,
        "policies": [],
        "identity_policies": [],
        "entity_id": "",
        "display_name": "",
        "capabilities": {},
        "revoked": None,
    }
    try:
        if lane:
            info = lookup(lane)
            report["policies"] = list(info.get("policies") or [])
            report["identity_policies"] = list(info.get("identity_policies") or [])
            report["entity_id"] = str(info.get("entity_id") or "")
            report["display_name"] = str(info.get("display_name") or "")
            if mode == "member":
                report["capabilities"] = capabilities(lane)
    except (RuntimeError, json.JSONDecodeError, OSError) as exc:
        report["error"] = redact(str(exc))
        report["openbao_error"] = report["openbao_error"] or report["error"]
    finally:
        if lane:
            report["revoked"] = revoke(lane)
            lane = ""
    report["uid"], report["uid_source"] = resolve_uid(args.uid, report.pop("display_name"), MOUNT)
    report["assurance"], report["assurance_source"] = resolve_assurance(
        args.assurance, report["login"] == "ok", ROLE_REQUIRES_AAL2
    )
    failures = assess(report)
    if report.get("error"):
        failures = [report["error"], *failures]
    receipt = write_receipt(f"reuse-surface-hub-writer-{mode}.txt", receipt_lines(report, now, failures))
    sys.stderr.write(f"receipt: {receipt}\n")
    return 0 if not failures else 1


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