#!/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 use the first local token that may call ``sys/capabilities`` (``BAO_TOKEN`` or ``VAULT_TOKEN``, then ``~/.local/openbao/platform-admin.token``, then ``~/.vault-token``), 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") # This OpenBao issues s. service tokens. hvs/hvb/hvr are the newer Vault shapes. TOKEN_RE = re.compile(r"\b(?:hvs|hvb|hvr|s)\.[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 token_candidates() -> list[tuple[str, str]]: found: list[tuple[str, str]] = [] for name in ("BAO_TOKEN", "VAULT_TOKEN"): value = os.environ.get(name, "").strip() if value: found.append((name, value)) home = Path(os.environ.get("HOME", str(Path.home()))) for label, path in ( ("~/.local/openbao/platform-admin.token", home / ".local/openbao/platform-admin.token"), ("~/.vault-token", home / ".vault-token"), ): if path.is_file(): value = path.read_text(encoding="utf-8").strip() if value: found.append((label, value)) return found def write_capabilities(caller: str, subject: str, paths: list[str]) -> subprocess.CompletedProcess[str]: body = json.dumps({"token": subject, "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() return run_bao(["write", "-format=json", "sys/capabilities", f"@{tmp.name}"], caller) finally: try: os.remove(tmp.name) except OSError: pass def operator_token() -> str: """Return a token allowed to call sys/capabilities, or an empty string.""" for _label, token in token_candidates(): probe = write_capabilities(token, token, ["auth/token/lookup-self"]) if probe.returncode == 0: return token 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 TOKEN_RE.fullmatch(token) is None: sys.stderr.write(redact(proc.stderr or "login output was not a token\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 token can call sys/capabilities " "(BAO_TOKEN, VAULT_TOKEN, ~/.local/openbao/platform-admin.token, or ~/.vault-token)" ) proc = write_capabilities(operator, lane, PATHS) 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 operator = operator_token() if not operator: sys.stderr.write( "no token can call sys/capabilities. Checked BAO_TOKEN, VAULT_TOKEN, " "~/.local/openbao/platform-admin.token, and ~/.vault-token.\n" ) return 1 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" ) 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))