"""Construction plans derived from a SecurityCase (MASON-WP-0010-T02). For a KnownSecurityCase the plan is rendered from the type's materialization templates instead of being written by hand, and the case acceptance takes the place of a fresh executive summary. Validation and rendering are net-kingdom's (canon security-case-types_v0.1); ops-mason imports that validator rather than copying it, then runs its own consistency review against existing structure. Nothing here writes to OpenBao. The live survey is read-only and metadata-only. """ from __future__ import annotations import importlib.util import json import os import subprocess from dataclasses import dataclass, field from pathlib import Path import yaml DEFAULT_NK_ROOT = Path(os.environ.get("NETKINGDOM_ROOT", Path.home() / "net-kingdom")) VALIDATOR = Path("tools/security-case-types/security_case_validator.py") DEFAULT_CATALOG = Path.home() / "ops-warden" / "registry" / "routing" / "catalog.yaml" class CasePlanError(RuntimeError): """The case cannot become a construction plan; never treated as an empty plan.""" def __init__(self, message: str, issues: list[str] | None = None): super().__init__(message) self.issues = issues or [] @dataclass class Structure: """Existing names the review compares against: metadata only, never values.""" policies: set[str] = field(default_factory=set) roles: set[str] = field(default_factory=set) # "/" kv_paths: set[str] = field(default_factory=set) catalog_ids: set[str] = field(default_factory=set) source: str = "none" def load_validator(nk_root: Path = DEFAULT_NK_ROOT): path = nk_root / VALIDATOR if not path.is_file(): raise CasePlanError(f"net-kingdom case validator not found at {path}") spec = importlib.util.spec_from_file_location("security_case_validator", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _issues(validator, issues) -> list[str]: return [f"{i.path}: {i.message}" for i in issues if i.level == "ERROR"] def render(case_path: Path, nk_root: Path = DEFAULT_NK_ROOT, types_root: Path | None = None, allow_unimplemented: bool = False) -> tuple[dict, dict, dict]: """Validate the case against its implemented type and render it. Returns (case, type, render). allow_unimplemented mirrors net-kingdom's validator flag: only the reference case that takes a type to `implemented` may render against an acknowledged type. """ v = load_validator(nk_root) case = yaml.safe_load(Path(case_path).read_text()) or {} types_root = types_root or nk_root / "security-case-types" issues, type_data, _ = v.load_type(types_root, str((case.get("spec") or {}).get("type", ""))) if type_data is None: raise CasePlanError("case names no loadable type", _issues(v, issues)) decls, issues = v.discover_declarations(type_data, nk_root.parent) if v.has_errors(issues): raise CasePlanError("type's provider declarations are missing", _issues(v, issues)) declarations, load_issues = v.composer.load_declarations(decls) if v.has_errors(load_issues): raise CasePlanError("type's provider declarations do not load", _issues(v, load_issues)) status = type_data.get("spec", {}).get("status") if allow_unimplemented and status != "acknowledged": raise CasePlanError(f"--allow-unimplemented applies only to an acknowledged type, not {status!r}") issues, rendered = v.render_case(case, type_data, declarations, require_implemented=not allow_unimplemented) if rendered is None: raise CasePlanError("case refused", _issues(v, issues)) return case, type_data, rendered def _bao_json(args: list[str]) -> object | None: env = dict(os.environ) env.setdefault("BAO_ADDR", "http://127.0.0.1:18200") r = subprocess.run(["bao", *args, "-format=json"], capture_output=True, text=True, env=env) if r.returncode != 0: return None try: return json.loads(r.stdout or "null") except json.JSONDecodeError: return None def live_structure(rendered: dict, catalog: Path = DEFAULT_CATALOG) -> Structure: """Read-only survey under the ops-mason-survey grant: lists and metadata only.""" s = Structure(source="live") policies = _bao_json(["policy", "list"]) if policies is None: raise CasePlanError("live survey failed: cannot list policies (grant missing or expired?)") s.policies = set(policies) for mount in {_role_mount(i) for i in rendered["materialization"] if i["kind"] == "openbao_role"}: roles = _bao_json(["list", f"auth/{mount}/role"]) or [] s.roles |= {f"{mount}/{r}" for r in roles} for item in rendered["materialization"]: if item["kind"] == "kv_path": mount, rest = item["template"].split("/", 1) if _bao_json(["kv", "metadata", "get", f"-mount={mount}", rest]) is not None: s.kv_paths.add(item["template"]) s.catalog_ids = catalog_ids(catalog) return s def catalog_ids(catalog: Path = DEFAULT_CATALOG) -> set[str]: if not Path(catalog).is_file(): return set() data = yaml.safe_load(Path(catalog).read_text()) or {} entries = data.get("entries") if isinstance(data, dict) else data return {str(e.get("id")) for e in entries or [] if isinstance(e, dict)} def _role_mount(item: dict) -> str: body = yaml.safe_load(item.get("body") or "") or {} return str(body.get("auth") or "kubernetes") DEFAULT_BUILDS = Path("builds") def own_objects(case_id: str, builds_root: Path = DEFAULT_BUILDS) -> set[tuple[str, str]]: """Objects this case's own committed build bundle created (kind, name). A live re-render of a partly built case finds them in OpenBao; they are this case's work, not someone else's structure, so they are not collisions. """ rollback = Path(builds_root) / f"case-{case_id}" / "rollback.yaml" if not rollback.is_file(): return set() data = yaml.safe_load(rollback.read_text()) or {} if (data.get("security_case") or {}).get("case") != case_id: return set() return {(o["kind"], o["name"]) for o in data.get("removes_exactly", [])} def _existing(rendered: dict, structure: Structure) -> list[tuple[dict, str]]: found = [] for item in rendered["materialization"]: name, kind = item["template"], item["kind"] if kind == "openbao_policy" and name in structure.policies: found.append((item, f"policy {name} already exists")) elif kind == "openbao_role" and f"{_role_mount(item)}/{name}" in structure.roles: found.append((item, f"role auth/{_role_mount(item)}/role/{name} already exists")) elif kind == "kv_path" and name in structure.kv_paths: found.append((item, f"KV path {name} already holds an entry")) elif kind == "catalog_entry" and name in structure.catalog_ids: found.append((item, f"catalog entry {name} already exists")) return found def review(rendered: dict, structure: Structure, own: set[tuple[str, str]] = frozenset()) -> list[str]: """Collisions with existing structure. A collision is a finding for a human, not an auto-merge.""" return [msg for item, msg in _existing(rendered, structure) if (item["kind"], item["template"]) not in own] def already_built(rendered: dict, structure: Structure, own: set[tuple[str, str]]) -> list[str]: return [f"{item['kind']} {item['template']}" for item, _ in _existing(rendered, structure) if (item["kind"], item["template"]) in own] def acceptance_ref(case: dict) -> dict | None: acc = (case.get("spec") or {}).get("acceptance") or {} if not acc.get("record"): return None return {"record": acc["record"], "at": acc.get("at"), "by": [a.get("role") for a in acc.get("by", [])]} def plan_document(case: dict, type_data: dict, rendered: dict, structure: Structure, collisions: list[str], built: list[str] = ()) -> str: cspec, tspec = case["spec"], type_data["spec"] acc = acceptance_ref(case) front = { "id": f"case-{cspec['id']}", "demand_source": f"SecurityCase {cspec['id']} (owner {cspec['owner']['repo']})", "consumer_repo": cspec["owner"]["repo"], "credential_type": "security-case", "derived_from_type": rendered["type"], "render_digest": rendered["digest"], "status": "reviewed" if acc and not collisions and structure.source == "live" else "draft", "case_acceptance": acc, "reference_case": tspec.get("status") != "implemented", "approved_by": None, "approved_at": None, } lines = ["---", yaml.safe_dump(front, sort_keys=False).rstrip(), "---", "", f"# Construction plan (derived-from-type): {cspec['id']}", "", f"Rendered from `{rendered['type']}` ({tspec['title']}). Do not edit by hand:", f"re-run `ops-mason plan --case` instead. Render digest `{rendered['digest']}`.", ""] if tspec.get("status") != "implemented": lines += [f"**Reference case:** the type is `{tspec.get('status')}`, not yet `implemented`. This case", "is the one whose live materialization takes the type to `implemented`.", ""] lines += [ "## 1. Demand", "", tspec["purpose"].strip(), "", "## 2. Existing-structure survey", "", f"Survey source: {structure.source} (metadata only).", ""] lines += [f"- **Collision:** {c}" for c in collisions] or ["No collision with existing policies, roles, KV paths or catalog entries."] if built: lines += ["", "Already present from this case's own build bundle (not collisions):", ""] lines += [f"- {b}" for b in built] lines += ["", "## 3. Objects (from the type's materialization)", "", "| Item | Kind | Name |", "|---|---|---|"] lines += [f"| {i['id']} | {i['kind']} | `{i['template']}` |" for i in rendered["materialization"]] for i in rendered["materialization"]: if i.get("body"): lines += ["", f"### {i['id']}", "", "```", i["body"].rstrip(), "```"] lines += ["", "## 4. Review notes", "", "Shape reviewed once, when the type was acknowledged; this review checks only", "this case's names against existing structure (section 2).", "", "Follow-through (ops-warden):", ""] lines += [f"- {s['id']} — {s['owner']}{' (attended)' if s.get('attended') else ''}: {s['entry_point']}" for s in rendered["follow_through"]] lines += ["", "Verification:", ""] + [f"- {c['id']} ({c['expect']}): {c['check']}" for c in rendered["verification"]] lines += ["", "Rollback, in order:", ""] + [f"- {r['action']}" for r in rendered["rollback"]] lines += ["", "## 5. Case acceptance (replaces the executive summary)", ""] if acc: lines.append(f"Accepted by {', '.join(acc['by'])} at {acc['at']}; record `{acc['record']}`.") else: lines.append("Not accepted yet. The case must carry its acceptance record before a build.") return "\n".join(lines) + "\n" def plan_from_case(case_path: Path, *, nk_root: Path = DEFAULT_NK_ROOT, types_root: Path | None = None, structure: Structure | None = None, catalog: Path = DEFAULT_CATALOG, allow_unimplemented: bool = False, builds_root: Path = DEFAULT_BUILDS) -> tuple[str, list[str], dict]: case, type_data, rendered = render(case_path, nk_root, types_root, allow_unimplemented) if structure is None: structure = live_structure(rendered, catalog) own = own_objects(case["spec"]["id"], builds_root) collisions = review(rendered, structure, own) built = already_built(rendered, structure, own) return plan_document(case, type_data, rendered, structure, collisions, built), collisions, rendered