tegwick
c3f1dfbc07
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a05e2e-805b-7042-a750-71f473bceea2
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
"""Canon T-06 — Approval Replay.
|
|
|
|
Reuse a previously valid approval artifact for a different target, parameter
|
|
set, or later time. Pass: parameter binding, expiry, or replay protection
|
|
rejects the request.
|
|
|
|
Consume-side replay follows GH-DEC-2026-003: same-digest retries are
|
|
idempotent; a different digest against a consumed object conflicts.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from approval_engine.binding import binding_digest
|
|
from approval_engine.errors import Conflict
|
|
from approval_engine.store import Engine
|
|
from tests.conftest import approve, binding, validity
|
|
|
|
|
|
def _consumer_accepts(claim, proposed_binding) -> bool:
|
|
if claim["issuer"] != "approval-engine":
|
|
return False
|
|
if not claim["valid_now"]:
|
|
return False
|
|
if claim["consumed"]:
|
|
return False
|
|
if claim["reason_code"] != "ok":
|
|
return False
|
|
return claim["binding"]["digest"] == binding_digest(proposed_binding)
|
|
|
|
|
|
def test_t06_wrong_target_rejected(engine):
|
|
obj = approve(engine)
|
|
claim = engine.claim(obj.id)
|
|
assert _consumer_accepts(claim, binding()) is True
|
|
other = binding(target={"id": "some-other-lane", "stage": "prod"})
|
|
assert _consumer_accepts(claim, other) is False
|
|
|
|
|
|
def test_t06_wrong_action_rejected(engine):
|
|
obj = approve(engine)
|
|
claim = engine.claim(obj.id)
|
|
other = binding(action="secrets.kv.read")
|
|
assert _consumer_accepts(claim, other) is False
|
|
|
|
|
|
def test_t06_later_time_expired():
|
|
jumping = {"t": datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)}
|
|
eng = Engine(":memory:", clock=lambda: jumping["t"])
|
|
obj = approve(eng)
|
|
claim = eng.claim(obj.id)
|
|
assert _consumer_accepts(claim, binding()) is True
|
|
jumping["t"] = datetime(2026, 8, 29, 16, 0, tzinfo=timezone.utc)
|
|
claim = eng.claim(obj.id)
|
|
assert claim["state"] == "expired"
|
|
assert _consumer_accepts(claim, binding()) is False
|
|
eng.close()
|
|
|
|
|
|
def test_t06_revoked_rejected(engine):
|
|
obj = approve(engine)
|
|
engine.revoke(obj.id)
|
|
assert _consumer_accepts(engine.claim(obj.id), binding()) is False
|
|
|
|
|
|
def test_t06_superseded_rejected(engine):
|
|
obj = approve(engine)
|
|
engine.supersede(obj.id)
|
|
assert _consumer_accepts(engine.claim(obj.id), binding()) is False
|
|
|
|
|
|
def test_t06_consumed_approval_rejects_different_request_digest(engine):
|
|
obj = approve(engine)
|
|
first = "sha256:" + "56" * 32
|
|
engine.consume(obj.id, first)
|
|
try:
|
|
engine.consume(obj.id, "sha256:" + "78" * 32)
|
|
raise AssertionError("consume-side replay must conflict")
|
|
except Conflict:
|
|
pass
|
|
|
|
|
|
def test_t06_consumed_approval_allows_same_request_retry(engine):
|
|
obj = approve(engine)
|
|
digest = "sha256:" + "9a" * 32
|
|
engine.consume(obj.id, digest)
|
|
retry = engine.consume(obj.id, digest)
|
|
assert retry["idempotent"] is True
|