tegwick
d97652a3bb
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
207 lines
7 KiB
Python
207 lines
7 KiB
Python
"""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
|
|
|
|
|
|
@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
|