access-engine: secret-read evaluator for warden fetch checks (CUST-WP-0075)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 3775@bnt-lap001
Assistant-Session: cfacb6a5-ad5b-4f79-ac67-053e3b0d34a3

tools/gen_ops_warden_fetch.py

@@ -0,0 +1,192 @@
+#!/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())