90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
"""Append-only audit chain for AIOS V2R1."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from typing import Any
|
|
|
|
from .tenant_context import canonical_json, isoformat, require_ident
|
|
|
|
|
|
class AuditLog:
|
|
def __init__(self, database: str | Path) -> None:
|
|
self.database = str(database)
|
|
with sqlite3.connect(self.database) as db:
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
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
|
|
)
|
|
"""
|
|
)
|
|
|
|
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)
|
|
with sqlite3.connect(self.database) as db:
|
|
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"],
|
|
),
|
|
)
|
|
return digest
|
|
|
|
def verify(self) -> bool:
|
|
previous = "GENESIS"
|
|
with sqlite3.connect(self.database) 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:
|
|
body = {
|
|
"tenant_id": row[0],
|
|
"trace_id": row[1],
|
|
"event_type": row[2],
|
|
"payload": json.loads(row[3]),
|
|
"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
|