coulomb/ops-mason
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plan --case: a case's own built objects are not collisions on re-render

Browse source 
Objects listed in the case's committed build bundle (same case id) are shown
as already present from this case; others still collide. 2 tests; suite 177.
Forgejo plan marked stale after net-kingdom's type change (new render digest).

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-10 13:17:22 +02:00
parent 1c07b65a3c
commit ab17d859cb
3 changed files with 81 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

54
src/ops_mason/case_plan.py
View file

@ -130,22 +130,49 @@ def _role_mount(item: dict) -> str:
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."""
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(f"policy {name} already exists")
found.append((item, 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")
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(f"KV path {name} already holds an entry")
found.append((item, 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")
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"):
@ -153,7 +180,8 @@ def acceptance_ref(case: dict) -> dict | 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:
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 = {
@ -181,6 +209,9 @@ def plan_document(case: dict, type_data: dict, rendered: dict, structure: Struct
"## 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"]]
@ -205,9 +236,12 @@ def plan_document(case: dict, type_data: dict, rendered: dict, structure: Struct
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) -> tuple[str, list[str], dict]:
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)
collisions = review(rendered, structure)
return plan_document(case, type_data, rendered, structure, collisions), collisions, rendered
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

31
tests/test_case_plan.py
View file

@ -78,7 +78,8 @@ 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)
doc, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing,
builds_root=tmp_path / "no-builds")
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
@ -124,3 +125,31 @@ def test_allow_unimplemented_refused_for_non_acknowledged_type(tmp_path, types_r
f.write_text(yaml.safe_dump(data, sort_keys=False))
with pytest.raises(cp.CasePlanError):
cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=cp.Structure(), allow_unimplemented=True)
def test_objects_from_the_cases_own_bundle_are_not_collisions(tmp_path, types_root):
builds = tmp_path / "builds" / "case-forgejo-netkingdom-oidc"
builds.mkdir(parents=True)
(builds / "rollback.yaml").write_text(yaml.safe_dump({
"security_case": {"case": "forgejo-netkingdom-oidc"},
"removes_exactly": [{"kind": "openbao_policy", "name": "workload-kv-read-keycape-forgejo-eso"},
{"kind": "openbao_role", "name": "external-secrets-keycape-forgejo"}]}))
existing = cp.Structure(policies={"workload-kv-read-keycape-forgejo-eso", "workload-kv-read-keycape-forgejo-forgejo"},
roles={"kubernetes/external-secrets-keycape-forgejo"}, source="live")
doc, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing,
builds_root=tmp_path / "builds")
assert collisions == ["policy workload-kv-read-keycape-forgejo-forgejo already exists"]
assert "Already present from this case's own build bundle" in doc
assert "- openbao_role external-secrets-keycape-forgejo" in doc
def test_another_cases_bundle_does_not_excuse_a_collision(tmp_path, types_root):
builds = tmp_path / "builds" / "case-forgejo-netkingdom-oidc"
builds.mkdir(parents=True)
(builds / "rollback.yaml").write_text(yaml.safe_dump({
"security_case": {"case": "someone-else"},
"removes_exactly": [{"kind": "openbao_policy", "name": "workload-kv-read-keycape-forgejo-eso"}]}))
existing = cp.Structure(policies={"workload-kv-read-keycape-forgejo-eso"}, source="live")
_, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing,
builds_root=tmp_path / "builds")
assert collisions == ["policy workload-kv-read-keycape-forgejo-eso already exists"]

7
workplans/MASON-WP-0010-security-case-materialization.md
View file

@ -202,3 +202,10 @@ net-kingdom changed the type to `KEYCAPE_{client_id:env}_CLIENT_SECRET`
(ops-warden `23ba584`), so the render digest will change; and a live re-render
of a partly applied case will report its own objects as collisions. The KV
value waits on mint-secret (RPF-WP-0054).
2026-10-10 (T03 follow-up): `plan --case` now recognises objects listed in the
case's own committed bundle (`builds/case-<id>/rollback.yaml`, same case id) as
"already present from this case's own build", not collisions; anything else
existing still collides (2 tests, suite 177 passed). The committed Forgejo plan
is stale: net-kingdom's `{client_id:env}` change moves the render digest from
`fb8b114d…` to `5e97931e…`. Re-render it live under the next survey grant.