tegwick
f11b948e8c
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929
76 lines
3.3 KiB
Python
76 lines
3.3 KiB
Python
"""A verified decision must remain usable through audit custody and dispatch."""
|
|
import time
|
|
from dataclasses import replace
|
|
from datetime import timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from hub_core.runtime.tables import runtime_messages, runtime_outcome_outbox
|
|
from hub_core.security import boundary
|
|
from hub_core.security.boundary import Decision
|
|
from hub_core.security.policy import verify_decision
|
|
from test_access_boundary import HEADERS
|
|
from test_access_policy import case, sign
|
|
from test_outcome_audit import target, rows, message
|
|
|
|
|
|
@pytest.mark.parametrize('limit', ['decision', 'caller', 'freshness'])
|
|
def test_signed_policy_preserves_earliest_verified_deadline(limit):
|
|
request,envelope,now = case()
|
|
expected = now + timedelta(seconds=30)
|
|
if limit == 'decision':
|
|
expected = now + timedelta(seconds=1)
|
|
envelope['lifetime']['expires_at'] = expected.isoformat()
|
|
elif limit == 'caller':
|
|
expected = now + timedelta(seconds=2)
|
|
envelope['provenance']['caller']['not_after'] = expected.isoformat()
|
|
result = verify_decision(envelope,request=request,keys=sign(envelope),
|
|
caller='workload:hub',now=now)
|
|
assert result.valid_until == expected.timestamp()
|
|
|
|
|
|
@pytest.mark.parametrize('deadline', [None, True, '123', float('nan'), float('inf')])
|
|
def test_decision_requires_finite_deadline(deadline):
|
|
with pytest.raises(ValueError,match='deadline'):
|
|
Decision(True,'decision','policy',deadline)
|
|
|
|
|
|
@pytest.mark.parametrize('elapsed,status', [(0,202), (1,503), (2,503)])
|
|
def test_expiry_during_audit_prevents_business_commit(target,monkeypatch,elapsed,status):
|
|
client,store,owners,_ = target
|
|
now = time.time()
|
|
clock = SimpleNamespace(time=lambda: now,monotonic=time.monotonic)
|
|
monkeypatch.setattr(boundary,'time',clock)
|
|
async def facts(actor,resource):
|
|
return replace(owners.facts,checked_at=clock.time())
|
|
owners.resolve = facts
|
|
evaluate,append = owners.evaluate,owners.append
|
|
async def policy(request):
|
|
return replace(await evaluate(request),valid_until=now+1)
|
|
async def audit(record):
|
|
await append(record)
|
|
clock.time = lambda: now+elapsed
|
|
owners.evaluate,owners.append = policy,audit
|
|
response = client.post('/ports/messaging/messages',headers=HEADERS,json=message())
|
|
assert response.status_code == status
|
|
if status == 503:
|
|
assert response.json()['detail'] == 'policy_decision_expired'
|
|
assert not rows(store,runtime_messages)
|
|
assert not rows(store,runtime_outcome_outbox)
|
|
assert [record['outcome'] for record in owners.records] == ['authorized','refused']
|
|
assert owners.records[0]['correlation_id'] == owners.records[1]['correlation_id']
|
|
else:
|
|
assert len(rows(store,runtime_messages)) == len(rows(store,runtime_outcome_outbox)) == 1
|
|
|
|
|
|
def test_already_expired_adapter_decision_is_not_authorized(target):
|
|
client,store,owners,_ = target
|
|
async def policy(request):
|
|
return Decision(True,'expired','policy',1)
|
|
owners.evaluate = policy
|
|
response = client.post('/ports/messaging/messages',headers=HEADERS,json=message())
|
|
assert response.status_code == 503
|
|
assert [record['outcome'] for record in owners.records] == ['refused']
|
|
assert not rows(store,runtime_messages)
|
|
assert not rows(store,runtime_outcome_outbox)
|