tegwick
7ce58ae638
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0291a-1e87-7151-9934-fcbfe3f65eb1
185 lines
7.2 KiB
Python
185 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Report explicit lane -> workload resolution for security-zones_v0.1.
|
|
|
|
The catalog owner declares whether each lane is workload-applicable. Managed
|
|
deployables use the exact Repo Manager v1 ``(rapp_id, name, deployable?)``
|
|
reference. Independently governed operational workloads use ``name`` plus an
|
|
owner declaration reference. Unknown and not-applicable are explicit results.
|
|
|
|
This script never parses a credential path, consults ``owner_repo`` as an
|
|
identity hint, or substitutes a repository name. It reads declarations only.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
def load_rapp_workloads(root: Path) -> dict[tuple[str, str], dict[str, Any]]:
|
|
"""Exact Repo Manager v1 key -> authoritative declaration projection."""
|
|
out: dict[tuple[str, 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 {}
|
|
rapp_id = data.get("rapp_id")
|
|
name = identity.get("name")
|
|
if not rapp_id or not name:
|
|
continue
|
|
deployables = []
|
|
for member in (data.get("composition") or {}).get("member_repos") or []:
|
|
deployables.extend(str(value) for value in member.get("deployables") or [])
|
|
out[(str(rapp_id), str(name))] = {
|
|
"source": str(decl),
|
|
"deployables": sorted(set(deployables)),
|
|
"data_classification": data.get("data_classification"),
|
|
"criticality": data.get("criticality"),
|
|
"readiness_state": data.get("readiness_state"),
|
|
}
|
|
return out
|
|
|
|
|
|
def _direct_declaration_path(
|
|
declaration_ref: str, *, catalog_path: Path, estate_root: Path
|
|
) -> Path:
|
|
ref = Path(declaration_ref)
|
|
if ref.is_absolute():
|
|
return ref
|
|
local = catalog_path.resolve().parents[2] / ref
|
|
return local if local.exists() else estate_root / ref
|
|
|
|
|
|
def _resolve_direct(
|
|
ref: dict[str, Any], *, catalog_path: Path, estate_root: Path
|
|
) -> tuple[dict[str, Any] | None, str | None]:
|
|
source = _direct_declaration_path(
|
|
str(ref["declaration_ref"]), catalog_path=catalog_path, estate_root=estate_root
|
|
)
|
|
if not source.exists():
|
|
return None, f"declaration not found: {source}"
|
|
try:
|
|
declaration = yaml.safe_load(source.read_text()) or {}
|
|
except yaml.YAMLError as exc:
|
|
return None, f"invalid declaration YAML: {exc}"
|
|
identity = declaration.get("workload_identity") or {}
|
|
if identity.get("name") != ref.get("name"):
|
|
return None, (
|
|
f"declared workload_identity.name={identity.get('name')!r}, "
|
|
f"expected {ref.get('name')!r}"
|
|
)
|
|
context = (declaration.get("zones") or {}).get("context", {})
|
|
return {
|
|
"source": str(source),
|
|
"data_classification": context.get("data_classification"),
|
|
"criticality": context.get("criticality"),
|
|
"maturity": context.get("maturity"),
|
|
"declared_zone": (declaration.get("zones") or {}).get("membership"),
|
|
}, None
|
|
|
|
|
|
def build(catalog_path: Path, estate_root: Path) -> dict[str, Any]:
|
|
entries = (yaml.safe_load(catalog_path.read_text()) or {}).get("entries", [])
|
|
managed = load_rapp_workloads(estate_root)
|
|
posture_path = catalog_path.parent.parent / "policy" / "security-posture.yaml"
|
|
floor = (yaml.safe_load(posture_path.read_text()) or {}).get("dataclass_floor", {})
|
|
|
|
resolved: list[dict[str, Any]] = []
|
|
unknown: list[dict[str, Any]] = []
|
|
not_applicable: list[dict[str, Any]] = []
|
|
for entry in entries:
|
|
lane = str(entry.get("id"))
|
|
ref = entry.get("workload_ref") or {}
|
|
applicability = ref.get("applicability")
|
|
if applicability == "not-applicable":
|
|
not_applicable.append({"lane": lane, "reason": ref.get("reason")})
|
|
continue
|
|
if applicability != "applicable":
|
|
unknown.append({"lane": lane, "reason": "applicability missing or invalid"})
|
|
continue
|
|
if ref.get("unknown_reason"):
|
|
unknown.append({"lane": lane, "reason": ref["unknown_reason"]})
|
|
continue
|
|
|
|
projection: dict[str, Any] | None
|
|
error: str | None = None
|
|
if ref.get("rapp_id"):
|
|
key = (str(ref.get("rapp_id")), str(ref.get("name")))
|
|
projection = managed.get(key)
|
|
if projection is None:
|
|
error = f"Repo Manager reference does not resolve: {key[0]}/{key[1]}"
|
|
elif ref.get("deployable") and ref["deployable"] not in projection["deployables"]:
|
|
error = f"deployable {ref['deployable']!r} is not declared by {key[0]}/{key[1]}"
|
|
else:
|
|
projection, error = _resolve_direct(
|
|
ref, catalog_path=catalog_path, estate_root=estate_root
|
|
)
|
|
if error or projection is None:
|
|
unknown.append({"lane": lane, "reason": error or "reference unresolved"})
|
|
continue
|
|
|
|
classification = projection.get("data_classification")
|
|
resolved.append(
|
|
{
|
|
"lane": lane,
|
|
"workload_ref": ref,
|
|
"source": projection.get("source"),
|
|
"data_classification": classification,
|
|
"criticality": projection.get("criticality"),
|
|
"maturity": projection.get("maturity") or floor.get(classification),
|
|
"declared_zone": projection.get("declared_zone"),
|
|
"unmapped_classification": bool(classification) and classification not in floor,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"contract": "helixforge.workload-reference/v1",
|
|
"resolved": resolved,
|
|
"unknown": unknown,
|
|
"not_applicable": not_applicable,
|
|
"ok": len(resolved) + len(unknown) + len(not_applicable) == len(entries),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--estate-root", "--rapp-root", dest="estate_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, args.estate_root)
|
|
if args.json:
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
return 0 if report["ok"] else 1
|
|
|
|
print("explicit lane -> workload resolution\n")
|
|
print(f" resolved: {len(report['resolved'])}")
|
|
print(f" unknown: {len(report['unknown'])}")
|
|
print(f" not-applicable: {len(report['not_applicable'])}\n")
|
|
for row in report["resolved"]:
|
|
ref = row["workload_ref"]
|
|
prefix = f"{ref.get('rapp_id')}/" if ref.get("rapp_id") else ""
|
|
print(f"RESOLVED {row['lane']:34} -> {prefix}{ref.get('name')}")
|
|
for row in report["unknown"]:
|
|
print(f"UNKNOWN {row['lane']:34} {row['reason']}")
|
|
for row in report["not_applicable"]:
|
|
print(f"NOT-APPLICABLE {row['lane']:34} {row['reason']}")
|
|
return 0 if report["ok"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|