tegwick
eece796b8b
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
176 lines
7.7 KiB
Python
176 lines
7.7 KiB
Python
import json
|
|
from datetime import timedelta
|
|
from urllib.parse import urlsplit
|
|
from urllib.error import HTTPError
|
|
|
|
import pytest
|
|
|
|
from approval_engine.api import call
|
|
from approval_engine.errors import Conflict
|
|
from approval_engine.execution_confirmation import ExecutionConfirmationClient
|
|
from approval_engine.pep import ApprovalProtocolError
|
|
from tests.conftest import FROZEN, binding, validity
|
|
from tests.test_pep import Response
|
|
|
|
FULL = "sha256:" + "ab" * 32
|
|
FREE = "sha256:" + "cd" * 32
|
|
|
|
|
|
def client(tmp_path, opener):
|
|
token = tmp_path / "test-token"
|
|
token.write_text("test-token")
|
|
token.chmod(0o600)
|
|
return ExecutionConfirmationClient("http://approval-engine", token, opener=opener)
|
|
|
|
|
|
def consumed(engine):
|
|
obj = engine.create(binding(), validity(), pdp_digest=FREE, pdp_path=True,
|
|
human_control=True)
|
|
engine.add_entry(obj.id, "human:founder", principal_type="human")
|
|
engine.consume(obj.id, FULL, decision_id="decision:original")
|
|
return obj
|
|
|
|
|
|
def confirmer(c, obj):
|
|
return c.confirm_consumption(obj.id, FULL, "decision:original",
|
|
expected_binding=binding(), expected_pdp_digest=FREE)
|
|
|
|
|
|
def test_confirmation_reads_authenticated_current_object_without_mutation(engine, app, tmp_path):
|
|
obj = consumed(engine)
|
|
before = list(engine.undrained())
|
|
requests = []
|
|
|
|
def opener(req, timeout):
|
|
requests.append(req)
|
|
status, result = call(app, req.get_method(), urlsplit(req.full_url).path,
|
|
authorization=req.get_header("Authorization"))
|
|
assert status == 200
|
|
return Response(result)
|
|
|
|
result = confirmer(client(tmp_path, opener), obj)
|
|
assert result == {"approval_id": obj.id, "request_digest": FULL,
|
|
"decision_id": "decision:original", "consumed_at": FROZEN.isoformat()}
|
|
assert requests[0].get_method() == "GET"
|
|
assert engine.undrained() == before
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [
|
|
("id", "different"), ("status", "approved"),
|
|
("consumption", {"request_digest": FULL, "decision_id": "different", "consumed_at": FROZEN.isoformat()}),
|
|
("consumption", {"request_digest": FREE, "decision_id": "decision:original", "consumed_at": FROZEN.isoformat()}),
|
|
("consumption", {"request_digest": FULL, "decision_id": "decision:original", "consumed_at": "naive"}),
|
|
("consumption", None), ("effect", "ALLOW"),
|
|
])
|
|
def test_wrong_consumption_object_refuses(engine, tmp_path, field, value):
|
|
obj = consumed(engine)
|
|
result = engine.get(obj.id).as_dict()
|
|
result[field] = value
|
|
with pytest.raises(ApprovalProtocolError):
|
|
confirmer(client(tmp_path, lambda *_a, **_k: Response(result)), obj)
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [
|
|
("action", "exec"), ("target", {"id": "sibling"}), ("actor", "forged"),
|
|
("principal", "forged"), ("purpose", "different"), ("pdp_digest", FULL),
|
|
("pdp_path", "true"), ("human_control", 1),
|
|
])
|
|
def test_wrong_act_or_digest_or_declaration_refuses(engine, tmp_path, field, value):
|
|
obj = consumed(engine)
|
|
result = engine.get(obj.id).as_dict()
|
|
result["binding"][field] = value
|
|
with pytest.raises(ApprovalProtocolError):
|
|
confirmer(client(tmp_path, lambda *_a, **_k: Response(result)), obj)
|
|
|
|
|
|
def test_retry_is_historical_after_expiry_and_retains_original_decision(engine, tmp_path):
|
|
obj = consumed(engine)
|
|
engine.clock = lambda: FROZEN + timedelta(days=1)
|
|
retry = engine.consume(obj.id, FULL, decision_id="decision:new")
|
|
assert retry["idempotent"] is True
|
|
assert retry["decision_id"] == "decision:original"
|
|
with pytest.raises(ApprovalProtocolError):
|
|
client(tmp_path, lambda *_a, **_k: Response(retry)).consume_exact(obj.id, FULL, "decision:new")
|
|
assert engine.claim(obj.id)["valid_now"] is False
|
|
assert confirmer(client(tmp_path, lambda *_a, **_k: Response(engine.get(obj.id).as_dict())), obj)
|
|
with pytest.raises(Conflict):
|
|
engine.revoke(obj.id)
|
|
with pytest.raises(Conflict):
|
|
engine.supersede(obj.id)
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [
|
|
("approval_id", "wrong"), ("decision_id", None), ("idempotent", "false"),
|
|
("consumed_at", "2026-10-11T12:00:00"), ("effect", "ALLOW"),
|
|
])
|
|
def test_exact_consume_response_refuses_loose_receipt(engine, tmp_path, field, value):
|
|
obj = consumed(engine)
|
|
result = engine.consume(obj.id, FULL, decision_id="decision:original")
|
|
result[field] = value
|
|
with pytest.raises(ApprovalProtocolError):
|
|
client(tmp_path, lambda *_a, **_k: Response(result)).consume_exact(obj.id, FULL, "decision:original")
|
|
|
|
|
|
def test_exact_consume_filters_extraneous_payload(engine, tmp_path):
|
|
obj = consumed(engine)
|
|
result = engine.consume(obj.id, FULL, decision_id="decision:original")
|
|
result["notes"] = "discard"
|
|
confirmed = client(tmp_path, lambda *_a, **_k: Response(result)).consume_exact(obj.id, FULL, "decision:original")
|
|
assert "notes" not in confirmed
|
|
|
|
|
|
def test_denied_read_never_represents_confirmation(engine, app, tmp_path):
|
|
obj = consumed(engine)
|
|
def opener(req, timeout):
|
|
status, result = call(app, "GET", urlsplit(req.full_url).path, authorization=None)
|
|
assert status == 401
|
|
raise HTTPError(req.full_url, status, "refused", {}, None)
|
|
with pytest.raises(ApprovalProtocolError, match="unauthorized"):
|
|
confirmer(client(tmp_path, opener), obj)
|
|
|
|
|
|
@pytest.mark.parametrize("original,substitute", [(True, 1), (False, 0), (1, 1.0), (1, True)])
|
|
@pytest.mark.parametrize("recompute_served_digest", [False, True])
|
|
def test_confirmation_refuses_python_equal_json_distinct_target(
|
|
engine, tmp_path, original, substitute, recompute_served_digest
|
|
):
|
|
expected = binding(target={"nested": [{"value": original}]})
|
|
obj = engine.create(expected, validity(), pdp_digest=FREE, pdp_path=True,
|
|
human_control=True)
|
|
engine.add_entry(obj.id, "human:founder", principal_type="human")
|
|
engine.consume(obj.id, FULL, decision_id="decision:original")
|
|
result = engine.get(obj.id).as_dict()
|
|
result["binding"]["target"] = {"nested": [{"value": substitute}]}
|
|
if recompute_served_digest:
|
|
from approval_engine.binding import binding_digest
|
|
result["binding"]["digest"] = binding_digest(result["binding"])
|
|
assert result["binding"]["target"] == expected["target"]
|
|
with pytest.raises(ApprovalProtocolError, match="binding does not match"):
|
|
client(tmp_path, lambda *_a, **_k: Response(result)).confirm_consumption(
|
|
obj.id, FULL, "decision:original", expected_binding=expected,
|
|
expected_pdp_digest=FREE)
|
|
|
|
|
|
@pytest.mark.parametrize("digest", [None, "sha256:" + "00" * 32, True])
|
|
def test_confirmation_requires_served_native_digest_correspondence(engine, tmp_path, digest):
|
|
obj = consumed(engine)
|
|
result = engine.get(obj.id).as_dict()
|
|
if digest is None:
|
|
result["binding"].pop("digest")
|
|
else:
|
|
result["binding"]["digest"] = digest
|
|
with pytest.raises(ApprovalProtocolError, match="binding does not match"):
|
|
confirmer(client(tmp_path, lambda *_a, **_k: Response(result)), obj)
|
|
|
|
|
|
def test_confirmation_accepts_nested_json_types_and_key_order(engine, tmp_path):
|
|
expected = binding(target={"b": [True, False, 1, 1.0], "a": {"unicode": "ä"}})
|
|
obj = engine.create(expected, validity(), pdp_digest=FREE, pdp_path=True,
|
|
human_control=True)
|
|
engine.add_entry(obj.id, "human:founder", principal_type="human")
|
|
engine.consume(obj.id, FULL, decision_id="decision:original")
|
|
result = engine.get(obj.id).as_dict()
|
|
result["binding"]["target"] = dict(reversed(list(expected["target"].items())))
|
|
assert client(tmp_path, lambda *_a, **_k: Response(result)).confirm_consumption(
|
|
obj.id, FULL, "decision:original", expected_binding=expected,
|
|
expected_pdp_digest=FREE)["approval_id"] == obj.id
|