64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import pytest
|
|
import aiosqlite
|
|
|
|
from app.security.audit import verify_chain, write_audit_event
|
|
from app.db import MIGRATIONS_DIR
|
|
|
|
|
|
async def _fresh_db() -> aiosqlite.Connection:
|
|
conn = await aiosqlite.connect(":memory:")
|
|
for migration_file in sorted(MIGRATIONS_DIR.glob("*.sql")):
|
|
await conn.executescript(migration_file.read_text(encoding="utf-8"))
|
|
return conn
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chain_intact_after_writes():
|
|
conn = await _fresh_db()
|
|
for i in range(5):
|
|
await write_audit_event(
|
|
conn, event_type="test_event", user_id=None, client_ip="127.0.0.1", details={"i": i}
|
|
)
|
|
await conn.commit()
|
|
|
|
intact, broken_at = await verify_chain(conn)
|
|
assert intact is True
|
|
assert broken_at is None
|
|
await conn.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tamper_detected_after_direct_update():
|
|
conn = await _fresh_db()
|
|
for i in range(3):
|
|
await write_audit_event(
|
|
conn, event_type="test_event", user_id=None, client_ip="127.0.0.1", details={"i": i}
|
|
)
|
|
await conn.commit()
|
|
|
|
# Der Append-only-Trigger blockt normale UPDATEs -- simuliert wird hier
|
|
# eine Umgehung auf DB-Ebene (z.B. Datei-Manipulation waehrend der App
|
|
# gestoppt ist), um zu zeigen, dass verify_chain() dies unabhaengig von
|
|
# den Triggern erkennt.
|
|
await conn.execute("DROP TRIGGER no_audit_update")
|
|
await conn.execute("UPDATE audit_log SET details_json = '{\"i\": 999}' WHERE id = 2")
|
|
await conn.commit()
|
|
|
|
intact, broken_at = await verify_chain(conn)
|
|
assert intact is False
|
|
assert broken_at == 2
|
|
await conn.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_append_only_trigger_blocks_update():
|
|
conn = await _fresh_db()
|
|
await write_audit_event(
|
|
conn, event_type="test_event", user_id=None, client_ip="127.0.0.1", details={}
|
|
)
|
|
await conn.commit()
|
|
|
|
with pytest.raises(aiosqlite.Error):
|
|
await conn.execute("UPDATE audit_log SET event_type = 'tampered' WHERE id = 1")
|
|
await conn.close()
|