tegwick
477c683991
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
110 lines
4.5 KiB
Python
110 lines
4.5 KiB
Python
"""Source evidence binds authenticated consumer, approved act and PDP digest."""
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from approval_engine.api import App, call
|
|
from approval_engine.audit import audit_envelope
|
|
from approval_engine.store import Engine
|
|
from tests.conftest import FROZEN, binding, validity
|
|
from tests.test_auth import _jwt
|
|
from tests.test_tenant_provenance import authenticator
|
|
|
|
FULL = "sha256:" + "ab" * 32
|
|
FREE = "sha256:" + "cd" * 32
|
|
|
|
|
|
def approved(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")
|
|
return obj
|
|
|
|
|
|
def token(private, **overrides):
|
|
claims = dict(sub="service:mediator", tenant="tenant:platform",
|
|
principal_type="service", tenant_source="registration",
|
|
scope="approval:consume")
|
|
claims.update(overrides)
|
|
return _jwt(private, **claims)
|
|
|
|
|
|
def consume(app, obj, bearer, **body):
|
|
return call(app, "POST", f"/v1/approvals/{obj.id}/consume",
|
|
{"request_digest": FULL, "decision_id": "decision:original", **body},
|
|
authorization="Bearer " + bearer)
|
|
|
|
|
|
def uses(engine):
|
|
return [r for r in engine.undrained() if r["class"] == "use"]
|
|
|
|
|
|
def test_authenticated_consumer_and_exact_binding_survive_restart_and_drain(tmp_path):
|
|
private, auth = authenticator()
|
|
path = tmp_path / "approval.sqlite"
|
|
engine = Engine(path, clock=lambda: FROZEN)
|
|
obj = approved(engine)
|
|
bearer = token(private)
|
|
status, result = consume(App(engine, auth), obj, bearer,
|
|
consumer={"subject_id": "spoofed"}, pdp_digest="spoofed")
|
|
assert status == 200 and result["idempotent"] is False
|
|
event = uses(engine)[0]
|
|
details = event["payload"]["details"]
|
|
assert event["payload"]["actor"] == binding()["actor"]
|
|
assert details["binding"] == binding()
|
|
assert details["binding_digest"] == obj.binding_digest
|
|
assert details["pdp_digest"] == FREE and details["pdp_path"] is True
|
|
assert details["request_digest"] == FULL
|
|
assert details["decision_id"] == "decision:original"
|
|
assert details["consumer"] == {
|
|
"subject_id": "service:mediator", "issuer": "https://keycape.example",
|
|
"principal_type": "service", "tenant": "tenant:platform",
|
|
"tenant_source": "registration", "evidence_ref": auth.authenticate("Bearer " + bearer).evidence_ref,
|
|
}
|
|
assert bearer not in json.dumps(event)
|
|
assert "consumer" not in json.dumps(engine.claim(obj.id))
|
|
engine.close()
|
|
engine = Engine(path, clock=lambda: FROZEN, auto_migrate=False)
|
|
assert uses(engine) == [event]
|
|
# A second authenticated caller cannot rewrite the first consumer evidence.
|
|
status, retry = consume(App(engine, auth), obj, token(private, sub="service:retry"),
|
|
decision_id="decision:other")
|
|
assert status == 200 and retry["idempotent"] is True
|
|
assert retry["decision_id"] == "decision:original"
|
|
assert uses(engine) == [event]
|
|
delivered = []
|
|
engine.drain(lambda payload: delivered.append(audit_envelope(payload)))
|
|
use = next(e for e in delivered if e["type"] == "approval.use")
|
|
assert use["id"] == event["event_id"]
|
|
assert use["data"]["details"] == details
|
|
engine.close()
|
|
|
|
|
|
@pytest.mark.parametrize("overrides,expected", [
|
|
({"tenant": "tenant:foreign"}, 403),
|
|
({"principal_type": "human"}, 403),
|
|
({"scope": "approval:read"}, 403),
|
|
({"tenant_source": "spoofed"}, 401),
|
|
])
|
|
def test_unadmitted_consumer_cannot_create_evidence(engine, overrides, expected):
|
|
private, auth = authenticator()
|
|
obj = approved(engine)
|
|
assert consume(App(engine, auth), obj, token(private, **overrides))[0] == expected
|
|
assert uses(engine) == [] and engine.get(obj.id).status == "approved"
|
|
|
|
|
|
def test_outbox_failure_rolls_back_consumer_and_spend(engine):
|
|
private, auth = authenticator()
|
|
obj = approved(engine)
|
|
engine.fail_outbox = True
|
|
assert consume(App(engine, auth), obj, token(private))[0] == 503
|
|
assert engine.get(obj.id).status == "approved" and uses(engine) == []
|
|
engine.fail_outbox = False
|
|
assert consume(App(engine, auth), obj, token(private, sub="service:recovered"))[0] == 200
|
|
assert uses(engine)[0]["payload"]["details"]["consumer"]["subject_id"] == "service:recovered"
|
|
|
|
|
|
def test_internal_calls_do_not_invent_authenticated_consumer(engine):
|
|
obj = approved(engine)
|
|
engine.consume(obj.id, FULL)
|
|
assert uses(engine)[0]["payload"]["details"]["consumer"] is None
|