tegwick
4aa0b75449
Schema v9 stores who revoked an approval and an optional bounded reason. The bounded-execution carrier example is proven in process. APPROVAL-WP-0005, 0006, and 0007 are blocked because they are not current work. Task statuses stay as they were, apart from T03 and T04, which are done in source. Assistant: grok Assistant-Session: 01a1b5fd-0c33-7ba1-b383-53937d525595
200 lines
8.1 KiB
Python
200 lines
8.1 KiB
Python
"""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()
|