File size: 5,078 Bytes
ecf9cb9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """
tests/test_audit.py
WORM chain integrity, replay store, semantic hash as chain anchor.
"""
import json
import os
import tempfile
import pytest
from datetime import datetime, timedelta, timezone
from audit.worm import WORMAuditChain, AuditRecord, attributes_hash
from audit.replay import ReplayStore, ReplayDetectedError, FileBackend
def _record(event='ASSERTION_ACCEPTED', seq=0) -> AuditRecord:
return AuditRecord(
timestamp='2026-08-07T00:00:00+00:00',
semantic_hash='a' * 64,
assertion_id=f'_id_{seq}',
issuer='https://vault-live.local/idp',
subject='alice',
gate_result=True,
entropy=0.0,
attributes_hash='b' * 64,
event=event,
)
@pytest.fixture
def tmp_worm():
with tempfile.NamedTemporaryFile(suffix='.jsonl', delete=False) as f:
path = f.name
chain = WORMAuditChain(path)
yield chain, path
os.unlink(path)
# ── WORM chain tests ──────────────────────────────────────────────────────────
def test_chain_grows_sequentially(tmp_worm):
chain, path = tmp_worm
for i in range(5):
chain.append(_record(seq=i))
result = WORMAuditChain.verify_chain(path)
assert result.valid is True
assert result.total_entries == 5
assert result.broken_at is None
def test_chain_verify_empty_file():
with tempfile.NamedTemporaryFile(suffix='.jsonl', delete=False) as f:
path = f.name
result = WORMAuditChain.verify_chain(path)
assert result.valid is True
assert result.total_entries == 0
os.unlink(path)
def test_chain_tamper_detected(tmp_worm):
chain, path = tmp_worm
for i in range(5):
chain.append(_record(seq=i))
# Tamper the third record
with open(path, 'r') as f:
lines = f.readlines()
entry = json.loads(lines[2])
entry['record']['event'] = 'TAMPERED'
lines[2] = json.dumps(entry) + '\n'
with open(path, 'w') as f:
f.writelines(lines)
result = WORMAuditChain.verify_chain(path)
assert result.valid is False
assert result.broken_at == 2
def test_semantic_hash_is_chain_anchor(tmp_worm):
"""Two records with the same semantic_hash get distinct chain hashes."""
chain, path = tmp_worm
r1 = _record(seq=0)
r2 = _record(seq=1)
r2.semantic_hash = r1.semantic_hash # same semantic hash
e1 = chain.append(r1)
e2 = chain.append(r2)
assert e1.entry_hash != e2.entry_hash # chain hashes differ
assert e1.record.semantic_hash == e2.record.semantic_hash
def test_attributes_hash_not_plaintext():
"""attributes_hash output must not contain any raw value."""
attrs = {'Role': ['VaultUser'], 'Email': ['alice@example.com']}
h = attributes_hash(attrs)
assert 'VaultUser' not in h
assert 'alice@example.com' not in h
assert len(h) == 64
# ── Replay store tests ────────────────────────────────────────────────────────
def _future(seconds: int = 300) -> str:
return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat()
def _past(seconds: int = 10) -> str:
return (datetime.now(timezone.utc) - timedelta(seconds=seconds)).isoformat()
def test_replay_store_blocks_duplicate():
store = ReplayStore(ttl_seconds=300)
store.check_and_store('_id1', _future())
with pytest.raises(ReplayDetectedError) as exc_info:
store.check_and_store('_id1', _future())
assert exc_info.value.assertion_id == '_id1'
def test_replay_store_allows_different_ids():
store = ReplayStore()
store.check_and_store('_id_a', _future())
store.check_and_store('_id_b', _future()) # different ID — should not raise
def test_replay_store_allows_after_expiry():
store = ReplayStore(ttl_seconds=300)
store.check_and_store('_id_exp', _past(10)) # already expired
# Should be treated as new since it's expired
store.check_and_store('_id_exp', _future()) # re-admissible
def test_purge_expired_clears_stale():
store = ReplayStore(ttl_seconds=300)
for i in range(10):
store._backend.put(f'_stale_{i}', datetime.now(timezone.utc) - timedelta(seconds=1), 'seen')
purged = store._purge_expired()
assert purged == 10
assert all(not store.is_seen(f'_stale_{i}') for i in range(10))
def test_file_backend_survives_reload():
with tempfile.NamedTemporaryFile(suffix='.json', delete=False) as f:
path = f.name
try:
store1 = ReplayStore(backend=FileBackend(path))
store1.check_and_store('_persist_id', _future())
store2 = ReplayStore(backend=FileBackend(path))
with pytest.raises(ReplayDetectedError):
store2.check_and_store('_persist_id', _future())
finally:
os.unlink(path)
|