tegwick
146f4e1dd3
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1046e-465c-7d43-8c4c-d10f0a1feadd
276 lines
8.7 KiB
Python
276 lines
8.7 KiB
Python
"""Hash-chain integrity for operational custody (AUDIT-WP-0007).
|
|
|
|
See ``docs/integrity.md`` for the proof bound. This module is the shared
|
|
arithmetic; backends persist and walk rows.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Mapping
|
|
|
|
SCHEMA = "audit-core.chain-head.v1"
|
|
# How old a chain-head attestation may be before the tamper-evidence claim
|
|
# degrades. Declared in ``docs/integrity.md`` alongside the attestation
|
|
# cadence; the two are one contract and must move together.
|
|
DEFAULT_ATTESTATION_MAX_AGE_HOURS = 168.0
|
|
GENESIS = "0" * 64
|
|
# Documented advisory-lock key so concurrent accepts cannot fork the head.
|
|
CHAIN_LOCK_KEY = 0xA0D17007
|
|
|
|
|
|
def chain_link(previous: str, payload_hash: str, event_id: str) -> str:
|
|
"""Next ``chain_hash`` = SHA-256(previous || payload_hash || event_id)."""
|
|
material = f"{previous}|{payload_hash}|{event_id}".encode("utf-8")
|
|
return hashlib.sha256(material).hexdigest()
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ChainRow:
|
|
event_id: str
|
|
payload_hash: str
|
|
chain_hash: str
|
|
chain_prev: str
|
|
accepted_at: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ChainReport:
|
|
intact: bool
|
|
events: int
|
|
head: str
|
|
head_event_id: str | None
|
|
head_accepted_at: str | None
|
|
first_break: str | None
|
|
attestation_match: bool | None = None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"intact": self.intact,
|
|
"events": self.events,
|
|
"head": self.head,
|
|
"head_event_id": self.head_event_id,
|
|
"head_accepted_at": self.head_accepted_at,
|
|
"first_break": self.first_break,
|
|
"attestation_match": self.attestation_match,
|
|
}
|
|
|
|
|
|
def order_chain_rows(rows: Iterable[ChainRow]) -> list[ChainRow]:
|
|
"""Follow stored links, independently of historical acceptance timestamps.
|
|
|
|
Keep unreachable/forked rows in the result so verify_rows reports a break
|
|
rather than silently dropping evidence outside the genesis chain.
|
|
"""
|
|
remaining = list(rows)
|
|
successors: dict[str, list[ChainRow]] = {}
|
|
for row in remaining:
|
|
successors.setdefault(row.chain_prev, []).append(row)
|
|
ordered: list[ChainRow] = []
|
|
used: set[str] = set()
|
|
previous = GENESIS
|
|
while len(successors.get(previous, [])) == 1:
|
|
row = successors[previous][0]
|
|
if row.event_id in used:
|
|
break
|
|
ordered.append(row)
|
|
used.add(row.event_id)
|
|
previous = row.chain_hash
|
|
return ordered + [row for row in remaining if row.event_id not in used]
|
|
|
|
|
|
def verify_rows(
|
|
rows: Iterable[ChainRow],
|
|
*,
|
|
attestation: Mapping[str, Any] | None = None,
|
|
) -> ChainReport:
|
|
"""Walk accept order and recompute every link.
|
|
|
|
``first_break`` is an event id, or ``attestation_mismatch`` when a cited
|
|
head is not present in the live chain.
|
|
"""
|
|
ordered = list(rows)
|
|
prev = GENESIS
|
|
first_break: str | None = None
|
|
for row in ordered:
|
|
expected = chain_link(prev, row.payload_hash, row.event_id)
|
|
if row.chain_prev != prev or row.chain_hash != expected:
|
|
first_break = row.event_id
|
|
break
|
|
prev = row.chain_hash
|
|
|
|
intact = first_break is None
|
|
last = ordered[-1] if ordered else None
|
|
head = last.chain_hash if last else GENESIS
|
|
report = ChainReport(
|
|
intact=intact,
|
|
events=len(ordered),
|
|
head=head,
|
|
head_event_id=last.event_id if last else None,
|
|
head_accepted_at=last.accepted_at if last else None,
|
|
first_break=first_break,
|
|
)
|
|
if attestation is None:
|
|
return report
|
|
return _apply_attestation(report, ordered, attestation)
|
|
|
|
|
|
def _apply_attestation(
|
|
report: ChainReport,
|
|
rows: list[ChainRow],
|
|
attestation: Mapping[str, Any],
|
|
) -> ChainReport:
|
|
cited = str(attestation.get("chain_hash") or "")
|
|
live_hashes = {row.chain_hash for row in rows}
|
|
empty_ok = not rows and cited in {"", GENESIS}
|
|
matched = cited in live_hashes or empty_ok
|
|
first_break = report.first_break
|
|
intact = report.intact
|
|
if not matched:
|
|
intact = False
|
|
if first_break is None:
|
|
first_break = "attestation_mismatch"
|
|
return ChainReport(
|
|
intact=intact,
|
|
events=report.events,
|
|
head=report.head,
|
|
head_event_id=report.head_event_id,
|
|
head_accepted_at=report.head_accepted_at,
|
|
first_break=first_break,
|
|
attestation_match=matched,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TamperEvidenceState:
|
|
"""Whether the ``tamper_evidence`` claim is currently earned.
|
|
|
|
``docs/integrity.md`` permits the claim only while both preconditions are
|
|
live. Absence or staleness degrades it — it does not leave it standing.
|
|
"""
|
|
|
|
claimed: bool
|
|
reason: str
|
|
observed_at: str | None = None
|
|
age_seconds: float | None = None
|
|
max_age_seconds: float | None = None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"tamper_evidence": self.claimed,
|
|
"reason": self.reason,
|
|
"attestation_observed_at": self.observed_at,
|
|
"attestation_age_seconds": self.age_seconds,
|
|
"attestation_max_age_seconds": self.max_age_seconds,
|
|
}
|
|
|
|
|
|
def _parse_iso(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(str(value))
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed
|
|
|
|
|
|
def evaluate_tamper_evidence(
|
|
report: ChainReport | None,
|
|
attestation: Mapping[str, Any] | None,
|
|
*,
|
|
max_age_hours: float = DEFAULT_ATTESTATION_MAX_AGE_HOURS,
|
|
now: datetime | None = None,
|
|
) -> TamperEvidenceState:
|
|
"""Derive the tamper-evidence claim from live state.
|
|
|
|
``report`` must be the chain report produced *against* ``attestation``, so
|
|
that ``attestation_match`` reflects the cited head. ``report is None``
|
|
means the chain could not be walked at all, which is not a licence to keep
|
|
claiming.
|
|
"""
|
|
max_age_seconds = float(max_age_hours) * 3600.0
|
|
if report is None:
|
|
return TamperEvidenceState(
|
|
False, "chain_unreadable", max_age_seconds=max_age_seconds
|
|
)
|
|
if not report.intact:
|
|
return TamperEvidenceState(
|
|
False, "chain_break", max_age_seconds=max_age_seconds
|
|
)
|
|
if attestation is None:
|
|
return TamperEvidenceState(
|
|
False, "no_attestation", max_age_seconds=max_age_seconds
|
|
)
|
|
observed_at = str(attestation.get("observed_at") or "") or None
|
|
if report.attestation_match is not True:
|
|
return TamperEvidenceState(
|
|
False,
|
|
"attestation_mismatch",
|
|
observed_at=observed_at,
|
|
max_age_seconds=max_age_seconds,
|
|
)
|
|
stamped = _parse_iso(observed_at)
|
|
if stamped is None:
|
|
return TamperEvidenceState(
|
|
False,
|
|
"attestation_undated",
|
|
observed_at=observed_at,
|
|
max_age_seconds=max_age_seconds,
|
|
)
|
|
reference = now or datetime.now(timezone.utc)
|
|
if reference.tzinfo is None:
|
|
reference = reference.replace(tzinfo=timezone.utc)
|
|
age_seconds = (reference - stamped).total_seconds()
|
|
if age_seconds > max_age_seconds:
|
|
return TamperEvidenceState(
|
|
False,
|
|
"attestation_stale",
|
|
observed_at=observed_at,
|
|
age_seconds=age_seconds,
|
|
max_age_seconds=max_age_seconds,
|
|
)
|
|
return TamperEvidenceState(
|
|
True,
|
|
"attested",
|
|
observed_at=observed_at,
|
|
age_seconds=age_seconds,
|
|
max_age_seconds=max_age_seconds,
|
|
)
|
|
|
|
|
|
def attestation_from_report(report: ChainReport, *, observed_at: str | None = None) -> dict[str, Any]:
|
|
return {
|
|
"schema": SCHEMA,
|
|
"genesis": GENESIS,
|
|
"chain_hash": report.head,
|
|
"event_id": report.head_event_id,
|
|
"accepted_at": report.head_accepted_at,
|
|
"event_count": report.events,
|
|
"observed_at": observed_at or utc_now(),
|
|
}
|
|
|
|
|
|
def load_attestation(path: str | Path) -> dict[str, Any]:
|
|
payload = json.loads(Path(path).read_text())
|
|
if not isinstance(payload, dict) or not payload.get("chain_hash"):
|
|
raise ValueError("attestation must be an object with chain_hash")
|
|
return payload
|
|
|
|
|
|
def write_attestation(path: str | Path, report: ChainReport) -> dict[str, Any]:
|
|
body = attestation_from_report(report)
|
|
destination = Path(path)
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text(json.dumps(body, indent=2, sort_keys=True) + "\n")
|
|
return body
|