ops-mason/scripts/custody-inventory.py

#!/usr/bin/env python3
"""custody-inventory — what credentials exist, what each is for, who owns it.

    ./scripts/custody-inventory.py                  operators/ and platform/workloads/
    ./scripts/custody-inventory.py operators        one mount or prefix
    ./scripts/custody-inventory.py --undescribed    only paths missing metadata

Reads metadata only — never a value — so it runs under ops-mason-build and can
be handed to anyone orienting themselves without granting them a single secret.

A path with no description is a finding, not a formatting problem: it is a
credential nobody can identify without reading it, which is how a store turns
back into the drawer of unlabelled keys it was meant to replace.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys

REQUIRED = ("description", "owner", "used_by", "rotation", "on_loss")
DEFAULT_ROOTS = ("operators", "platform/workloads")


class InventoryError(RuntimeError):
    """Inventory could not be completed; never report this as an empty store."""


def bao(*args: str) -> dict | list:
    env = dict(os.environ)
    env.setdefault("BAO_ADDR", "http://127.0.0.1:18200")
    grant = os.path.expanduser("~/.claude-bao-token")
    if "BAO_TOKEN" not in env and os.path.exists(grant):
        with open(grant) as f:
            env["BAO_TOKEN"] = f.read().strip()
    if not env.get("BAO_TOKEN"):
        raise InventoryError("No scoped BAO_TOKEN or ops-mason grant; inventory was not run.")
    # The Vault/OpenBao CLI rejects flags placed after a positional argument,
    # so -format=json goes immediately before the path, not at the end.
    argv = ["bao", *args[:-1], "-format=json", args[-1]]
    try:
        p = subprocess.run(argv, capture_output=True, text=True, timeout=30, env=env)
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise InventoryError("OpenBao metadata command unavailable or timed out") from exc
    if p.returncode != 0:
        raise InventoryError(f"OpenBao metadata request failed for {args[-1]}; inventory incomplete")
    try:
        return json.loads(p.stdout)
    except json.JSONDecodeError as exc:
        raise InventoryError("Invalid OpenBao metadata response; inventory incomplete") from exc


def walk(prefix: str) -> list[str]:
    keys = bao("kv", "list", prefix)
    if not isinstance(keys, list):
        raise InventoryError(f"Invalid metadata listing for {prefix}")
    out: list[str] = []
    for k in keys:
        if not isinstance(k, str) or not k.rstrip("/") or "/" in k.rstrip("/") or k.rstrip("/") in {".", ".."}:
            raise InventoryError(f"Invalid child in metadata listing for {prefix}")
        child = f"{prefix.rstrip('/')}/{k.rstrip('/')}"
        out.extend(walk(child) if k.endswith("/") else [child])
    return out


def main() -> int:
    only_undescribed = "--undescribed" in sys.argv
    roots = [a for a in sys.argv[1:] if not a.startswith("-")] or list(DEFAULT_ROOTS)

    total = incomplete = 0
    for root in roots:
        for path in walk(root):
            total += 1
            meta = bao("kv", "metadata", "get", path)
            if not isinstance(meta, dict) or not isinstance(meta.get("data"), dict):
                raise InventoryError(f"Invalid metadata response for {path}")
            d = meta["data"]
            cm = d.get("custom_metadata") or {}
            if not isinstance(cm, dict):
                raise InventoryError(f"Invalid custom metadata for {path}")
            missing = [k for k in REQUIRED if not cm.get(k)]
            if missing:
                incomplete += 1
            if only_undescribed and not missing:
                continue

            flag = "  " if not missing else "! "
            print(f"{flag}{path}   (v{d.get('current_version', '?')}, "
                  f"{str(d.get('created_time', ''))[:10]})")
            for label, key in (("", "description"), ("owner:   ", "owner"),
                               ("used by: ", "used_by"), ("if lost: ", "on_loss")):
                if cm.get(key):
                    print(f"      {label}{cm[key]}")
            if missing:
                print(f"      MISSING: {', '.join(missing)}")
            print()

    print("─" * 45)
    print(f"{total} credential path(s), {incomplete} missing required metadata")
    if incomplete:
        print("Paths marked ! need describing — see platform-root-custody.md.")
    return 1 if incomplete else 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except InventoryError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(2)