ops-warden/tests/test_case_plan.py

"""WARDEN-WP-0044-T02: plan a KnownSecurityCase run."""
from __future__ import annotations

import copy
import os
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(os.environ.get("WARDEN_TEST_NETKINGDOM_ROOT",
                                Path(__file__).resolve().parents[2] / "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