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])}
    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