WARDEN-WP-0044-T02: warden case plan; close WP-0044-T01 and WP-0043-T02
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s

warden case plan resolves an accepted, materialized SecurityCase against its
acknowledged type: warden, attended and owner steps, type defaults, conditions,
rendered entry points and verification. Executes nothing; refuses unmaterialized
cases, missing or unreviewed lanes and secret or type-fixed parameters.

T01 closes on net-kingdom's R8 option 1 and the founder's acknowledgement of
the type (memo r4). WP-0043-T02 closes on railiance-platform b77cb41.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 427672@bnt-lap001
Assistant-Session: a0a2367e-4697-440c-a4e5-9713c009f6d8
This commit is contained in:
tegwick 2026-10-09 14:46:17 +02:00
parent 6258d73732
commit 9a85405e7a
5 changed files with 446 additions and 8 deletions

194
src/warden/case_plan.py Normal file
View file

@ -0,0 +1,194 @@
"""Plan a KnownSecurityCase run (WARDEN-WP-0044-T02).
``warden case plan <case.yaml>`` checks that a case is accepted and materialized
(canon ``security-case-types_v0.1``: ``proposed -> accepted -> materialized ->
live``), checks its type with :mod:`warden.case_type`, resolves each
follow-through step against the case's parameters, and prints the run: which
steps ops-warden executes as the caller's proxy, which run in an attended
session, which the owner runs, and what verification follows.
Read-only: it parses YAML and the catalog, executes nothing, and never handles a
secret value. Parameters are rendered into entry points and checks, so a
parameter whose type declares ``sensitivity: secret`` is refused outright.
"""
from __future__ import annotations
import re
import string
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional
import yaml
from warden.case_type import Finding, check_follow_through
from warden.routing.catalog import Catalog
PLANNABLE_CASE_STATUS = "materialized"
PLANNABLE_TYPE_STATUSES = ("acknowledged", "implemented")
_WHEN = re.compile(r"^\s*([a-z_][a-z0-9_]*)\s*==\s*(\S+)\s*$")
class CaseError(ValueError):
"""The case cannot be planned; the message says why."""
@dataclass(frozen=True)
class PlannedStep:
id: str
owner: str
mode: str # "warden" | "attended" | "owner"
entry_point: str
auth: str
lane: Optional[str]
value_handling: str
ready_check: Optional[str] = None
def as_dict(self) -> dict:
return {k: v for k, v in self.__dict__.items() if v is not None}
@dataclass
class CasePlan:
case_id: str
type_ref: str
steps: List[PlannedStep] = field(default_factory=list)
skipped: List[str] = field(default_factory=list)
attended_sessions: List[str] = field(default_factory=list)
verification: List[dict] = field(default_factory=list)
def as_dict(self) -> dict:
return {
"case": self.case_id,
"type": self.type_ref,
"steps": [s.as_dict() for s in self.steps],
"skipped": self.skipped,
"attended_sessions": self.attended_sessions,
"verification": self.verification,
}
def load_case(path: Path) -> dict:
with open(path) as fh:
doc = yaml.safe_load(fh)
if not isinstance(doc, dict) or doc.get("kind") != "SecurityCase":
raise CaseError(f"{path}: not a SecurityCase document")
return doc
def default_type_path(case_path: Path, type_ref: str) -> Path:
"""net-kingdom layout: security-cases/<case>.yaml beside security-case-types/<id>/type.yaml."""
type_id = type_ref.split("@", 1)[0]
return case_path.resolve().parent.parent / "security-case-types" / type_id / "type.yaml"
def _applies(when: Optional[str], params: Dict[str, str]) -> bool:
if not when:
return True
match = _WHEN.match(when)
if not match:
raise CaseError(f"unsupported condition {when!r}; only '<parameter> == <value>' is planned")
name, expected = match.groups()
if name not in params:
raise CaseError(f"condition {when!r} names unknown parameter '{name}'")
return str(params[name]).lower() == expected.lower()
def _render(text: Optional[str], params: Dict[str, str], where: str) -> Optional[str]:
if text is None:
return None
names = {f for _, f, _, _ in string.Formatter().parse(text) if f}
missing = sorted(names - set(params))
if missing:
raise CaseError(f"{where}: parameters {missing} are not set by the case")
return text.format(**params)
def plan_case(case_doc: dict, type_doc: dict, catalog: Optional[Catalog]) -> CasePlan:
case = case_doc.get("spec") or {}
spec = type_doc.get("spec") or {}
case_id = str(case.get("id", ""))
type_ref = str(case.get("type", ""))
status = case.get("status")
if status != PLANNABLE_CASE_STATUS:
hint = {"proposed": "it is not accepted", "accepted": "ops-mason has not materialized it",
"live": "it is already live", "retired": "it is retired"}.get(status, "unknown status")
raise CaseError(f"case {case_id} is '{status}' ({hint}); planning needs '{PLANNABLE_CASE_STATUS}'")
if not case.get("acceptance"):
raise CaseError(f"case {case_id} carries no acceptance record")
want = f"{spec.get('id')}@{spec.get('version')}"
if type_ref != want:
raise CaseError(f"case names type {type_ref}, type file is {want}")
if spec.get("status") not in PLANNABLE_TYPE_STATUSES:
raise CaseError(f"type {want} is '{spec.get('status')}', not acknowledged")
report = check_follow_through(type_doc, catalog)
if report.findings:
raise CaseError("type is not plannable:\n" + "\n".join(_line(f) for f in report.findings))
declared = {p["name"]: p for p in spec.get("parameters") or [] if isinstance(p, dict) and "name" in p}
params = {k: str(v) for k, v in (case.get("parameters") or {}).items()}
unknown = sorted(set(params) - set(declared))
if unknown:
raise CaseError(f"case sets parameters the type does not declare: {unknown}")
fixed = sorted(k for k in params if declared[k].get("case_settable") is False)
if fixed:
raise CaseError(f"case sets parameters the type fixes: {fixed}")
secret = sorted(k for k in params if declared[k].get("sensitivity") == "secret")
if secret:
raise CaseError(f"case carries secret-sensitivity parameters {secret}; values never enter a plan")
for name, decl in declared.items():
if name not in params and "default" in decl:
params[name] = str(decl["default"])
plan = CasePlan(case_id=case_id, type_ref=type_ref)
for step in spec.get("follow_through") or []:
sid = step["id"]
if not _applies(step.get("when"), params):
plan.skipped.append(sid)
continue
mode = "warden" if step["warden_executes"] else "attended" if step["attended"] else "owner"
where = f"follow_through[{sid}]"
plan.steps.append(PlannedStep(
id=sid, owner=_render(str(step["owner"]), params, where), mode=mode,
entry_point=_render(step["entry_point"], params, where),
auth=_render(step["auth"], params, where),
lane=step.get("lane"), value_handling=step["value_handling"],
ready_check=_render(step.get("ready_check"), params, where),
))
if mode == "attended" and step["lane"] not in plan.attended_sessions:
plan.attended_sessions.append(step["lane"])
for item in spec.get("verification") or []:
if not _applies(item.get("when"), params):
continue
plan.verification.append({
"id": item.get("id"),
"check": _render(item.get("check"), params, f"verification[{item.get('id')}]"),
"expect": item.get("expect"),
})
return plan
def _line(f: Finding) -> str:
return f"[{f.kind}] {f.code} {f.where}: {f.message}"
def format_plan(plan: CasePlan) -> str:
labels = {"warden": "warden (caller proxy)", "attended": "ATTENDED", "owner": "owner"}
lines = [f"case {plan.case_id} ({plan.type_ref}): plan only, nothing executed", ""]
for n, s in enumerate(plan.steps, 1):
lane = f" lane={s.lane}" if s.lane else ""
lines.append(f"{n}. {s.id} [{labels[s.mode]}] owner={s.owner}{lane} values={s.value_handling}")
lines.append(f" entry: {s.entry_point}")
lines.append(f" auth: {s.auth}")
if s.ready_check:
lines.append(f" ready: {s.ready_check}")
if plan.skipped:
lines.append(f"skipped (condition false): {', '.join(plan.skipped)}")
lines.append(f"attended sessions: {', '.join(plan.attended_sessions) or '-'}")
lines.append("verification:")
lines.extend(f" - {v['id']}: {v['check']} (expect {v['expect']})" for v in plan.verification)
return "\n".join(lines)

View file

@ -2243,3 +2243,32 @@ def case_check_type(
if report.findings:
console.print(summarize(report.findings), markup=False)
raise typer.Exit(0 if report.plannable else 1)
@case_app.command("plan")
def case_plan(
case_file: Annotated[Path, typer.Argument(help="SecurityCase yaml (accepted and materialized)")],
type_file: Annotated[
Optional[Path],
typer.Option("--type", help="type.yaml (default: ../security-case-types/<type>/type.yaml beside the case)"),
] = None,
output_json: Annotated[bool, typer.Option("--json", help="Output JSON")] = False,
) -> None:
"""Print a case's run (warden, attended, owner steps and verification); executes nothing (exit 1 if refused)."""
from warden.case_plan import CaseError, default_type_path, format_plan, load_case, plan_case
from warden.case_type import load_type
try:
case_doc = load_case(case_file)
type_path = type_file or default_type_path(case_file, str((case_doc.get("spec") or {}).get("type", "")))
plan = plan_case(case_doc, load_type(type_path), _load_catalog())
except CaseError as exc:
err.print(str(exc), markup=False)
raise typer.Exit(1)
except (OSError, ValueError, yaml.YAMLError) as exc:
err.print(f"[red]{exc}[/red]")
raise typer.Exit(2)
if output_json:
print(json.dumps(plan.as_dict(), indent=2))
else:
console.print(format_plan(plan), markup=False)

184
tests/test_case_plan.py Normal file
View file

@ -0,0 +1,184 @@
"""WARDEN-WP-0044-T02: plan a KnownSecurityCase run."""
from __future__ import annotations
import copy
from pathlib import Path
import pytest
import yaml
from typer.testing import CliRunner
from warden.case_plan import CaseError, plan_case
from warden.cli import app
from warden.routing.catalog import load_catalog
LIVE_CASE = Path.home() / "net-kingdom/security-cases/forgejo-netkingdom-oidc.yaml"
TYPE = {
"apiVersion": "netkingdom.io/security-case-type/v0.1",
"kind": "SecurityCaseType",
"spec": {
"id": "t",
"version": "1.0",
"status": "acknowledged",
"parameters": [
{"name": "client_id", "case_settable": True, "sensitivity": "operational"},
{"name": "client_type", "case_settable": True, "sensitivity": "operational"},
{"name": "consumer", "case_settable": True, "sensitivity": "operational"},
{"name": "issuer", "case_settable": False, "default": "https://kc.example"},
],
"follow_through": [
{"id": "proxy", "owner": "o", "entry_point": "ensure {client_id}", "auth": "caller",
"lane": "key-cape-oidc-login", "warden_executes": True, "attended": False,
"value_handling": "none"},
{"id": "mint", "owner": "o", "entry_point": "mint {client_id}", "auth": "caller",
"lane": "key-cape-oidc-login", "warden_executes": True, "attended": False,
"value_handling": "generated_in_custody", "when": "client_type == confidential"},
{"id": "login", "owner": "o", "entry_point": "apply {client_id}", "auth": "operator",
"lane": "key-cape-oidc-login", "warden_executes": False, "attended": True,
"value_handling": "none", "attended_justification": "live registry is a Secret"},
{"id": "handover", "owner": "{consumer}", "entry_point": "use {issuer}", "auth": "a",
"warden_executes": False, "attended": False, "value_handling": "none",
"ready_check": "{consumer} ready"},
],
"verification": [
{"id": "v", "check": "GET {issuer} for {client_id}", "expect": "present",
"reads_secret_value": False},
{"id": "vc", "check": "custody {client_id}", "expect": "deny",
"reads_secret_value": False, "when": "client_type == confidential"},
],
"rollback": [{"reverses": s} for s in ("proxy", "mint", "login", "handover")],
},
}
CASE = {
"apiVersion": "netkingdom.io/security-case/v0.1",
"kind": "SecurityCase",
"spec": {
"id": "c",
"type": "t@1.0",
"status": "materialized",
"acceptance": {"at": "2026-10-09", "record": "r"},
"parameters": {"client_id": "forgejo", "client_type": "public", "consumer": "railiance-apps"},
},
}
@pytest.fixture(scope="module")
def catalog():
return load_catalog()
def _case(**spec):
doc = copy.deepcopy(CASE)
doc["spec"].update(spec)
return doc
def _type():
return copy.deepcopy(TYPE)
def test_plan_resolves_modes_params_and_defaults(catalog):
plan = plan_case(_case(), _type(), catalog)
modes = {s.id: s.mode for s in plan.steps}
assert modes == {"proxy": "warden", "login": "attended", "handover": "owner"}
assert plan.skipped == ["mint"]
assert plan.attended_sessions == ["key-cape-oidc-login"]
handover = next(s for s in plan.steps if s.id == "handover")
assert handover.owner == "railiance-apps"
assert handover.entry_point == "use https://kc.example"
assert handover.ready_check == "railiance-apps ready"
assert [v["id"] for v in plan.verification] == ["v"]
assert plan.verification[0]["check"] == "GET https://kc.example for forgejo"
def test_condition_true_includes_step_and_check(catalog):
params = dict(CASE["spec"]["parameters"], client_type="confidential")
plan = plan_case(_case(parameters=params), _type(), catalog)
assert "mint" in [s.id for s in plan.steps] and not plan.skipped
assert [v["id"] for v in plan.verification] == ["v", "vc"]
@pytest.mark.parametrize("status,needle", [
("proposed", "not accepted"),
("accepted", "not materialized"),
("live", "already live"),
])
def test_refuse_case_not_materialized(catalog, status, needle):
with pytest.raises(CaseError, match=needle):
plan_case(_case(status=status), _type(), catalog)
def test_refuse_case_without_acceptance(catalog):
with pytest.raises(CaseError, match="acceptance"):
plan_case(_case(acceptance=None), _type(), catalog)
def test_refuse_type_mismatch_and_unacknowledged_type(catalog):
with pytest.raises(CaseError, match="type file is"):
plan_case(_case(type="t@2.0"), _type(), catalog)
doc = _type()
doc["spec"]["status"] = "proposed"
with pytest.raises(CaseError, match="not acknowledged"):
plan_case(_case(), doc, catalog)
def test_refuse_missing_catalog_lane(catalog):
doc = _type()
doc["spec"]["follow_through"][0]["lane"] = "no-such-lane"
with pytest.raises(CaseError, match="LANE-MISSING"):
plan_case(_case(), doc, catalog)
def test_refuse_unreviewed_catalog_lane(catalog):
draft = next((e.id for e in catalog.entries if not (e.is_active and e.reviewed)), None)
if draft is None:
pytest.skip("catalog has no inactive or unreviewed lane")
doc = _type()
doc["spec"]["follow_through"][0]["lane"] = draft
with pytest.raises(CaseError, match="LANE-UNREVIEWED"):
plan_case(_case(), doc, catalog)
def test_refuse_bad_parameters(catalog):
with pytest.raises(CaseError, match="does not declare"):
plan_case(_case(parameters=dict(CASE["spec"]["parameters"], extra="x")), _type(), catalog)
with pytest.raises(CaseError, match="fixes"):
plan_case(_case(parameters=dict(CASE["spec"]["parameters"], issuer="https://evil")), _type(), catalog)
params = {k: v for k, v in CASE["spec"]["parameters"].items() if k != "consumer"}
with pytest.raises(CaseError, match="not set by the case"):
plan_case(_case(parameters=params), _type(), catalog)
def test_refuse_secret_parameter(catalog):
doc = _type()
doc["spec"]["parameters"].append({"name": "pw", "case_settable": True, "sensitivity": "secret"})
with pytest.raises(CaseError, match="secret-sensitivity"):
plan_case(_case(parameters=dict(CASE["spec"]["parameters"], pw="hunter2")), doc, catalog)
def test_refuse_unsupported_condition(catalog):
doc = _type()
doc["spec"]["follow_through"][0]["when"] = "client_type in (a, b)"
with pytest.raises(CaseError, match="unsupported condition"):
plan_case(_case(), doc, catalog)
def test_cli_plan_shows_attended_and_exit_codes(tmp_path):
case_file, type_file = tmp_path / "case.yaml", tmp_path / "type.yaml"
case_file.write_text(yaml.safe_dump(CASE))
type_file.write_text(yaml.safe_dump(TYPE))
runner = CliRunner()
ok = runner.invoke(app, ["case", "plan", str(case_file), "--type", str(type_file)])
assert ok.exit_code == 0, ok.output
assert "ATTENDED" in ok.output and "nothing executed" in ok.output
case_file.write_text(yaml.safe_dump(_case(status="accepted")))
refused = runner.invoke(app, ["case", "plan", str(case_file), "--type", str(type_file)])
assert refused.exit_code == 1
@pytest.mark.skipif(not LIVE_CASE.exists(), reason="net-kingdom checkout absent")
def test_live_forgejo_case_is_refused_until_materialized():
result = CliRunner().invoke(app, ["case", "plan", str(LIVE_CASE)])
assert result.exit_code == 1

View file

@ -9,7 +9,7 @@ flavor: implementation
owner: grok
topic_slug: platform-suppliers
created: "2026-10-08"
updated: "2026-10-08"
updated: "2026-10-09"
depends_on: []
related:
- repo:railiance-platform
@ -66,9 +66,8 @@ Done 2026-10-08: catalog `platform-supplier-credentials` plus
```task
id: WARDEN-WP-0043-T02
status: wait
status: done
priority: high
blocked_on: message-from:railiance-platform
state_hub_task_id: "c384b228-e710-580d-a3de-a3046a86e666"
```
@ -77,6 +76,11 @@ and `tenants/` for a delegated change. `platform/suppliers/` needs that
allowlist widened before a CCR can create these paths. ops-warden does not
apply the change.
Done 2026-10-09: railiance-platform `b77cb41` maps mount `platform` to
`platform/workloads/` and `platform/suppliers/`. Under `suppliers/` it requires
exactly two non-empty segments, and tests cover `platform/suppliers/ionos/main`.
The IONOS CCR can now be drafted against this shape.
## Provision supplier accounts
```task

View file

@ -41,7 +41,7 @@ caller's identity.
```task
id: WARDEN-WP-0044-T01
status: progress
status: done
depends_on: [NK-WP-0048-T03]
priority: high
state_hub_task_id: "ef0920fc-418e-52cb-acff-9a5c0c728bf5"
@ -84,11 +84,15 @@ credential. `keycape-client-secret-generate` needs a generate-in-custody entry
point from railiance-platform, and the broker does not have one yet. Both were
sent to their owners.
2026-10-09: done. net-kingdom took R8 option 1, so `ensure-groups` runs attended
in the `register-client` session and `identity-provisioner-groups` is dropped
(net-kingdom `0e86f95`). The founder acknowledged the type with memo r4 (hub
decision `e418263e`, NK-WP-0048-T05 done).
```task
id: WARDEN-WP-0044-T02
status: wait
status: done
depends_on: [WARDEN-WP-0044-T01]
blocking_reason: "Planning a run needs the agreed follow-through section."
priority: high
state_hub_task_id: "5f68c991-f490-58f8-a121-92f74073d272"
```
@ -103,11 +107,34 @@ verification is. No secret values; no execution.
Tests: refuse an unaccepted or unmaterialized case; refuse a step whose catalog
entry is missing or not reviewed; show attended steps explicitly.
2026-10-09: done. `warden case plan <case.yaml> [--type] [--json]`
(`src/warden/case_plan.py`, 15 tests in `tests/test_case_plan.py`). It refuses:
- a case that is not `materialized`, or has no acceptance record
- a type that is mismatched or not acknowledged
- any `case check-type` finding, including a missing or unreviewed lane
- undeclared, type-fixed or secret-sensitivity parameters
- a placeholder the case does not set, or an unsupported `when`
It fills type defaults (`issuer`), evaluates `<param> == <value>` conditions,
renders owner, entry point, auth and ready_check, groups attended steps per
lane, and lists the applicable verification.
The live Forgejo case is correctly refused: it is `accepted`, not
materialized. A materialized copy, with a stand-in lane for
`keycape-client-secret-generate`, renders six steps: two attended steps plus
`register-client` in one `key-cape-oidc-login` session, one warden step
(`mint-secret`), and two owner steps. Seven checks follow.
Observation for net-kingdom (not blocking): `bind-env` runs before
`register-client`, in type order. Its entry carries a literal
`KEYCAPE_<CLIENT_ID>_CLIENT_SECRET`, which is not a `{param}` placeholder, so it
reaches the plan unrendered.
```task
id: WARDEN-WP-0044-T03
status: wait
depends_on: [WARDEN-WP-0044-T02]
blocking_reason: "Executing follows a reviewed plan."
depends_on: [WARDEN-WP-0044-T02, MASON-WP-0010-T02]
blocking_reason: "The Forgejo case must be materialized (MASON-WP-0010-T02), and lane keycape-client-secret-generate needs railiance-platform's generate-in-custody entry point."
priority: high
state_hub_task_id: "bcbf6c91-2c04-5c16-ae8c-5d6c06fd6b89"
```