diff --git a/scripts/report_workload_join.py b/scripts/report_workload_join.py new file mode 100755 index 0000000..8fce7dd --- /dev/null +++ b/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///`, 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())