102 lines
4.0 KiB
Python
102 lines
4.0 KiB
Python
"""Executable completeness contract for the append-only audit chain."""
|
|
from __future__ import annotations
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
import os
|
|
import sqlite3
|
|
import pytest
|
|
from aios_v2r1.audit import AuditLog
|
|
from aios_v2r1.tenant_context import ConflictError, ValidationError
|
|
|
|
def append(log: AuditLog, index: int = 0, *, tenant_id: str = "tenant-a") -> str:
|
|
return log.append(tenant_id=tenant_id, trace_id=f"trace-{index}",
|
|
event_type="tool_call",
|
|
payload={"index": index, "text": "虾马 café"})
|
|
|
|
def test_first_event_is_anchored_to_genesis(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
log = AuditLog(database)
|
|
append(log)
|
|
with sqlite3.connect(database) as db:
|
|
assert db.execute("SELECT previous_hash FROM audit_events").fetchone()[0] == "GENESIS"
|
|
assert log.verify()
|
|
|
|
def test_unicode_payload_is_canonical_across_restart(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
digest = append(AuditLog(database))
|
|
reopened = AuditLog(database)
|
|
assert reopened.verify()
|
|
with sqlite3.connect(database) as db:
|
|
stored = db.execute("SELECT event_hash,payload FROM audit_events").fetchone()
|
|
assert stored[0] == digest
|
|
assert "虾马" in stored[1]
|
|
|
|
def test_payload_tamper_is_detected(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
log = AuditLog(database)
|
|
append(log)
|
|
with sqlite3.connect(database) as db:
|
|
db.execute("UPDATE audit_events SET payload='{}'")
|
|
assert not log.verify()
|
|
|
|
def test_middle_event_deletion_is_detected(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
log = AuditLog(database)
|
|
for index in range(3):
|
|
append(log, index)
|
|
with sqlite3.connect(database) as db:
|
|
db.execute("DELETE FROM audit_events WHERE seq=2")
|
|
assert not log.verify()
|
|
|
|
def test_event_reordering_is_detected(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
log = AuditLog(database)
|
|
append(log, 1)
|
|
append(log, 2)
|
|
with sqlite3.connect(database) as db:
|
|
db.execute("UPDATE audit_events SET seq=100 WHERE seq=1")
|
|
assert not log.verify()
|
|
|
|
def test_concurrent_appends_form_one_valid_chain(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
log = AuditLog(database)
|
|
with ThreadPoolExecutor(max_workers=16) as executor:
|
|
digests = list(executor.map(lambda index: append(log, index), range(24)))
|
|
assert len(set(digests)) == 24
|
|
assert log.verify()
|
|
with sqlite3.connect(database) as db:
|
|
assert db.execute("SELECT count(*) FROM audit_events").fetchone()[0] == 24
|
|
|
|
@pytest.mark.parametrize("kwargs", [
|
|
{"tenant_id": "Tenant-A"}, {"trace_id": "../trace"},
|
|
{"event_type": "tool/call"}])
|
|
def test_audit_identifiers_are_validated(tmp_path: Path, kwargs: dict[str, str]) -> None:
|
|
values = {"tenant_id": "tenant-a", "trace_id": "trace-a",
|
|
"event_type": "tool_call", "payload": {"ok": True}}
|
|
values.update(kwargs)
|
|
with pytest.raises(ValidationError):
|
|
AuditLog(tmp_path / "audit.sqlite3").append(**values)
|
|
|
|
def test_tenants_share_no_hash_identity_for_equal_payloads(tmp_path: Path) -> None:
|
|
log = AuditLog(tmp_path / "audit.sqlite3")
|
|
first = append(log, tenant_id="tenant-a")
|
|
second = append(log, tenant_id="tenant-b")
|
|
assert first != second
|
|
assert log.verify()
|
|
|
|
def test_database_is_owner_only(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
AuditLog(database)
|
|
assert os.stat(database).st_mode & 0o777 == 0o600
|
|
|
|
def test_sqlite_failures_are_sanitized(tmp_path: Path) -> None:
|
|
database = tmp_path / "audit.sqlite3"
|
|
log = AuditLog(database)
|
|
database.write_bytes(b"not-a-sqlite-database")
|
|
with pytest.raises(ConflictError, match="audit storage unavailable") as raised:
|
|
log.verify()
|
|
assert not isinstance(raised.value, sqlite3.Error)
|
|
assert "sqlite" not in str(raised.value).lower()
|
|
assert raised.value.__cause__ is None
|
|
assert raised.value.__context__ is None
|