ops-warden/tests/test_case_inputs.py

"""Actual producer validation and Executor; fake owner evidence, no native claims."""

import os
from pathlib import Path

import pytest
import yaml

from warden.case_inputs import BrowserInputs, SOURCE_PINS
from warden.case_run import digest
from warden.case_receipt_state import Outcome, Refused
from tests.test_case_run import rig as base_rig


@pytest.fixture
def rig(tmp_path):
    return base_rig.__wrapped__(tmp_path)


NK = Path(
    os.environ.get(
        "WARDEN_TEST_NETKINGDOM_ROOT",
        Path(__file__).resolve().parents[2] / "net-kingdom",
    )
)


def browser(rig, version="1.1", mutate=None):
    executor, _, calls, sources, kwargs = rig
    case = yaml.safe_load(
        (NK / "security-cases/forgejo-netkingdom-oidc.yaml").read_bytes()
    )
    relative = (
        "security-case-types"
        if version == "1.1"
        else "security-case-type-revisions/1.2"
    )
    typ = yaml.safe_load(
        (NK / relative / "oidc-browser-relying-party/type.yaml").read_bytes()
    )
    case["spec"]["type"] = "oidc-browser-relying-party@" + version
    case["spec"]["status"] = "materialized"
    if version == "1.2":
        case["spec"]["parameters"]["consumer_namespace"] = "forgejo"
    # This extension and all native provenance are TEST ONLY, never source assent.
    typ["spec"]["execution_descriptors"] = {"mint-secret": digest(sources[2])}
    for step in typ["spec"]["follow_through"]:
        if step["id"] == "mint-secret":  # TEST ONLY: route to the fixture lane
            step.update(owner="owner", lane="issuer-lane")
    if mutate:
        mutate(case, typ)
    changed = (
        yaml.safe_dump(case).encode(),
        yaml.safe_dump(typ).encode(),
        *sources[2:],
    )
    old = executor.evidence_verifier

    def verified(purpose, source, evidence):
        return old(
            purpose,
            source.replace(digest(changed[0]), digest(sources[0])).replace(
                digest(changed[1]), digest(sources[1])
            ),
            evidence,
        )

    executor.evidence_verifier = verified
    executor.browser_inputs = BrowserInputs(NK)
    return executor, calls, changed, kwargs


@pytest.mark.parametrize("version", ["1.1", "1.2"])
def test_actual_browser_executor_and_owner_projection(rig, version):
    executor, calls, sources, kwargs = browser(rig, version)
    args = executor.admit(*sources, **kwargs)
    assert args[1].target == "forgejo"
    assert not calls
    assert executor.run(*args).last.outcome is Outcome.SUCCESS
    assert calls[0].target == "forgejo"
    assert not hasattr(calls[0], "parameters")
    # Owner callbacks/journal get target and fingerprints, no full metadata.
    persisted = (executor.journal.directory / "journal.sqlite").read_bytes()
    assert b"https://forgejo.coulomb.social" not in persisted
    assert b"forge-admins" not in persisted


@pytest.mark.parametrize(
    "field,value",
    [
        ("redirect_uris", "https://*.example.org/callback"),
        ("redirect_uris", "http://example.org/callback"),
        ("allowed_scopes", "openid,admin"),
        ("client_type", "arbitrary"),
        ("mfa_required", False),
        ("unknown", "value"),
        ("issuer", "https://other.example"),
        ("display_name", "hvs." + "A" * 30),
    ],
)
def test_case_refusals_even_with_test_evidence(rig, field, value):
    executor, calls, sources, kwargs = browser(
        rig, mutate=lambda c, t: c["spec"]["parameters"].update({field: value})
    )
    with pytest.raises(Refused, match="exact-version"):
        executor.admit(*sources, **kwargs)
    assert not calls


@pytest.mark.parametrize(
    "mutation",
    [
        lambda c, t: t["spec"].update(version="1.3"),
        lambda c, t: t["spec"]["parameters"][2].update(sensitivity="public"),
        lambda c, t: t["spec"]["parameters"][2].update(constraints={}),
        lambda c, t: t["spec"]["parameters"].append(
            {
                "name": "secret",
                "type": "string",
                "case_settable": True,
                "sensitivity": "secret_reference",
            }
        ),
    ],
)
def test_type_revision_cannot_widen_reviewed_parameters(rig, mutation):
    def change(case, typ):
        mutation(case, typ)
        case["spec"]["type"] = typ["spec"]["id"] + "@" + typ["spec"]["version"]

    executor, calls, sources, kwargs = browser(rig, mutate=change)
    with pytest.raises(Refused, match="exact-version"):
        executor.admit(*sources, **kwargs)
    assert not calls


def test_namespace_and_missing_required_are_validated(rig):
    executor, calls, sources, kwargs = browser(
        rig, "1.2", lambda c, t: c["spec"]["parameters"].update(consumer_namespace="*")
    )
    with pytest.raises(Refused, match="exact-version"):
        executor.admit(*sources, **kwargs)
    assert not calls


@pytest.mark.parametrize("changed_dependency", list(SOURCE_PINS))
def test_missing_producer_and_changed_transitive_source_refuse(
    rig, tmp_path, changed_dependency
):
    executor, calls, sources, kwargs = browser(rig)
    executor.browser_inputs = BrowserInputs(tmp_path)
    for relative in SOURCE_PINS:
        path = tmp_path / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        if relative != changed_dependency:
            path.write_bytes((NK / relative).read_bytes())
    with pytest.raises(Refused, match="exact-version"):
        executor.admit(*sources, **kwargs)
    (tmp_path / changed_dependency).write_text('raise RuntimeError("must not execute")')
    with pytest.raises(Refused, match="exact-version"):
        executor.admit(*sources, **kwargs)
    assert not calls


def test_browser_without_validator_still_refuses(rig):
    executor, calls, sources, kwargs = browser(rig)
    executor.browser_inputs = None
    with pytest.raises(Refused, match="not configured"):
        executor.admit(*sources, **kwargs)
    assert not calls


def test_real_unextended_type_stays_plan_only(rig):
    executor, calls, sources, kwargs = browser(rig)
    typ = yaml.safe_load(sources[1])
    del typ["spec"]["execution_descriptors"]
    with pytest.raises(Refused, match="plan only"):
        executor.admit(sources[0], yaml.safe_dump(typ).encode(), *sources[2:], **kwargs)
    assert not calls


def test_required_browser_parameter_is_not_defaulted_away(rig):
    executor, calls, sources, kwargs = browser(
        rig, mutate=lambda c, t: c["spec"]["parameters"].pop("redirect_uris")
    )
    with pytest.raises(Refused, match="exact-version"):
        executor.admit(*sources, **kwargs)
    assert not calls


@pytest.mark.parametrize("version", ["1.1", "1.2"])
@pytest.mark.parametrize(
    "pins",
    [
        None,
        [],
        {"mint-secret": "sha256:" + "A" * 64},
        {"mint-secret": "sha256:" + "a" * 64 + "\n"},
        {"undeclared-step": "sha256:" + "a" * 64},
        {"mint-secret": ["sha256:" + "a" * 64]},
        {"step-" + str(n): "sha256:" + "a" * 64 for n in range(129)},
    ],
)
def test_actual_producer_refuses_malformed_descriptor_pins(rig, version, pins):
    executor, calls, sources, _ = browser(rig, version)
    case, typ = (yaml.safe_load(source) for source in sources[:2])
    typ["spec"]["execution_descriptors"] = pins
    with pytest.raises(Refused, match="exact-version"):
        executor.browser_inputs.target(case, typ, "client_id")
    assert not calls