diff --git a/src/ops_mason/case_plan.py b/src/ops_mason/case_plan.py new file mode 100644 index 0000000..e4e766e --- /dev/null +++ b/src/ops_mason/case_plan.py @@ -0,0 +1,199 @@ +"""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) -> tuple[dict, dict, dict]: + """Validate the case against its implemented type and render it. Returns (case, type, render).""" + 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)) + issues, rendered = v.render_case(case, type_data, declarations, require_implemented=True) + 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") + + +def review(rendered: dict, structure: Structure) -> list[str]: + """Collisions with existing structure. A collision is a finding for a human, not an auto-merge.""" + found = [] + for item in rendered["materialization"]: + name, kind = item["template"], item["kind"] + if kind == "openbao_policy" and name in structure.policies: + found.append(f"policy {name} already exists") + elif kind == "openbao_role" and f"{_role_mount(item)}/{name}" in structure.roles: + found.append(f"role auth/{_role_mount(item)}/role/{name} already exists") + elif kind == "kv_path" and name in structure.kv_paths: + found.append(f"KV path {name} already holds an entry") + elif kind == "catalog_entry" and name in structure.catalog_ids: + found.append(f"catalog entry {name} already exists") + return found + + +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]) -> 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 else "draft", + "case_acceptance": acc, + "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']}`.", "", + "## 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."] + 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) -> tuple[str, list[str], dict]: + case, type_data, rendered = render(case_path, nk_root, types_root) + if structure is None: + structure = live_structure(rendered, catalog) + collisions = review(rendered, structure) + return plan_document(case, type_data, rendered, structure, collisions), collisions, rendered diff --git a/src/ops_mason/cli.py b/src/ops_mason/cli.py index 891556f..158397e 100644 --- a/src/ops_mason/cli.py +++ b/src/ops_mason/cli.py @@ -39,11 +39,38 @@ def _parser() -> argparse.ArgumentParser: apply_parser.add_argument("--readiness-repo", type=Path) apply_parser.add_argument("--activation", choices=("APPROVED", "BREAK_GLASS"), default="APPROVED") apply_parser.add_argument("--break-glass-reason") + + plan = families.add_parser("plan", help="construction plan derived from a SecurityCase") + plan.add_argument("--case", required=True, type=Path) + plan.add_argument("--out", type=Path, help="write the plan here (default: stdout)") + plan.add_argument("--no-live", action="store_true", + help="skip the live OpenBao survey (marks the plan's survey source 'none')") + plan.add_argument("--netkingdom-root", type=Path, default=None) return parser +def _plan(args) -> int: + from ops_mason import case_plan + + nk_root = args.netkingdom_root or case_plan.DEFAULT_NK_ROOT + try: + doc, collisions, rendered = case_plan.plan_from_case( + args.case, nk_root=nk_root, structure=case_plan.Structure() if args.no_live else None) + except case_plan.CasePlanError as exc: + print(json.dumps({"ok": False, "error": str(exc), "issues": exc.issues}), file=sys.stderr) + return 2 + if args.out: + args.out.write_text(doc) + else: + sys.stdout.write(doc) + print(json.dumps({"ok": not collisions, "digest": rendered["digest"], "collisions": collisions}), file=sys.stderr) + return 1 if collisions else 0 + + def main(argv: Sequence[str] | None = None) -> int: args = _parser().parse_args(argv) + if args.family == "plan": + return _plan(args) try: bundle = PlaneBundle.load(args.bundle) if args.command == "render": diff --git a/tests/test_case_plan.py b/tests/test_case_plan.py new file mode 100644 index 0000000..f1f7333 --- /dev/null +++ b/tests/test_case_plan.py @@ -0,0 +1,109 @@ +import shutil +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import yaml + +from ops_mason import case_plan as cp + +NK = cp.DEFAULT_NK_ROOT +TYPE = NK / "security-case-types" / "oidc-browser-relying-party" +CASE = TYPE / "examples" / "forgejo.yaml" +pytestmark = pytest.mark.skipif(not CASE.is_file(), reason="net-kingdom checkout not present") + + +@pytest.fixture +def types_root(tmp_path): + """The reference type as it will be once implemented; the live one is only acknowledged.""" + root = tmp_path / "types" + shutil.copytree(TYPE, root / TYPE.name) + f = root / TYPE.name / "type.yaml" + data = yaml.safe_load(f.read_text()) + data["spec"]["status"] = "implemented" + f.write_text(yaml.safe_dump(data, sort_keys=False)) + return root + + +def _case(tmp_path, **changes): + case = yaml.safe_load(CASE.read_text()) + for key, value in changes.items(): + if key == "acceptance": + case["spec"]["status"] = "accepted" + case["spec"]["acceptance"] = value + else: + case["spec"]["parameters"][key] = value + path = tmp_path / "case.yaml" + path.write_text(yaml.safe_dump(case, sort_keys=False)) + return path + + +def test_renders_reference_case_as_derived_plan(tmp_path, types_root): + doc, collisions, rendered = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=cp.Structure()) + assert collisions == [] + front = yaml.safe_load(doc.split("---")[1]) + assert front["derived_from_type"] == "oidc-browser-relying-party@1.0" + assert front["render_digest"] == rendered["digest"] and front["status"] == "draft" + names = {i["template"] for i in rendered["materialization"]} + assert {"platform/workloads/keycape/forgejo", "workload-kv-read-keycape-forgejo-eso", + "workload-kv-read-keycape-forgejo-forgejo", "external-secrets-keycape-forgejo", + "forgejo-keycape-client"} <= names + assert "Not accepted yet" in doc and "{client_id}" not in doc + + +def test_acceptance_replaces_executive_summary(tmp_path, types_root): + acc = {"at": "2026-10-08", "record": "informed-decision:forgejo-oidc", "by": [{"role": "owner"}, {"role": "founder"}]} + doc, _, _ = cp.plan_from_case(_case(tmp_path, acceptance=acc), types_root=types_root, structure=cp.Structure()) + front = yaml.safe_load(doc.split("---")[1]) + assert front["status"] == "reviewed" + assert front["case_acceptance"]["record"] == "informed-decision:forgejo-oidc" + + +def test_refuses_case_whose_type_is_not_implemented(tmp_path): + with pytest.raises(cp.CasePlanError) as e: + cp.plan_from_case(_case(tmp_path), structure=cp.Structure()) + assert any("only implemented types" in i for i in e.value.issues) + + +def test_refuses_parameter_outside_type_validation(tmp_path, types_root): + with pytest.raises(cp.CasePlanError) as e: + cp.plan_from_case(_case(tmp_path, redirect_uris="http://forgejo.example/cb"), types_root=types_root, + structure=cp.Structure()) + assert any("redirect_uris" in i for i in e.value.issues) + + +def test_flags_collision_with_existing_structure(tmp_path, types_root): + existing = cp.Structure(policies={"workload-kv-read-keycape-forgejo-eso"}, + roles={"kubernetes/external-secrets-keycape-forgejo"}, + catalog_ids={"forgejo-keycape-client"}, source="test") + doc, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing) + assert len(collisions) == 3 + assert yaml.safe_load(doc.split("---")[1])["status"] == "draft" + assert "**Collision:** policy workload-kv-read-keycape-forgejo-eso already exists" in doc + + +def test_live_survey_failure_is_not_an_empty_structure(tmp_path, types_root): + _, _, rendered = cp.render(_case(tmp_path), types_root=types_root) + with patch.object(cp.subprocess, "run", return_value=MagicMock(returncode=2, stdout="", stderr="denied")): + with pytest.raises(cp.CasePlanError): + cp.live_structure(rendered, catalog=tmp_path / "none.yaml") + + +def test_live_survey_reads_only_lists_and_metadata(tmp_path, types_root): + _, _, rendered = cp.render(_case(tmp_path), types_root=types_root) + calls = [] + + def fake(args, **kw): + calls.append(args) + return MagicMock(returncode=0 if args[1] in ("policy", "list") else 2, stdout='["default"]', stderr="") + with patch.object(cp.subprocess, "run", side_effect=fake): + s = cp.live_structure(rendered, catalog=tmp_path / "none.yaml") + assert s.policies == {"default"} and s.source == "live" + assert all(a[1] in ("policy", "list", "kv") for a in calls) + assert not any(a[1] == "kv" and a[2] != "metadata" for a in calls) + + +def test_cli_plan_refuses_unimplemented_type_with_exit_2(tmp_path, capsys): + from ops_mason.cli import main + assert main(["plan", "--case", str(_case(tmp_path)), "--no-live"]) == 2 + assert "only implemented types" in capsys.readouterr().err diff --git a/workplans/MASON-WP-0004-credential-inventory-descriptions.md b/workplans/MASON-WP-0004-credential-inventory-descriptions.md index f177eaf..758ad62 100644 --- a/workplans/MASON-WP-0004-credential-inventory-descriptions.md +++ b/workplans/MASON-WP-0004-credential-inventory-descriptions.md @@ -177,7 +177,7 @@ user-engine plan once its ESO rotation rehearsal is recorded. id: MASON-WP-0004-T04 status: wait depends_on: [MASON-WP-0004-T03] -blocking_reason: "Blocked on T03." +blocking_reason: "After T03. A scheduled --undescribed report needs an unattended metadata-read credential and a named reader; survey grants are human-minted and expire within an hour." priority: low state_hub_task_id: "9a0cea06-eb63-54ad-963d-db4ea79c7c66" ``` diff --git a/workplans/MASON-WP-0010-security-case-materialization.md b/workplans/MASON-WP-0010-security-case-materialization.md index 4e02ddf..85a85ae 100644 --- a/workplans/MASON-WP-0010-security-case-materialization.md +++ b/workplans/MASON-WP-0010-security-case-materialization.md @@ -4,12 +4,12 @@ type: workplan title: "Materialize SecurityCases from their type, with acceptance as the only gate" domain: infotech repo: ops-mason -status: active +status: blocked flavor: implementation owner: claude-code topic_slug: netkingdom created: "2026-10-08" -updated: "2026-10-08" +updated: "2026-10-09" depends_on: [] related_repos: - net-kingdom @@ -100,7 +100,7 @@ the type is `acknowledged` (net-kingdom `0e86f95`). T02 is unblocked. ```task id: MASON-WP-0010-T02 -status: todo +status: done depends_on: [MASON-WP-0010-T01, NK-WP-0048-T05] priority: high state_hub_task_id: "fd385537-7378-5539-8a6e-f45851296f42" @@ -118,11 +118,25 @@ Tests: render the reference case; refuse a case whose type is not `implemented`; refuse parameter values outside the type's validation; flag a collision with existing structure. +2026-10-09 done: `ops-mason plan --case [--out F] [--no-live]` +(`src/ops_mason/case_plan.py`). Validation and rendering are net-kingdom's own +`security_case_validator.py` (imported, not copied), with provider declarations +loaded through its composer. ops-mason adds the review against existing +structure: a read-only live survey (policy list, role lists, KV metadata, +ops-warden catalog ids) that fails loudly instead of returning an empty +structure, and flags each collision as a finding. Output is a construction +plan with `derived_from_type` and `render_digest`, and section 5 is the case +acceptance; status `reviewed` only when accepted and collision-free. Exit 0 +clean, 1 collisions, 2 refused. `tests/test_case_plan.py` (8, on a copy of the +type marked `implemented`); suite 167 passed. The live Forgejo case is refused +today, correctly: the type is `acknowledged`, not yet `implemented`. + ```task id: MASON-WP-0010-T03 status: wait depends_on: [MASON-WP-0010-T02] -blocking_reason: "Building from an accepted case follows the rendering." +blocked_on: message-from:net-kingdom +blocking_reason: "Needs the type implemented (net-kingdom) and the Forgejo case accepted; builds that write are founder-applied through the railiance-platform applier, so the build path is designed against that applier, and its acceptance is the Forgejo case materializing live." priority: high state_hub_task_id: "56cea6be-c613-586c-80cd-125340b374da" ``` @@ -142,7 +156,7 @@ path. id: MASON-WP-0010-T04 status: wait depends_on: [MASON-WP-0010-T03] -blocking_reason: "Adoption follows a working light path." +blocking_reason: "After T03, and only once a type fits: secrets-engine-openbao is a service (JWT) client, which oidc-browser-relying-party does not cover." priority: medium state_hub_task_id: "f6ad0d3c-8f47-5fdf-8115-6ca9d8a1fead" ```