ops-mason/tests/test_case_build.py

import json
import shutil
from pathlib import Path

import pytest
import yaml

from ops_mason import case_build as cb
from ops_mason import case_plan as cp

NK = cp.DEFAULT_NK_ROOT
TYPE = NK / "security-case-types" / "oidc-browser-relying-party"
CASE = NK / "security-cases" / "forgejo-netkingdom-oidc.yaml"
pytestmark = pytest.mark.skipif(not CASE.is_file(), reason="net-kingdom checkout not present")


@pytest.fixture
def types_root(tmp_path):
    root = tmp_path / "types"
    shutil.copytree(TYPE, root / TYPE.name)
    f = root / TYPE.name / "type.yaml"
    data = yaml.safe_load(f.read_text())
    data["spec"]["status"] = "implemented"
    f.write_text(yaml.safe_dump(data, sort_keys=False))
    return root


def _case(tmp_path, **spec):
    case = yaml.safe_load(CASE.read_text())
    case["spec"].update(spec)
    p = tmp_path / "case.yaml"
    p.write_text(yaml.safe_dump(case, sort_keys=False))
    return p


def _plan(tmp_path, case, types_root, source="live"):
    doc, _, _ = cp.plan_from_case(case, types_root=types_root, structure=cp.Structure(source=source))
    p = tmp_path / "plan.md"
    p.write_text(doc)
    return p


def test_builds_bundle_naming_case_type_and_acceptance(tmp_path, types_root):
    case = _case(tmp_path)
    out, log = tmp_path / "bundle", tmp_path / "audit.jsonl"
    r = cb.build_from_case(case, _plan(tmp_path, case, types_root), out, types_root=types_root, audit_log=log)
    assert r["case"] == "forgejo-netkingdom-oidc" and r["type"] == yaml.safe_load(case.read_text())["spec"]["type"]
    ccrs = {Path(p).name: yaml.safe_load(Path(p).read_text()) for p in r["ccrs"]}
    eso = ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]
    assert eso["status"] == "proposed" and eso["openbao"]["auth"]["role"] == "external-secrets-keycape-forgejo"
    assert eso["openbao"]["auth"]["bound_claims_confirmed"] is False
    consumer = ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml"]
    assert consumer["status"] == "in_flight" and consumer["in_flight"]["missing_fields"] == ["openbao.auth"]
    for doc in ccrs.values():
        assert doc["security_case"]["acceptance_record"] and "case acceptance replaces" in doc["requester"]["reason"]
    hcl = (out / "openbao/policies/workload-kv-read-keycape-forgejo-eso.hcl").read_text()
    assert 'path "platform/data/workloads/keycape/forgejo"' in hcl and '["read"]' in hcl and "forgejo-netkingdom-oidc" in hcl
    # token_self_lifecycle: true needs the self stanzas, or the platform applier refuses the file.
    assert 'path "auth/token/lookup-self"' in hcl and 'path "auth/token/revoke-self"' in hcl
    stanzas = [line for line in hcl.splitlines() if line.startswith("path ")]
    assert len(stanzas) == 3
    rollback = yaml.safe_load((out / "rollback.yaml").read_text())
    assert rollback["removes_exactly"][-1] == {"kind": "kv_path", "name": "platform/workloads/keycape/forgejo"}
    assert rollback["removes_exactly"][0]["kind"] == "catalog_entry"
    manifest = json.loads((out / "manifest.json").read_text())
    assert set(manifest["files"]) >= {"rollback.yaml", "catalog-proposal.yaml"}
    record = json.loads(log.read_text().splitlines()[-1])
    assert record["objects"]["event"] == "case-build-prepared" and record["approved_by"].startswith("case acceptance")


def test_refuses_case_without_acceptance(tmp_path, types_root):
    case = _case(tmp_path, status="proposed")
    case.write_text(case.read_text())  # acceptance block remains but status is proposed
    with pytest.raises(cb.CaseBuildRefused):
        cb.build_from_case(case, _plan(tmp_path, case, types_root), tmp_path / "b", types_root=types_root,
                           audit_log=tmp_path / "a.jsonl")


def test_refuses_plan_that_is_not_reviewed(tmp_path, types_root):
    case = _case(tmp_path)
    with pytest.raises(cb.CaseBuildRefused, match="not 'reviewed'"):
        cb.build_from_case(case, _plan(tmp_path, case, types_root, source="none"), tmp_path / "b",
                           types_root=types_root, audit_log=tmp_path / "a.jsonl")
    assert not (tmp_path / "b").exists()


def test_refuses_when_case_changed_since_review(tmp_path, types_root):
    case = _case(tmp_path)
    plan = _plan(tmp_path, case, types_root)
    data = yaml.safe_load(case.read_text())
    data["spec"]["parameters"]["display_name"] = "Forgejo (renamed)"
    case.write_text(yaml.safe_dump(data, sort_keys=False))
    with pytest.raises(cb.CaseBuildRefused, match="digest"):
        cb.build_from_case(case, plan, tmp_path / "b", types_root=types_root, audit_log=tmp_path / "a.jsonl")


def test_refuses_to_overwrite_an_existing_bundle(tmp_path, types_root):
    case = _case(tmp_path)
    plan = _plan(tmp_path, case, types_root)
    (tmp_path / "b").mkdir()
    with pytest.raises(FileExistsError):
        cb.build_from_case(case, plan, tmp_path / "b", types_root=types_root, audit_log=tmp_path / "a.jsonl")


def test_platform_validator_accepts_the_drafts(tmp_path, types_root):
    if not (cb.RAILIANCE_PLATFORM / "scripts" / "credential-change.py").is_file():
        pytest.skip("railiance-platform checkout not present")
    case = _case(tmp_path)
    r = cb.build_from_case(case, _plan(tmp_path, case, types_root), tmp_path / "b", types_root=types_root,
                           audit_log=tmp_path / "a.jsonl")
    assert cb.validate_with_platform([Path(p) for p in r["ccrs"]]) == []


def test_two_role_backed_readers_keep_distinct_ccrs_and_complete_structure(tmp_path, types_root):
    from ops_mason import case_structure
    type_path = types_root / TYPE.name / "type.yaml"
    doc = yaml.safe_load(type_path.read_text())
    doc["spec"]["materialization"].insert(-1, {
        "id": "consumer-eso-role", "kind": "openbao_role",
        "when": "client_type == confidential",
        "template": "external-secrets-keycape-{client_id}-{consuming_service}",
        "body": "auth: kubernetes\nbound_service_account_names: [forgejo-keycape-forgejo-eso]\nbound_service_account_namespaces: [forgejo]\ntoken_policies: [workload-kv-read-keycape-{client_id}-{consuming_service}]\ntoken_no_default_policy: true\ntoken_ttl: 15m\ntoken_max_ttl: 15m\n",
    })
    doc["spec"]["rollback"].append({"reverses": "consumer-eso-role", "when": "client_type == confidential", "action": "remove this created role only"})
    type_path.write_text(yaml.safe_dump(doc, sort_keys=False))
    case = _case(tmp_path)
    plan = _plan(tmp_path, case, types_root)
    out = tmp_path / "bundle"
    result = cb.build_from_case(case, plan, out, types_root=types_root, audit_log=tmp_path / "a.jsonl")
    assert len(result["ccrs"]) == len(set(result["ccrs"])) == 2
    ccrs = {Path(p).name: yaml.safe_load(Path(p).read_text()) for p in result["ccrs"]}
    assert ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]["openbao"]["auth"]["role"] == "external-secrets-keycape-forgejo"
    consumer = ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml"]
    assert ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]["target"]["workload"] == "keycape"
    assert consumer["target"]["workload"] == "forgejo"
    assert consumer["openbao"]["auth"]["role"] == "external-secrets-keycape-forgejo-forgejo"
    assert consumer["openbao"]["auth"]["bound_claims"] == {"service_account_names": ["forgejo-keycape-forgejo-eso"], "service_account_namespaces": ["forgejo"]}
    assert not consumer["openbao"]["auth"]["bound_claims_confirmed"]
    deactivate = consumer["lifecycle"]["deactivate"].lower()
    for guard in ["created", "preexisting", "consumer reference", "revoke issued", "verified expiry", "no new login", "no remint"]:
        assert guard in deactivate
    assert "delete the role, then policy" in ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]["lifecycle"]["deactivate"]
    assert cb.validate_with_platform([Path(p) for p in result["ccrs"]]) == []
    request = case_structure.prepare(case, plan, out, types_root=types_root)
    assert len(request.objects) == 6
    assert {o.id for o in request.objects} >= {"eso-role", "consumer-eso-role"}


def test_duplicate_source_output_refuses_before_overwriting_first_ccr(tmp_path, types_root, monkeypatch):
    original = cb._ccr
    def same_identity(*args, **kwargs):
        doc = original(*args, **kwargs)
        doc["id"] = "CCR-DRAFT-COLLISION"
        return doc
    monkeypatch.setattr(cb, "_ccr", same_identity)
    case = _case(tmp_path)
    out = tmp_path / "bundle"
    with pytest.raises(cb.CaseBuildRefused, match="duplicate output identity"):
        cb.build_from_case(case, _plan(tmp_path, case, types_root), out, types_root=types_root, audit_log=tmp_path / "a.jsonl")
    first = yaml.safe_load((out / "credential-change-requests/CCR-DRAFT-COLLISION.yaml").read_text())
    assert first["openbao"]["policy_name"] == "workload-kv-read-keycape-forgejo-eso"
    assert not (out / "manifest.json").exists()


def test_generated_consumer_lifecycle_requires_created_only_token_recovery(types_root):
    case = yaml.safe_load(CASE.read_text())
    policy = {"id": "consumer-reader-policy", "template": "workload-kv-read-keycape-forgejo-forgejo"}
    role = {"template": "external-secrets-keycape-forgejo-forgejo", "body": "auth: kubernetes\nbound_service_account_names: [forgejo-keycape-forgejo-eso]\nbound_service_account_namespaces: [forgejo]\ntoken_policies: [workload-kv-read-keycape-forgejo-forgejo]\n"}
    provenance = {"case": "forgejo-netkingdom-oidc", "type": "fixture-source-only", "render_digest": "fixture", "accepted_by": ["fixture-owner"], "accepted_at": "fixture-date", "acceptance_record": "fixture-not-authority"}
    doc = cb._ccr(case, {}, policy, role, "platform/workloads/keycape/forgejo", provenance)
    for guard in ["created", "preexisting", "consumer reference", "revoke issued", "verified expiry", "no new login", "no remint"]:
        assert guard in doc["lifecycle"]["deactivate"].lower()