access-engine/tools/gen_ops_warden_fetch.py

#!/usr/bin/env python3
"""Generate the ops-warden secret-read pin from ops-warden's routing catalog.

The flex-auth-ops-warden-fetch pin answers warden's ``check_fetch_policy``
(action ``read`` on resource type ``secret``, resource.system = the lane's
``owner_repo``). Which (need_id, owner_repo) pairs are readable is not authored
here: it is every catalog entry with ``status: active``, ``exec_capable: true``
and ``lane: secret``. This script writes that set into three places:

  examples/ops-warden-fetch/registry_snapshot.json   systems + one resource per lane
  examples/ops-warden-fetch/policy_package.md        the GENERATED catalog_lanes block
  values/ops-warden-fetch.yaml                       one caller binding per owner system

Usage:
  python3 tools/gen_ops_warden_fetch.py [--catalog PATH]          # write
  python3 tools/gen_ops_warden_fetch.py [--catalog PATH] --check  # exit 1 on drift

The default catalog is ~/ops-warden/registry/routing/catalog.yaml. The image
digest already in the values file is preserved. No secret material is read:
the catalog is a pointer layer (ops-warden ADR-0001).
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parent.parent
EXAMPLE_DIR = ROOT / "examples" / "ops-warden-fetch"
REGISTRY = EXAMPLE_DIR / "registry_snapshot.json"
POLICY = EXAMPLE_DIR / "policy_package.md"
VALUES = ROOT / "values" / "ops-warden-fetch.yaml"
DEFAULT_CATALOG = Path(os.path.expanduser("~/ops-warden/registry/routing/catalog.yaml"))

CALLER_PRINCIPAL = "system:serviceaccount:ops-warden:ops-warden"
TENANT = "tenant:platform"
BEGIN = "# BEGIN GENERATED catalog_lanes"
END = "# END GENERATED catalog_lanes"


def readable_lanes(catalog_path: Path) -> list[tuple[str, str]]:
    doc = yaml.safe_load(catalog_path.read_text())
    lanes = []
    for entry in doc.get("entries", []):
        if entry.get("status") != "active":
            continue
        if entry.get("exec_capable") is not True:
            continue
        if entry.get("lane") != "secret":
            continue
        need_id, owner = entry.get("id"), entry.get("owner_repo")
        if not need_id or not owner:
            raise SystemExit(f"catalog entry without id/owner_repo: {entry!r}")
        lanes.append((need_id, owner))
    if not lanes:
        raise SystemExit("catalog yielded no readable lanes; refusing to write an empty policy")
    return sorted(lanes)


def render_registry(lanes: list[tuple[str, str]]) -> str:
    systems = sorted({owner for _, owner in lanes})
    snapshot = {
        "systems": [
            {
                "id": system,
                "name": f"{system} secrets read through ops-warden",
                "resource_types": [
                    {"name": "secret", "scope_level": "Resource", "planes": ["Secret", "Audit"]}
                ],
                "actions": [
                    {
                        "name": "read",
                        "capabilities": ["View", "Use"],
                        "planes": ["Secret", "Audit"],
                        "exposure_modes": ["SecretMaterial"],
                    }
                ],
                "caring_profiles": ["caring-0.4.0-rc2"],
                "metadata": {
                    "tenant": TENANT,
                    "conduit": "ops-warden",
                    "generated_by": "tools/gen_ops_warden_fetch.py",
                },
            }
            for system in systems
        ],
        "resource_manifests": [
            {
                "id": f"ops-warden-fetch-{system}",
                "system": system,
                "resources": [
                    {
                        "id": f"secret:{need_id}",
                        "type": "secret",
                        "labels": ["ops-warden-fetch", "lane:secret"],
                        "owner": f"repo:{system}",
                        "attributes": {"need_id": need_id, "catalog_lane": "secret"},
                    }
                    for need_id, owner in lanes
                    if owner == system
                ],
            }
            for system in systems
        ],
        "tenants": [{"id": TENANT, "name": "Platform Tenant"}],
    }
    return json.dumps(snapshot, indent=2) + "\n"


def render_policy(current: str, lanes: list[tuple[str, str]]) -> str:
    body = "\n".join(f'  "{need_id}": "{owner}",' for need_id, owner in lanes)
    block = f"{BEGIN} (tools/gen_ops_warden_fetch.py; do not edit by hand)\ncatalog_lanes := {{\n{body}\n}}\n{END}"
    pattern = re.compile(re.escape(BEGIN) + r".*?" + re.escape(END), re.S)
    if not pattern.search(current):
        raise SystemExit(f"{POLICY}: generated markers not found")
    return pattern.sub(lambda _: block, current, count=1)


def render_values(current: str | None, lanes: list[tuple[str, str]]) -> str:
    digest = ""
    if current:
        match = re.search(r"^\s*digest:\s*(\S*)\s*$", current, re.M)
        if match:
            digest = match.group(1).strip("\"'")
    systems = sorted({owner for _, owner in lanes})
    bindings = [f"{system}={CALLER_PRINCIPAL}" for system in systems]
    extra = "".join(f"\n    - {b}" for b in bindings[1:]) or " []"
    return f"""# ops-warden secret-read pin (CUST-WP-0075, access-engine ADHOC-2026-10-03).
# Answers warden check_fetch_policy: action read on resource type secret, system =
# the lane's owner_repo. Separate from flex-auth-ops-warden (SSH signing) so each
# decision class rolls independently (FLEX-WP-0016). Isolated: the workstation
# reaches it through the ops-bridge tunnel flex-auth-ops-warden-fetch-railiance01
# (127.0.0.1:19091). Every owner system is bound to the ops-warden ServiceAccount,
# so only ops-warden may ask; callerAuth enforce as on the signing pin.
#
# GENERATED by tools/gen_ops_warden_fetch.py from ops-warden's routing catalog,
# except image.digest, which the script preserves. Regenerate, do not hand-edit.
name: flex-auth-ops-warden-fetch
image:
  repository: forgejo.coulomb.social/coulomb/flex-auth
  digest: {digest or '""'}
args:
  - serve
  - --addr
  - 0.0.0.0:8080
  - --registry
  - /opt/flex-auth/examples/ops-warden-fetch/registry_snapshot.json
  - --policy
  - /opt/flex-auth/examples/ops-warden-fetch/policy_package.md
callerAuth:
  mode: enforce
  kubernetesURL: https://10.43.0.1
  binding: {bindings[0]}
  extraBindings:{extra}
consumer:
  isolated: true
"""


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG)
    parser.add_argument("--check", action="store_true", help="report drift, write nothing")
    args = parser.parse_args()

    lanes = readable_lanes(args.catalog)
    values_current = VALUES.read_text() if VALUES.exists() else None
    outputs = {
        REGISTRY: render_registry(lanes),
        POLICY: render_policy(POLICY.read_text(), lanes),
        VALUES: render_values(values_current, lanes),
    }
    drift = [p for p, text in outputs.items() if not p.exists() or p.read_text() != text]
    if args.check:
        for path in drift:
            print(f"drift: {path.relative_to(ROOT)}", file=sys.stderr)
        print(f"{len(lanes)} readable lanes; {'drift' if drift else 'in sync'}")
        return 1 if drift else 0
    for path in drift:
        path.write_text(outputs[path])
        print(f"wrote {path.relative_to(ROOT)}")
    print(f"{len(lanes)} readable lanes across {len({o for _, o in lanes})} owner systems")
    return 0


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