120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
"""Append-only, tamper-evident audit chain for AIOS V2R1."""
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from contextlib import closing
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from typing import Any, TypeVar
|
|
|
|
from .tenant_context import ConflictError, canonical_json, isoformat, require_ident
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class AuditLog:
|
|
def __init__(self, database: str | Path) -> None:
|
|
self.database = Path(database)
|
|
self.database.parent.mkdir(parents=True, exist_ok=True)
|
|
self._initialize()
|
|
os.chmod(self.database, 0o600)
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
db = sqlite3.connect(str(self.database), timeout=30)
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
db.execute("PRAGMA synchronous=FULL")
|
|
db.execute("PRAGMA busy_timeout=30000")
|
|
return db
|
|
|
|
def _storage(self, operation: Callable[[], T]) -> T:
|
|
failure: ConflictError | None = None
|
|
try:
|
|
return operation()
|
|
except sqlite3.Error:
|
|
failure = ConflictError("audit storage unavailable")
|
|
raise failure
|
|
|
|
def _initialize(self) -> None:
|
|
def initialize() -> None:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
db.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS audit_events (
|
|
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
tenant_id TEXT NOT NULL,
|
|
trace_id TEXT NOT NULL,
|
|
event_type TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
previous_hash TEXT NOT NULL,
|
|
event_hash TEXT NOT NULL UNIQUE,
|
|
created_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
self._storage(initialize)
|
|
|
|
def append(
|
|
self, *, tenant_id: str, trace_id: str, event_type: str,
|
|
payload: dict[str, Any],
|
|
) -> str:
|
|
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
|
|
trace_id = require_ident("trace_id", trace_id)
|
|
event_type = require_ident("event_type", event_type)
|
|
|
|
def write() -> str:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
previous = db.execute(
|
|
"SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
|
|
).fetchone()
|
|
previous_hash = previous[0] if previous else "GENESIS"
|
|
body = {
|
|
"tenant_id": tenant_id, "trace_id": trace_id,
|
|
"event_type": event_type, "payload": payload,
|
|
"previous_hash": previous_hash, "created_at": isoformat(),
|
|
}
|
|
digest = hashlib.sha256(canonical_json(body)).hexdigest()
|
|
db.execute(
|
|
"INSERT INTO audit_events VALUES(NULL,?,?,?,?,?,?,?)",
|
|
(
|
|
tenant_id, trace_id, event_type,
|
|
json.dumps(payload, ensure_ascii=False),
|
|
previous_hash, digest, body["created_at"],
|
|
),
|
|
)
|
|
db.commit()
|
|
return digest
|
|
|
|
return self._storage(write)
|
|
|
|
def verify(self) -> bool:
|
|
def check() -> bool:
|
|
previous = "GENESIS"
|
|
with closing(self._connect()) as connection, connection as db:
|
|
rows = db.execute(
|
|
"SELECT tenant_id,trace_id,event_type,payload,previous_hash,"
|
|
"event_hash,created_at FROM audit_events ORDER BY seq"
|
|
)
|
|
for row in rows:
|
|
try:
|
|
payload = json.loads(row[3])
|
|
except (json.JSONDecodeError, TypeError):
|
|
return False
|
|
body = {
|
|
"tenant_id": row[0], "trace_id": row[1],
|
|
"event_type": row[2], "payload": payload,
|
|
"previous_hash": row[4], "created_at": row[6],
|
|
}
|
|
if row[4] != previous:
|
|
return False
|
|
digest = hashlib.sha256(canonical_json(body)).hexdigest()
|
|
if digest != row[5]:
|
|
return False
|
|
previous = digest
|
|
return True
|
|
|
|
return self._storage(check)
|