spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : audit.py6# Description : Append-only, hash-chained audit log — every agent syscall is journaled here.7# =============================================================================8"""The audit log: who did what, in order, tamper-evident.910Every syscall an agent makes (successful OR failed) appends one record.11Records are hash-chained exactly like the ledger (SHA-256 over the canonical12JSON of the record plus the previous hash), so any after-the-fact edit,13deletion, or reordering breaks verification. Nothing is ever rewritten.14"""15from __future__ import annotations1617import hashlib18import json19from dataclasses import dataclass, field20from pathlib import Path21from typing import Any2223GENESIS_HASH = "0" * 642425_AUTHOR = "Simon-Pierre Boucher <contact@spboucher.ai>"262728def _canonical(record: dict[str, Any]) -> str:29    return json.dumps(record, sort_keys=True, separators=(",", ":"), default=str)303132@dataclass33class AuditLog:34    path: Path | None = None35    _records: list[dict[str, Any]] = field(default_factory=list)3637    def __post_init__(self) -> None:38        if self.path is not None and self.path.exists():39            for raw in self.path.read_text(encoding="utf-8").splitlines():40                if raw.strip():41                    self._records.append(json.loads(raw))4243    def append(44        self,45        actor: str,46        syscall: str,47        params: dict[str, Any],48        status: str,                    # "ok" | "error"49        detail: str = "",50        at: str | None = None,51    ) -> dict[str, Any]:52        record: dict[str, Any] = {53            "seq": len(self._records),54            "actor": actor,55            "syscall": syscall,56            "params": params,57            "status": status,58            "detail": detail,59            "at": at,60        }61        prev = self._records[-1]["hash"] if self._records else GENESIS_HASH62        record["prev_hash"] = prev63        record["hash"] = hashlib.sha256(64            (prev + _canonical({k: v for k, v in record.items() if k != "hash"}))65            .encode("utf-8")66        ).hexdigest()67        self._records.append(record)68        if self.path is not None:69            self.path.parent.mkdir(parents=True, exist_ok=True)70            with self.path.open("a", encoding="utf-8") as f:71                f.write(_canonical(record) + "\n")72        return record7374    def verify_chain(self) -> bool:75        prev = GENESIS_HASH76        for record in self._records:77            expected = hashlib.sha256(78                (prev + _canonical({k: v for k, v in record.items() if k != "hash"}))79                .encode("utf-8")80            ).hexdigest()81            if record.get("hash") != expected or record.get("prev_hash") != prev:82                return False83            prev = str(record["hash"])84        return True8586    @property87    def records(self) -> list[dict[str, Any]]:88        return list(self._records)89