fix(safety): il boot-check dell'audit tollera il lag benigno dell'anchor

L'anchor dell'audit (system_state.last_audit_hash) e' persistito best-effort:
sotto contesa di lock SQLite il mirror puo' restare indietro rispetto al log,
che invece cresce in avanti integro. Il check di boot armava il kill-switch
su qualsiasi disuguaglianza anchor!=coda, scambiando questo lag per
manomissione.

- audit_log: nuova tail_continues_from() — True solo se l'anchor e' un
  antenato valido della coda (presente in catena + chain forward integra a EOF)
- orchestrator._verify_audit_anchor: se lag benigno -> re-sync anchor + warning;
  solo anchor assente o chain rotta (troncamento/tamper) -> critical -> arma

Verificato live: anchor in ritardo -> re-sync senza armare; anchor bogus -> arma.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-05-29 19:55:19 +00:00
parent a5a39a6d27
commit 647e3e565f
2 changed files with 86 additions and 11 deletions
+50
View File
@@ -28,6 +28,7 @@ __all__ = [
"AuditChainError",
"AuditEntry",
"AuditLog",
"tail_continues_from",
"verify_chain",
]
@@ -157,6 +158,55 @@ def verify_chain(path: str | Path) -> int:
return count
def tail_continues_from(path: str | Path, anchor_hash: str) -> bool:
"""Return ``True`` when *anchor_hash* is a genuine ancestor of the tail.
That is: *anchor_hash* is the ``hash`` of some line in the log AND every
line after it forms a valid forward chain to EOF. This distinguishes a
**benign anchor lag** — the best-effort anchor persisted in
:func:`cerbero_bite.runtime.dependencies.build_runtime` fell behind the
file under SQLite write contention, yet the log grew forward
legitimately — from real **truncation/tampering** (anchor absent, or the
post-anchor chain broken).
Returns ``False`` if the file is missing/empty, the anchor is not found,
or the chain from the anchor to the tail does not verify. The single-
writer invariant of :class:`AuditLog` still holds; this only makes the
boot-time anchor check tolerant of the durability gap it documents.
"""
p = Path(path)
if not p.exists() or p.stat().st_size == 0:
return False
seen = False
expected_prev = ""
with p.open("r", encoding="utf-8") as fh:
for line in fh:
if not line.strip():
continue
try:
entry = _parse_line(line)
except AuditChainError:
if seen:
return False
continue
if seen:
if entry.prev_hash != expected_prev:
return False
recomputed = _compute_hash(
entry.timestamp.isoformat(),
entry.event,
_canonical_payload(entry.payload),
entry.prev_hash,
)
if recomputed != entry.hash:
return False
expected_prev = entry.hash
elif entry.hash == anchor_hash:
seen = True
expected_prev = entry.hash
return seen
def iter_entries(path: str | Path) -> Iterator[AuditEntry]:
"""Yield each :class:`AuditEntry` from *path* without verifying."""
p = Path(path)