tegwick
b2a4b863a8
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
171 lines
7.4 KiB
Python
171 lines
7.4 KiB
Python
from dataclasses import replace
|
|
import json
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from informed_decision.lifecycle_service import Caller, LifecycleError, LifecycleService
|
|
from informed_decision.policy import Observation
|
|
from informed_decision.provenance import Claim, Route
|
|
from informed_decision.store import EvidenceUnavailable
|
|
from test_lifecycle import AGENT, workspace
|
|
from test_policy_client import human, policy, PolicyTransport
|
|
|
|
|
|
def caller():
|
|
return Caller(AGENT, time.time() + 600, '{"principal_type_source":"authentication-derived"}')
|
|
|
|
|
|
def service(workspace, transport=None, **kwargs):
|
|
repo, memo = workspace
|
|
return LifecycleService(repo, policy(transport), **kwargs), memo
|
|
|
|
|
|
def test_agent_and_human_use_shared_service_and_replay_requires_new_decision(workspace):
|
|
app, memo = service(workspace)
|
|
agent = caller()
|
|
event = app.register(agent, memo, key="new")
|
|
assert app.register(agent, memo, key="new") == event
|
|
assert len(app.policy.transport.calls) == 2
|
|
person = Caller.from_human_session(replace(human(), tenant=Claim(AGENT.tenant, Route.DIRECTORY)))
|
|
app.annotate(person, memo.id, 1, kind="comment", body={"text": "Consider alternative B"}, key="comment")
|
|
app.revise(agent, memo.next_version(brief="Alternative B assessed"), expected_version=1,
|
|
summary="Addressed review", key="revise")
|
|
assert app.retrieve(person, memo.id).version == 2
|
|
events = app.events(agent, memo.id)
|
|
assert len(events) == 3
|
|
assert events[1]["actor"]["kind"] == "person"
|
|
observations = app.repository.store.policy_observations()
|
|
assert all(row["decision_attributable"] == 0 for row in observations)
|
|
identifiers = {row["id"] for row in observations}
|
|
assert all(event["body"]["policy_observation_id"] in identifiers for event in events)
|
|
request = app.policy.transport.calls[0][3]
|
|
assert request["action"] == "lifecycle.register"
|
|
assert request["subject"]["type"] == "agent"
|
|
assert request["resource"]["tenant"] == AGENT.tenant
|
|
assert "view_hash" not in json.dumps(request)
|
|
assert "synthetic-browser-token" not in json.dumps(observations)
|
|
|
|
|
|
@pytest.mark.parametrize("effect", ["deny", "redact", "audit_only", "not_applicable"])
|
|
def test_no_nonallow_effect_can_register(workspace, effect):
|
|
transport = PolicyTransport()
|
|
transport.change = lambda data: data.update(effect=effect)
|
|
app, memo = service(workspace, transport)
|
|
with pytest.raises(LifecycleError):
|
|
app.register(caller(), memo, key="new")
|
|
with pytest.raises(EvidenceUnavailable):
|
|
app.repository.retrieve(AGENT.tenant, memo.id)
|
|
assert len(app.repository.store.policy_observations()) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("mutate", [
|
|
lambda data: data["binding"].update(action="read"),
|
|
lambda data: data["binding"]["context"].update(command_digest="sha256:" + "0" * 64),
|
|
lambda data: data["provenance"].update(policy_package="unadmitted"),
|
|
lambda data: data.update(obligations=[{"type": "unknown"}]),
|
|
])
|
|
def test_policy_response_must_match_exact_lifecycle_request(workspace, mutate):
|
|
transport = PolicyTransport()
|
|
transport.change = mutate
|
|
app, memo = service(workspace, transport)
|
|
with pytest.raises(LifecycleError):
|
|
app.register(caller(), memo, key="new")
|
|
with pytest.raises(EvidenceUnavailable):
|
|
app.repository.store.memo(memo.id)
|
|
|
|
|
|
def test_revoked_permission_denies_replay_and_existing_and_missing_reads_equally(workspace):
|
|
app, memo = service(workspace)
|
|
agent = caller()
|
|
app.register(agent, memo, key="new")
|
|
app.policy.transport.change = lambda data: data.update(effect="deny")
|
|
operations = [lambda: app.register(agent, memo, key="new"),
|
|
lambda: app.retrieve(agent, memo.id), lambda: app.retrieve(agent, "missing"),
|
|
lambda: app.events(agent, memo.id),
|
|
lambda: app.annotate(agent, memo.id, 1, kind="comment", body={"text": "No"}, key="no")]
|
|
for operation in operations:
|
|
with pytest.raises(LifecycleError) as error:
|
|
operation()
|
|
assert error.value.status == 403
|
|
assert len(app.repository.events(AGENT.tenant, memo.id)) == 1
|
|
|
|
|
|
def test_identity_expiry_refuses_before_policy_call(workspace):
|
|
app, memo = service(workspace)
|
|
with pytest.raises(LifecycleError) as error:
|
|
app.register(replace(caller(), expires_at=1), memo, key="new")
|
|
assert error.value.status == 401
|
|
assert app.policy.transport.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("expiry", ["identity", "policy"])
|
|
def test_expiry_while_waiting_for_write_transaction_refuses_mutation(workspace, monkeypatch, expiry):
|
|
now = [time.time()]
|
|
app, memo = service(workspace, clock=lambda: now[0])
|
|
agent = replace(caller(), expires_at=now[0] + (10 if expiry == "identity" else 600))
|
|
original = app.repository.register
|
|
|
|
def delayed(*args, **kwargs):
|
|
now[0] += 40
|
|
return original(*args, **kwargs)
|
|
monkeypatch.setattr(app.repository, "register", delayed)
|
|
with pytest.raises(LifecycleError) as error:
|
|
app.register(agent, memo, key="new")
|
|
assert error.value.code == ("identity_expired" if expiry == "identity" else "decision_expired")
|
|
with pytest.raises(EvidenceUnavailable):
|
|
app.repository.store.memo(memo.id)
|
|
|
|
|
|
def test_command_is_snapshotted_before_external_policy_call(workspace):
|
|
app, memo = service(workspace)
|
|
agent = caller()
|
|
app.register(agent, memo, key="new")
|
|
body = {"text": "authorized content"}
|
|
app.policy.transport.change = lambda data: body.update(text="changed during network call")
|
|
app.annotate(agent, memo.id, 1, kind="comment", body=body, key="comment")
|
|
event = app.repository.events(AGENT.tenant, memo.id)[1]
|
|
assert event["body"]["text"] == "authorized content"
|
|
|
|
|
|
def test_read_checks_expiry_after_loading_content(workspace, monkeypatch):
|
|
now = [time.time()]
|
|
app, memo = service(workspace, clock=lambda: now[0])
|
|
agent = caller()
|
|
app.register(agent, memo, key="new")
|
|
original = app.repository.retrieve
|
|
def delayed(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
now[0] += 40
|
|
return result
|
|
monkeypatch.setattr(app.repository, "retrieve", delayed)
|
|
with pytest.raises(LifecycleError, match="decision_expired"):
|
|
app.retrieve(agent, memo.id)
|
|
|
|
|
|
def test_observation_from_another_request_is_refused(workspace, monkeypatch):
|
|
app, memo = service(workspace)
|
|
monkeypatch.setattr(app.policy, "check", lambda req: Observation('{}', None, "allow", time.time()+60, None))
|
|
with pytest.raises(LifecycleError, match="policy_request_mismatch"):
|
|
app.register(caller(), memo, key="new")
|
|
assert app.repository.store.policy_observations() == []
|
|
|
|
|
|
def test_missing_persisted_authorization_prevents_write(workspace, monkeypatch):
|
|
app, memo = service(workspace)
|
|
monkeypatch.setattr(app.repository.store, "observe_policy", lambda observed: "not-persisted")
|
|
with pytest.raises(LifecycleError, match="policy_observation_unavailable"):
|
|
app.register(caller(), memo, key="new")
|
|
with pytest.raises(EvidenceUnavailable):
|
|
app.repository.store.memo(memo.id)
|
|
|
|
|
|
@pytest.mark.parametrize("expiry", [float("nan"), float("inf"), True])
|
|
def test_unbounded_identity_lifetime_refused(expiry):
|
|
with pytest.raises(ValueError):
|
|
replace(caller(), expires_at=expiry)
|
|
|
|
|
|
def test_human_identity_must_be_authentication_derived():
|
|
with pytest.raises(ValueError):
|
|
Caller.from_human_session(replace(human(), principal_type=Claim("human", Route.REGISTRATION)))
|