"""APPROVAL-WP-0007-T04: synthetic bounded-execution carrier path. The request is the published example. Nothing here contacts a live service. """ import json from datetime import datetime, timezone from pathlib import Path from approval_engine.api import App, call from approval_engine.auth import Identity, StaticTokenAuthenticator from approval_engine.store import Engine from tests.conftest import FROZEN EXAMPLE = json.loads( (Path(__file__).parents[1] / "examples" / "bounded-execution-approval-request.json").read_text() ) ADMISSION_DIGEST = EXAMPLE["binding"]["target"]["admission_digest"] OTHER_DIGEST = "sha256:" + "ab" * 32 APPROVER = "uid=admin,ou=people,dc=netkingdom,dc=local" FOUNDER = "uid=founder,ou=people,dc=netkingdom,dc=local" def _identity(subject, principal_type, level): return Identity( subject=subject, issuer="https://keycape.example", audiences=("approval-engine",), principal_type=principal_type, tenant="tenant:platform", roles=frozenset(), scopes=frozenset({ "approval:create", "approval:read", "approval:approve", "approval:revoke", "approval:consume", }), assurance={"level": level, "amr": ["pwd", "otp"], "source": "key-cape"}, evidence_ref=f"test-identity:{subject}", ) def _app(engine): return App(engine, StaticTokenAuthenticator({ "executor": _identity(EXAMPLE["binding"]["actor"], "service", "aal1"), "aal1": _identity(APPROVER, "human", "aal1"), "human": _identity(APPROVER, "human", "aal2"), "founder": _identity(FOUNDER, "human", "aal2"), "other": _identity("service:someone-else", "service", "aal1"), })) def test_synthetic_bounded_execution_carrier(): clock = [FROZEN] engine = Engine(":memory:", clock=lambda: clock[0]) app = _app(engine) try: request = json.loads(json.dumps(EXAMPLE)) assert request["binding"]["action"] == "bounded-execution" assert request["human_control"] is True and request["pdp_path"] is False assert set(request["binding"]["target"]) == {"admission_id", "admission_digest"} status, created = call( app, "POST", "/v1/approvals", request, authorization="Bearer executor", ) assert status == 201 aid = created["id"] assert created["status"] == "requested" assert created["binding"]["actor"] == request["binding"]["actor"] assert created["binding"]["target"] == request["binding"]["target"] status, refused = call( app, "POST", f"/v1/approvals/{aid}/entries", {}, authorization="Bearer aal1", ) assert status == 403 and refused["error"] == "forbidden" status, approved = call( app, "POST", f"/v1/approvals/{aid}/entries", {}, authorization="Bearer human", ) assert status == 200 and approved["status"] == "approved" assert approved["entries"][0]["subject_id"] == APPROVER assert approved["entries"][0]["principal_type"] == "human" assert json.loads(approved["entries"][0]["assurance"])["level"] == "aal2" assert APPROVER != request["binding"]["principal"] status, fetched = call( app, "GET", f"/v1/approvals/{aid}", authorization="Bearer executor", ) assert status == 200 for field in ("action", "actor", "principal", "purpose", "target"): assert fetched["binding"][field] == request["binding"][field] assert fetched["binding"]["digest"] == created["binding"]["digest"] assert fetched["binding"]["human_control"] is True assert fetched["binding"]["pdp_path"] is False assert fetched["binding"]["required_assurance"] == {"level": "aal2"} assert fetched["binding"]["actor_consumes"] is True assert fetched["validity"] == request["validity"] status, other = call( app, "POST", f"/v1/approvals/{aid}/consume", {"request_digest": ADMISSION_DIGEST, "decision_id": "run-1"}, authorization="Bearer other", ) assert status == 403 and other["error"] == "forbidden" status, spent = call( app, "POST", f"/v1/approvals/{aid}/consume", {"request_digest": ADMISSION_DIGEST, "decision_id": "run-1"}, authorization="Bearer executor", ) assert status == 200 assert spent["status"] == "consumed" assert spent["request_digest"] == ADMISSION_DIGEST assert spent["idempotent"] is False status, again = call( app, "POST", f"/v1/approvals/{aid}/consume", {"request_digest": ADMISSION_DIGEST, "decision_id": "run-1"}, authorization="Bearer executor", ) assert status == 200 and again["idempotent"] is True assert again["request_digest"] == ADMISSION_DIGEST assert again["consumed_at"] == spent["consumed_at"] status, conflict = call( app, "POST", f"/v1/approvals/{aid}/consume", {"request_digest": OTHER_DIGEST, "decision_id": "run-2"}, authorization="Bearer executor", ) assert status == 409 and conflict["error"] == "conflict" # Revoke before consume, by someone who is not the bound principal. status, second = call( app, "POST", "/v1/approvals", request, authorization="Bearer executor", ) assert status == 201 second_id = second["id"] assert call( app, "POST", f"/v1/approvals/{second_id}/entries", {}, authorization="Bearer human", )[0] == 200 status, revoked = call( app, "POST", f"/v1/approvals/{second_id}/revoke", {"reason": "founder stopped the run before consume"}, authorization="Bearer founder", ) assert status == 200 and revoked["status"] == "revoked" assert revoked["revoked_by"]["subject"] == FOUNDER assert revoked["revoked_by"]["subject"] != request["binding"]["principal"] status, blocked = call( app, "POST", f"/v1/approvals/{second_id}/consume", {"request_digest": ADMISSION_DIGEST, "decision_id": "run-revoked"}, authorization="Bearer executor", ) assert status == 409 and blocked["error"] == "conflict" # Consumption outside the window is refused. Issue and approve while # the clock is inside, then move it onto the exclusive expiry. short = json.loads(json.dumps(request)) short["validity"] = { "not_before": "2026-08-29T11:00:00+00:00", "expires_at": "2026-08-29T12:30:00+00:00", } status, third = call( app, "POST", "/v1/approvals", short, authorization="Bearer executor", ) assert status == 201 third_id = third["id"] assert call( app, "POST", f"/v1/approvals/{third_id}/entries", {}, authorization="Bearer human", )[0] == 200 clock[0] = datetime(2026, 8, 29, 12, 30, tzinfo=timezone.utc) status, late = call( app, "POST", f"/v1/approvals/{third_id}/consume", {"request_digest": ADMISSION_DIGEST, "decision_id": "run-late"}, authorization="Bearer executor", ) assert status == 409 and "window" in late["message"] early = json.loads(json.dumps(request)) early["validity"] = { "not_before": "2026-08-29T13:00:00+00:00", "expires_at": "2026-08-29T15:00:00+00:00", } clock[0] = FROZEN status, fourth = call( app, "POST", "/v1/approvals", early, authorization="Bearer executor", ) assert status == 201 fourth_id = fourth["id"] assert call( app, "POST", f"/v1/approvals/{fourth_id}/entries", {}, authorization="Bearer human", )[0] == 200 status, soon = call( app, "POST", f"/v1/approvals/{fourth_id}/consume", {"request_digest": ADMISSION_DIGEST, "decision_id": "run-early"}, authorization="Bearer executor", ) assert status == 409 and "window" in soon["message"] finally: engine.close()