ops-warden/scripts/report_workload_join.py

#!/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())