MASON-WP-0010-T02 done: ops-mason plan --case renders a construction plan from a SecurityCase

Uses net-kingdom's case validator and renderer, adds a read-only collision
review against live structure, and puts the case acceptance in place of the
executive summary. 8 tests; suite 167 passed. WP-0010 now blocked on the type
being implemented; WP-0004-T04 blocker made precise.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 327203@bnt-lap001
Assistant-Session: 55028ae3-4914-4bd1-8f5c-815f3527aef0
This commit is contained in:
tegwick 2026-10-09 19:01:28 +02:00
parent 21b5a076f6
commit 4dafcf5ed4
5 changed files with 355 additions and 6 deletions

199
src/ops_mason/case_plan.py Normal file
View file

@ -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) # "<mount>/<role>"
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

View file

@ -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":

109
tests/test_case_plan.py Normal file
View file

@ -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

View file

@ -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"
```

View file

@ -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 <case.yaml> [--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"
```