tests/test_integration.py
2.6 KB · 79 lines · python Raw
1 """End-to-end integration tests."""
2
3 from __future__ import annotations
4
5 import pytest
6
7 from pqc_kv_cache.audit import KVAuditLog
8 from pqc_kv_cache.errors import TenantIsolationError
9 from pqc_kv_cache.isolation import TenantIsolationManager
10 from pqc_kv_cache.rotation import KeyRotationPolicy, RotationTrigger
11
12
13 def test_multi_tenant_flow(
14 tenant_alice, tenant_bob, sample_entry_factory
15 ) -> None:
16 mgr = TenantIsolationManager()
17 audit = KVAuditLog()
18
19 s_alice = mgr.create_session(tenant_alice)
20 s_bob = mgr.create_session(tenant_bob)
21
22 alice_encs = []
23 bob_encs = []
24 for pos in range(5):
25 a_entry = sample_entry_factory(s_alice, 0, pos)
26 b_entry = sample_entry_factory(s_bob, 0, pos)
27 ae = mgr.encrypt(tenant_alice.tenant_id, a_entry)
28 be = mgr.encrypt(tenant_bob.tenant_id, b_entry)
29 audit.log_encrypt(
30 tenant_alice.tenant_id, s_alice.session_id, 0, pos, ae.sequence_number
31 )
32 audit.log_encrypt(
33 tenant_bob.tenant_id, s_bob.session_id, 0, pos, be.sequence_number
34 )
35 alice_encs.append(ae)
36 bob_encs.append(be)
37
38 # Alice MUST NOT be able to decrypt any of Bob's entries
39 for be in bob_encs:
40 with pytest.raises(TenantIsolationError):
41 mgr.decrypt(tenant_alice.tenant_id, be)
42 audit.log_isolation_violation(
43 tenant_alice.tenant_id,
44 tenant_bob.tenant_id,
45 details=f"seq={be.sequence_number}",
46 )
47
48 # Alice can decrypt her own
49 for ae in alice_encs:
50 dec = mgr.decrypt(tenant_alice.tenant_id, ae)
51 assert dec.metadata.tenant_id == tenant_alice.tenant_id
52
53 violations = audit.entries(operation="isolation-violation")
54 assert len(violations) == 5
55
56
57 def test_rotation_flow(tenant_alice, sample_entry_factory) -> None:
58 mgr = TenantIsolationManager()
59 policy = KeyRotationPolicy(max_entries=50, max_age_seconds=3600)
60 session = mgr.create_session(tenant_alice)
61 original_key = session.symmetric_key
62
63 rotated = False
64 for pos in range(100):
65 entry = sample_entry_factory(session, 0, pos)
66 mgr.encrypt(tenant_alice.tenant_id, entry)
67 should, trigger = policy.should_rotate(session)
68 if should and not rotated:
69 assert trigger is RotationTrigger.ENTRY_COUNT
70 assert pos + 1 >= 50
71 policy.rotate(session)
72 rotated = True
73 new_key = session.symmetric_key
74 assert new_key != original_key
75 assert session.entries_encrypted == 0
76 assert session.next_sequence == 1
77
78 assert rotated is True
79