fix(cases): validate browser metadata against pinned owner source
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
This commit is contained in:
parent
d2cd3fa940
commit
958352089b
5 changed files with 356 additions and 9 deletions
185
tests/test_case_inputs.py
Normal file
185
tests/test_case_inputs.py
Normal file
|
|
@ -0,0 +1,185 @@
|
|||
"""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
|
||||
Loading…
Add table
Add a link
Reference in a new issue