#!/usr/bin/env python3 """Attended proof of the reuse-surface hub-writer caller. ./scripts/verify-reuse-surface-hub-writer.py member ./scripts/verify-reuse-surface-hub-writer.py non-member Logs in on auth/netkingdom role reuse-surface-hub-writer-workload-kv-read with ``bao login -token-only``, so the operator's token helper is left alone. The lane token stays in memory. Capability queries run as the operator token already present in the environment or token helper, because the workload policy has no sys/capabilities-self grant. The receipt records policies, entity id, path capabilities, and pass/fail. It does not record the token or any KV value. Sign in with a private browser window. A reused platform-admin session is the wrong principal for ``member``. """ from __future__ import annotations import json import os import re import stat import subprocess import sys import tempfile from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from ops_mason.reuse_surface_hub_writer import ( DATA, METADATA, MUST_DENY, PARENT_DATA, PARENT_METADATA, POLICY, ROLE, SIBLING, assess, ) BAO_ADDR = os.environ.get("BAO_ADDR", "http://127.0.0.1:18200") TOKEN_RE = re.compile(r"\b(?:hvs|hvb|hvr)\.[A-Za-z0-9_-]{8,}\b") PATHS = [DATA, METADATA, PARENT_DATA, PARENT_METADATA, SIBLING] def redact(text: str) -> str: return TOKEN_RE.sub("[redacted-token]", text or "") 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 operator_token() -> str: for name in ("BAO_TOKEN", "VAULT_TOKEN"): value = os.environ.get(name, "").strip() if value: return value helper = Path(os.environ.get("HOME", str(Path.home()))) / ".vault-token" if helper.is_file(): return helper.read_text(encoding="utf-8").strip() return "" 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 login() -> str: proc = run_bao( ["login", "-token-only", "-method=oidc", "-path=netkingdom", f"role={ROLE}"], token="", ) token = (proc.stdout or "").strip() if proc.returncode != 0 or not token or TOKEN_RE.fullmatch(token) is None: sys.stderr.write(redact(proc.stderr or "login refused\n")) return "" 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(operator: str, lane: str) -> dict[str, list[str]]: if not operator: raise RuntimeError( "no operator token for sys/capabilities " "(BAO_TOKEN, VAULT_TOKEN, or ~/.vault-token)" ) body = json.dumps({"token": lane, "paths": PATHS}) tmp = tempfile.NamedTemporaryFile("w", prefix="rs-caps-", delete=False) try: os.fchmod(tmp.fileno(), stat.S_IRUSR | stat.S_IWUSR) tmp.write(body) tmp.close() proc = run_bao(["write", "-format=json", "sys/capabilities", f"@{tmp.name}"], operator) finally: try: os.remove(tmp.name) except OSError: pass if proc.returncode != 0: raise RuntimeError(redact(proc.stderr or "sys/capabilities failed")) 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 main(argv: list[str]) -> int: mode = argv[1] if len(argv) > 1 else "" if mode not in {"member", "non-member"}: sys.stderr.write("usage: verify-reuse-surface-hub-writer.py member|non-member\n") return 2 sys.stderr.write( f"Sign in as the {mode} identity in a private browser window.\n" "A platform-admin session is the wrong principal for member.\n" ) operator = operator_token() now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") lane = login() report: dict = { "mode": mode, "login": "ok" if lane else "refused", "policies": [], "identity_policies": [], "entity_id": "", "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 "") if mode == "member": report["capabilities"] = capabilities(operator, lane) except (RuntimeError, json.JSONDecodeError, OSError) as exc: report["error"] = redact(str(exc)) finally: if lane: report["revoked"] = revoke(lane) lane = "" failures = assess(report) if report.get("error"): failures = [report["error"], *failures] lines = [ f"mode={mode}", f"at={now}", f"addr={BAO_ADDR}", f"role={ROLE}", f"policy={POLICY}", f"login={report['login']}", 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=" + item) receipt = write_receipt(f"reuse-surface-hub-writer-{mode}.txt", lines) sys.stderr.write(f"receipt: {receipt}\n") return 0 if not failures else 1 if __name__ == "__main__": sys.exit(main(sys.argv))