diff --git a/src/warden/case_plan.py b/src/warden/case_plan.py new file mode 100644 index 0000000..d0a2293 --- /dev/null +++ b/src/warden/case_plan.py @@ -0,0 +1,194 @@ +"""Plan a KnownSecurityCase run (WARDEN-WP-0044-T02). + +``warden case plan `` 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/.yaml beside security-case-types//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 ' == ' 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) diff --git a/src/warden/cli.py b/src/warden/cli.py index 80fd467..50569bc 100644 --- a/src/warden/cli.py +++ b/src/warden/cli.py @@ -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.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) diff --git a/tests/test_case_plan.py b/tests/test_case_plan.py new file mode 100644 index 0000000..61b196b --- /dev/null +++ b/tests/test_case_plan.py @@ -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 diff --git a/workplans/WARDEN-WP-0043-platform-supplier-credentials.md b/workplans/WARDEN-WP-0043-platform-supplier-credentials.md index e1d3d49..ce8ac26 100644 --- a/workplans/WARDEN-WP-0043-platform-supplier-credentials.md +++ b/workplans/WARDEN-WP-0043-platform-supplier-credentials.md @@ -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 diff --git a/workplans/WARDEN-WP-0044-security-case-follow-through.md b/workplans/WARDEN-WP-0044-security-case-follow-through.md index e5a32e1..538de99 100644 --- a/workplans/WARDEN-WP-0044-security-case-follow-through.md +++ b/workplans/WARDEN-WP-0044-security-case-follow-through.md @@ -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 [--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 ` == ` 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_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" ```