Add scripts/report_workload_join.py — measure the lane-to-workload join

Computes the join rather than asserting it, and offers both plausible workload
segments per path instead of a single positional guess, because the path
convention is inconsistent.

Measured result: 1 of 27 lanes matches a declared workload. The workload
declaration surface exists, but it covers almost none of the credential estate.

Read-only; touches no secret value and no live system.

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

scripts/report_workload_join.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""Report the lane -> workload join, for the security-zone model (ZONE-WP-0001-T03).
+
+A security zone's subject is the **workload**, not the lane — policy is about the
+running thing and whoever answers for it. ops-warden's catalog is a credential
+surface, so on its own it cannot name that subject. The other side of the join
+lives in `rapp-*/declarations/rapp.yaml`, which declares `workload_identity`
+along with `data_classification` and `criticality`; ops-warden's
+`registry/policy/security-posture.yaml` then maps classification to a minimum
+maturity via `dataclass_floor`.
+
+This script computes the join and reports its coverage. It asserts nothing it
+cannot derive: a lane whose workload cannot be established is reported as
+unmatched rather than guessed, because the unmatched set is the informative
+output — those are lanes existing for something that is not a declared workload.
+
+Read-only. Touches no secret value and no live system.
+
+Usage:
+  python scripts/report_workload_join.py [--rapp-root ~] [--json]
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+_SRC = Path(__file__).resolve().parent.parent / "src"
+if _SRC.is_dir() and str(_SRC) not in sys.path:
+    sys.path.insert(0, str(_SRC))
+
+import yaml  # noqa: E402
+
+
+def load_rapp_workloads(root: Path) -> Dict[str, Dict[str, Any]]:
+    """workload name -> declaration, from every rapp-*/declarations/rapp.yaml."""
+    out: Dict[str, Dict[str, Any]] = {}
+    for decl in sorted(root.glob("rapp-*/declarations/rapp.yaml")):
+        try:
+            data = yaml.safe_load(decl.read_text()) or {}
+        except yaml.YAMLError:
+            continue
+        identity = data.get("workload_identity") or {}
+        name = identity.get("name")
+        if not name:
+            continue
+        out[str(name)] = {
+            "rapp_id": data.get("rapp_id"),
+            "data_classification": data.get("data_classification"),
+            "criticality": data.get("criticality"),
+            "readiness_state": data.get("readiness_state"),
+            "bound_reefs": data.get("bound_reefs"),
+        }
+    return out
+
+
+def candidate_workloads(path_template: Optional[str]) -> List[str]:
+    """Workload names a path could plausibly name — never a single guess.
+
+    The convention is `platform/workloads/<domain>/<workload>/<bundle>`, but it
+    is not applied consistently: `platform/workloads/rapp-qonto/keycape-client`
+    has one level fewer, so position alone cannot say which segment is the
+    workload. Both are offered and matched against declarations.
+    """
+    if not path_template or "<" in path_template:
+        return []
+    parts = [p for p in path_template.split("/") if p]
+    if len(parts) >= 4 and parts[1] == "workloads":
+        return [parts[3], parts[2]]
+    if parts and parts[0] == "tenants" and len(parts) >= 3:
+        return [parts[2], parts[1]]
+    return []
+
+
+def build(catalog_path: Path, rapp_root: Path) -> Dict[str, Any]:
+    entries = (yaml.safe_load(catalog_path.read_text()) or {}).get("entries", [])
+    declared = load_rapp_workloads(rapp_root)
+    floor = (
+        yaml.safe_load(
+            (catalog_path.parent.parent / "policy" / "security-posture.yaml").read_text()
+        )
+        or {}
+    ).get("dataclass_floor", {})
+
+    matched, unmatched, no_path = [], [], []
+    for entry in entries:
+        lane = entry.get("id")
+        cands = candidate_workloads(entry.get("path_template"))
+        if not cands:
+            no_path.append({"lane": lane, "risk": entry.get("risk")})
+            continue
+        hit = next((c for c in cands if c in declared), None)
+        if hit is None:
+            unmatched.append(
+                {"lane": lane, "risk": entry.get("risk"), "candidates": cands}
+            )
+            continue
+        decl = declared[hit]
+        cls = decl.get("data_classification")
+        matched.append(
+            {
+                "lane": lane,
+                "workload": hit,
+                "risk": entry.get("risk"),
+                "data_classification": cls,
+                "criticality": decl.get("criticality"),
+                "min_maturity": floor.get(cls),
+                "unmapped_classification": bool(cls) and cls not in floor,
+            }
+        )
+    return {
+        "declared_workloads": declared,
+        "matched": matched,
+        "unmatched": unmatched,
+        "no_path": no_path,
+        "dataclass_floor": floor,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--rapp-root", type=Path, default=Path.home())
+    parser.add_argument("--json", action="store_true")
+    parser.add_argument(
+        "--catalog",
+        type=Path,
+        default=Path(__file__).resolve().parent.parent
+        / "registry"
+        / "routing"
+        / "catalog.yaml",
+    )
+    args = parser.parse_args()
+
+    report = build(args.catalog, Path(os.path.expanduser(str(args.rapp_root))))
+    if args.json:
+        print(json.dumps(report, indent=2, sort_keys=True))
+        return 0
+
+    print("lane -> workload join\n")
+    print(f"  declared workloads (rapp-*):  {len(report['declared_workloads'])}")
+    print(f"  lanes matched to a workload:  {len(report['matched'])}")
+    print(f"  lanes unmatched:              {len(report['unmatched'])}")
+    print(f"  lanes with no usable path:    {len(report['no_path'])}\n")
+
+    if report["matched"]:
+        print("MATCHED")
+        for m in report["matched"]:
+            flag = "  <-- classification unmapped by dataclass_floor" if m[
+                "unmapped_classification"
+            ] else ""
+            print(
+                f"  {m['lane']:34} -> {str(m['workload']):16} "
+                f"{str(m['data_classification']):13} crit={str(m['criticality']):9} "
+                f"min={str(m['min_maturity']):4} risk={m['risk']}{flag}"
+            )
+    if report["unmatched"]:
+        print("\nUNMATCHED — a lane exists for something no rapp declares")
+        for u in report["unmatched"]:
+            print(
+                f"  {u['lane']:34} risk={str(u['risk']):9} "
+                f"candidates={', '.join(u['candidates'])}"
+            )
+    if report["no_path"]:
+        print("\nNO USABLE PATH — not a KV lane, or a pattern rather than an address")
+        print("  " + ", ".join(n["lane"] for n in report["no_path"]))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())