"""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} env KEYCAPE_{client_id:env}_X", "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_env_format_spec_renders_like_keycape_client(catalog): params = dict(CASE["spec"]["parameters"], client_id="my-app") plan = plan_case(_case(parameters=params), _type(), catalog) login = next(s for s in plan.steps if s.id == "login") assert login.entry_point == "apply my-app env KEYCAPE_MY_APP_X" @pytest.mark.parametrize("template", ["{client_id:upper}", "{client_id:>9}", "{client_id!r}"]) def test_refuse_unknown_format_spec_or_conversion(catalog, template): doc = _type() doc["spec"]["follow_through"][0]["entry_point"] = template with pytest.raises(CaseError, match="format spec|conversions"): plan_case(_case(), 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 import hashlib digest = hashlib.sha256(case_file.read_bytes()).hexdigest() assert f"case-ref: c@sha256:{digest}" 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