tegwick
c9ad5c2dd7
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1046e-fdbf-7b93-a267-5f72b9b24526
97 lines
4.5 KiB
Python
97 lines
4.5 KiB
Python
"""Verified tenant origin remains evidence, including absence and legacy rows."""
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
|
|
from approval_engine.api import App, call
|
|
from approval_engine.auth import JWTAuthenticator
|
|
from approval_engine.errors import StoreUnavailable
|
|
from approval_engine.store import Engine, LATEST_SCHEMA_VERSION
|
|
from tests.conftest import FROZEN, binding, validity
|
|
from tests.test_auth import _JWKS, _jwt
|
|
|
|
|
|
def authenticator():
|
|
private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
auth = JWTAuthenticator(
|
|
issuer="https://keycape.example", audience="approval-engine",
|
|
jwks_url="https://keycape.example/jwks", jwks_client=_JWKS(private.public_key()),
|
|
)
|
|
return private, auth
|
|
|
|
|
|
@pytest.mark.parametrize("source", ["directory", "registration", "default", None])
|
|
def test_verified_provenance_survives_http_entry_restart_and_use(tmp_path, source):
|
|
private, auth = authenticator()
|
|
claims = dict(sub="human", tenant="tenant:platform", principal_type="human",
|
|
scope="approval:approve")
|
|
if source is not None:
|
|
claims["tenant_source"] = source
|
|
token = _jwt(private, **claims)
|
|
path = tmp_path / "approval.sqlite"
|
|
engine = Engine(path, clock=lambda: FROZEN)
|
|
obj = engine.create(binding(), validity(), human_control=True)
|
|
digest = obj.binding_digest
|
|
status, approved = call(App(engine, auth), "POST", f"/v1/approvals/{obj.id}/entries",
|
|
{"tenant_source": "spoofed"}, authorization="Bearer " + token)
|
|
assert status == 200
|
|
assert approved["entries"][0]["tenant_source"] == source
|
|
engine.close()
|
|
engine = Engine(path, clock=lambda: FROZEN, auto_migrate=False)
|
|
assert engine.get(obj.id).entries[0]["tenant_source"] == source
|
|
assert engine.get(obj.id).binding_digest == digest
|
|
assert "tenant_source" not in json.dumps(engine.claim(obj.id))
|
|
engine.consume(obj.id, digest)
|
|
rows = engine._conn().execute("SELECT class, payload_json FROM outbox").fetchall()
|
|
for row in rows:
|
|
payload = json.loads(row["payload_json"])
|
|
assert payload["details"]["threshold"]["approvers"][0]["tenant_source"] == source
|
|
assert {row["class"] for row in rows} == {"issuance", "use"}
|
|
engine.close()
|
|
|
|
|
|
@pytest.mark.parametrize("source", [None, "unknown", "", {}, [], 1])
|
|
def test_explicit_invalid_provenance_refuses_http_bind_without_mutation(engine, source):
|
|
private, auth = authenticator()
|
|
token = _jwt(private, sub="human", tenant="tenant:platform", principal_type="human",
|
|
scope="approval:approve", tenant_source=source)
|
|
obj = engine.create(binding(), validity(), human_control=True)
|
|
status, _ = call(App(engine, auth), "POST", f"/v1/approvals/{obj.id}/entries", {},
|
|
authorization="Bearer " + token)
|
|
assert status == 401
|
|
assert engine.get(obj.id).entries == []
|
|
assert engine._conn().execute("SELECT COUNT(*) FROM outbox").fetchone()[0] == 0
|
|
|
|
|
|
def test_v5_migration_preserves_unknown_provenance_and_existing_evidence(tmp_path):
|
|
path = tmp_path / "approval.sqlite"
|
|
engine = Engine(path, clock=lambda: FROZEN)
|
|
obj = engine.create(binding(), validity(), human_control=True)
|
|
engine.add_entry(obj.id, "human", principal_type="human")
|
|
digest = obj.binding_digest
|
|
evidence = engine._conn().execute("SELECT payload_json FROM outbox").fetchall()[0][0]
|
|
engine.close()
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("ALTER TABLE entries DROP COLUMN tenant_source")
|
|
conn.execute("PRAGMA user_version=5")
|
|
with pytest.raises(StoreUnavailable, match="migrate"):
|
|
Engine(path, auto_migrate=False)
|
|
engine = Engine(path, clock=lambda: FROZEN)
|
|
assert engine.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 6
|
|
assert engine.get(obj.id).entries[0]["tenant_source"] is None
|
|
assert engine.get(obj.id).binding_digest == digest
|
|
assert engine.get(obj.id).human_control is True
|
|
assert engine._conn().execute("SELECT payload_json FROM outbox").fetchone()[0] == evidence
|
|
engine.close()
|
|
|
|
|
|
def test_provenance_rolls_back_with_failed_outbox(engine):
|
|
obj = engine.create(binding(), validity(), human_control=True)
|
|
engine.fail_outbox = True
|
|
with pytest.raises(StoreUnavailable):
|
|
engine.add_entry(obj.id, "human", principal_type="human", tenant_source="directory")
|
|
engine.fail_outbox = False
|
|
assert engine.get(obj.id).entries == []
|
|
assert engine.get(obj.id).status == "requested"
|