ops-mason/tests/test_case_structure.py

"""Source structural projection and fake authenticated owner observations only."""

from dataclasses import asdict, replace
import hashlib
import json
import os
from pathlib import Path
import shutil
import pytest
import yaml
from ops_mason import case_plan, case_structure as structure

ROOT = Path("/home/worsch/ops-mason")
CASE = case_plan.DEFAULT_NK_ROOT / "security-cases/forgejo-netkingdom-oidc.yaml"


@pytest.fixture
def prepared(tmp_path):
    bundle = tmp_path / "bundle"
    shutil.copytree(ROOT / "builds/case-forgejo-netkingdom-oidc", bundle)
    plan = tmp_path / "plan.md"
    shutil.copy2(ROOT / "plans/case-forgejo-netkingdom-oidc.md", plan)
    # The committed historical bundle mislabels its consumer target keycape.
    # Preserve it in the owner checkout; corrected fixture metadata must match
    # the current generator rather than weakening the exact structural checker.
    rel = "credential-change-requests/CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml"
    ccr = yaml.safe_load((bundle / rel).read_text())
    ccr["target"]["workload"] = "forgejo"
    (bundle / rel).write_text(yaml.safe_dump(ccr, sort_keys=False, width=110))
    manifest = json.loads((bundle / "manifest.json").read_text())
    manifest["files"][rel] = hashlib.sha256((bundle / rel).read_bytes()).hexdigest()
    (bundle / "manifest.json").write_text(json.dumps(manifest, indent=1, sort_keys=True)+"\n")
    request = structure.prepare(CASE, plan, bundle, allow_unimplemented=True)
    return request, plan, bundle


def observations(request):
    result = []
    for obj in request.objects:
        disposition = "created" if obj.proof == "native-structure" else obj.proof
        result.append(
            structure.Observation(
                obj.id,
                obj.evidence_owner,
                obj.target,
                obj.expected_digest,
                disposition,
                "fixture-owner-evidence",
                "fixture-change-receipt" if disposition == "created" else None,
            )
        )
    return tuple(result)


def test_corrected_first_case_fixture_structure_and_full_execution_drift_remain(
    prepared,
):
    request, _, _ = prepared
    assert request.type_ref == "oidc-browser-relying-party@1.1"
    assert {o.proof for o in request.objects} == {
        "native-structure",
        "layout-only",
        "pointer-source",
    }
    assert len(request.objects) == 5
    assert "mint-secret" not in {o.id for o in request.objects}
    assert "secret-delivered" not in {o.id for o in request.objects}


@pytest.mark.parametrize(
    "change", ["extra-policy-rule", "role-binding", "catalog-pointer", "kv-inventory"]
)
def test_structural_mutation_refuses_even_after_attacker_rehashes_manifest(
    prepared, change
):
    _, plan, bundle = prepared
    if change == "extra-policy-rule":
        rel = "openbao/policies/workload-kv-read-keycape-forgejo-eso.hcl"
        (bundle / rel).write_text(
            (bundle / rel).read_text()
            + '\npath "platform/data/*" { capabilities = ["read"] }\n'
        )
    elif change == "role-binding":
        rel = "credential-change-requests/CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"
        data = yaml.safe_load((bundle / rel).read_text())
        data["openbao"]["auth"]["bound_claims"]["service_account_namespaces"] = [
            "other"
        ]
        (bundle / rel).write_text(yaml.safe_dump(data))
    elif change == "catalog-pointer":
        rel = "catalog-proposal.yaml"
        data = yaml.safe_load((bundle / rel).read_text())
        data["entry"] += "\nUNKNOWN-POINTER"
        (bundle / rel).write_text(yaml.safe_dump(data))
    else:
        rel = "rollback.yaml"
        data = yaml.safe_load((bundle / rel).read_text())
        data["removes_exactly"][-1]["name"] = "platform/workloads/keycape/sibling"
        (bundle / rel).write_text(yaml.safe_dump(data))
    manifest_path = bundle / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["files"][rel] = hashlib.sha256((bundle / rel).read_bytes()).hexdigest()
    manifest_path.write_text(json.dumps(manifest))
    with pytest.raises(structure.StructureRefused):
        structure.prepare(CASE, plan, bundle, allow_unimplemented=True)


def test_missing_owner_receipt_or_generic_status_label_never_proves_materialization(
    prepared,
):
    request, _, _ = prepared
    receipt = observations(request)
    with pytest.raises(structure.StructureRefused):
        structure.verify(request, receipt[:-1], lambda *args: True)
    with pytest.raises(structure.StructureRefused):
        structure.verify(request, receipt, lambda *args: False)
    with pytest.raises(structure.StructureRefused):
        structure.verify(
            request,
            (replace(receipt[0], disposition="secret-minted"), *receipt[1:]),
            lambda *args: True,
        )


def test_partial_case_objects_or_exact_source_drift_refuse(prepared):
    request, _, _ = prepared
    receipt = observations(request)
    with pytest.raises(structure.StructureRefused):
        structure.verify(
            request,
            (
                replace(receipt[1], expected_digest="sha256:" + "0" * 64),
                *receipt[:1],
                *receipt[2:],
            ),
            lambda *args: True,
        )


def test_created_vs_preexisting_projection_excludes_shared_objects_from_undo(prepared):
    request, _, _ = prepared
    receipt = list(observations(request))
    native = [i for i, o in enumerate(request.objects) if o.proof == "native-structure"]
    receipt[native[0]] = replace(
        receipt[native[0]], disposition="preexisting-matched", change_receipt=None
    )
    authenticated = {json.dumps(asdict(o), sort_keys=True) for o in receipt}
    result = structure.verify(
        request,
        tuple(receipt),
        lambda req, obj, obs: json.dumps(asdict(obs), sort_keys=True) in authenticated,
    )
    assert request.objects[native[0]] in result.preexisting
    assert request.objects[native[0]] not in result.rollback_targets
    assert result.rollback_targets == tuple(reversed(result.created))
    assert (
        request.objects[0] not in result.rollback_targets
    )  # layout is not secret custody undo


def test_source_bundle_inventory_path_escape_and_unsafe_extra_payload_refuse(prepared):
    _, plan, bundle = prepared
    manifest_path = bundle / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["files"]["../outside"] = "0" * 64
    manifest_path.write_text(json.dumps(manifest))
    with pytest.raises(structure.StructureRefused):
        structure.prepare(CASE, plan, bundle, allow_unimplemented=True)


def test_owner_verifier_diagnostics_and_interruptions_are_withheld(prepared, capfd):
    request, _, _ = prepared

    def broken(*args):
        print("SYNTHETIC-SECRET")
        os.write(2, b"SYNTHETIC-SECRET")
        raise KeyboardInterrupt("SYNTHETIC-SECRET")

    with pytest.raises(structure.StructureRefused) as err:
        structure.verify(request, observations(request), broken)
    assert "SYNTHETIC-SECRET" not in str(err.value)
    assert "SYNTHETIC-SECRET" not in "".join(capfd.readouterr())


def test_full_run_guard_change_keeps_structural_request_bound_and_explicit(prepared):
    _, plan, bundle = prepared
    previous_digest = structure.sha(b"previous reviewed execution guard")
    manifest_path = bundle / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    original = manifest["security_case"]["render_digest"]
    manifest["security_case"]["render_digest"] = previous_digest
    manifest_path.write_text(json.dumps(manifest))
    plan.write_text(plan.read_text().replace(original, previous_digest))
    request = structure.prepare(CASE, plan, bundle, allow_unimplemented=True)
    assert request.run_source_changed is True
    assert request.reviewed_render_digest == previous_digest
    assert request.render_digest != previous_digest


def test_read_only_cli_uses_no_live_survey_or_target_write(
    prepared, monkeypatch, capsys
):
    from ops_mason import cli

    _, plan, bundle = prepared
    monkeypatch.setattr(
        case_plan, "_bao_json", lambda *args: pytest.fail("live survey must not run")
    )
    before = {
        path.relative_to(bundle): path.read_bytes()
        for path in bundle.rglob("*")
        if path.is_file()
    }
    result = cli.main(
        [
            "structure-request",
            "--case",
            str(CASE),
            "--plan",
            str(plan),
            "--bundle",
            str(bundle),
            "--allow-unimplemented",
        ]
    )
    assert result == 0
    emitted = json.loads(capsys.readouterr().out)
    assert emitted["source_only"] is True
    assert emitted["request"]["structural_digest"].startswith("sha256:")
    assert before == {
        path.relative_to(bundle): path.read_bytes()
        for path in bundle.rglob("*")
        if path.is_file()
    }


def test_unknown_proof_kind_cannot_turn_credential_status_into_structure(prepared):
    request, _, _ = prepared
    forged = replace(
        request,
        objects=(
            replace(request.objects[0], proof="secret-minted"),
            *request.objects[1:],
        ),
    )
    receipt = observations(forged)
    with pytest.raises(structure.StructureRefused):
        structure.verify(forged, receipt, lambda *args: True)


def test_status_only_hash_binding_requires_real_owner_verification_and_no_mint(
    prepared,
):
    request, _, _ = prepared
    before = CASE.read_bytes()
    after_doc = yaml.safe_load(before)
    after_doc["spec"]["status"] = "materialized"
    after = yaml.safe_dump(after_doc).encode()
    bound = structure.bind_status_transition(
        request, before, after, observations(request), lambda *args: True
    )
    assert bound.accepted_case_digest == request.case_digest
    assert bound.materialized_case_digest == structure.sha(after)
    assert bound.materialized_case_digest != bound.accepted_case_digest
    assert bound.structural_digest == request.structural_digest
    with pytest.raises(structure.StructureRefused):
        structure.bind_status_transition(
            request, before, after, observations(request), lambda *args: False
        )


def test_status_hash_bridge_never_carries_parameter_or_acceptance_changes(prepared):
    request, _, _ = prepared
    before = CASE.read_bytes()
    for change in ("parameters", "acceptance"):
        after = yaml.safe_load(before)
        after["spec"]["status"] = "materialized"
        if change == "parameters":
            after["spec"]["parameters"]["client_id"] = "sibling-client"
        else:
            after["spec"]["acceptance"]["record"] = "new-unapproved-record"
        with pytest.raises(structure.StructureRefused):
            structure.bind_status_transition(
                request,
                before,
                yaml.safe_dump(after).encode(),
                observations(request),
                lambda *args: True,
            )


def test_same_target_policy_and_role_preserve_distinct_rollback_identity(prepared):
    request, _, _ = prepared
    shared_target = "fixture-shared-name"
    objects = tuple(
        replace(obj, target=shared_target)
        if obj.kind in ("openbao_policy", "openbao_role")
        else obj
        for obj in request.objects
    )
    request = replace(request, objects=objects)
    receipt = list(observations(request))
    native = [i for i, obj in enumerate(objects) if obj.proof == "native-structure"]
    # One shared policy was already present; newly created policy and role must
    # remain distinct even when their target names collide.
    receipt[native[0]] = replace(
        receipt[native[0]], disposition="preexisting-matched", change_receipt=None
    )
    result = structure.verify(request, tuple(receipt), lambda *args: True)
    assert objects[native[0]] in result.preexisting
    assert objects[native[0]] not in result.rollback_targets
    rollback = result.rollback_targets
    assert {obj.kind for obj in rollback} == {"openbao_policy", "openbao_role"}
    assert all(obj.target == shared_target for obj in rollback)
    assert len({obj.id for obj in rollback}) == len(rollback)
    assert all(obj.evidence_owner == "railiance-platform" for obj in rollback)


@pytest.mark.parametrize(
    "field,value",
    [
        ("case", "../invalid"),
        ("case", []),
        ("type_ref", "oidc-browser-relying-party"),
        ("type_ref", "oidc-browser-relying-party@unversioned"),
        ("run_source_changed", False),
        ("run_source_changed", 1),
    ],
)
def test_constructed_request_closed_identity_and_change_flag(prepared, field, value):
    request, _, _ = prepared
    forged = replace(request, **{field: value})
    with pytest.raises(structure.StructureRefused):
        structure.verify(forged, observations(request), lambda *args: True)


def test_historical_bundle_wrong_consumer_target_is_not_grandfathered(tmp_path):
    bundle = tmp_path / "historical-bundle"
    shutil.copytree(ROOT / "builds/case-forgejo-netkingdom-oidc", bundle)
    plan = tmp_path / "plan.md"
    shutil.copy2(ROOT / "plans/case-forgejo-netkingdom-oidc.md", plan)
    assert yaml.safe_load((bundle / "credential-change-requests/CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml").read_text())["target"]["workload"] == "keycape"
    with pytest.raises(structure.StructureRefused, match="role or reader binding differs"):
        structure.prepare(CASE, plan, bundle, allow_unimplemented=True)