class ChainedAttestationLog:
"""Append-only hash-chained JSONL log signed with an Ed25519 key.
Args:
path: JSONL log file (created if missing; resumed if present).
service: Signing service. Its public key is exported beside the
log as ``<path>.pub.pem`` for offline verifiers.
checkpoint_interval: If > 0, automatically append a checkpoint
record after every N regular appends.
anchor_callback: Called with an :class:`AnchorEvent` after every
checkpoint (manual or automatic). Exceptions propagate.
"""
def __init__(
self,
path: str | Path,
service: AttestationService,
*,
checkpoint_interval: int = 0,
anchor_callback: Callable[[AnchorEvent], None] | None = None,
) -> None:
self._path = Path(path)
self._service = service
self._checkpoint_interval = checkpoint_interval
self._anchor_callback = anchor_callback
self._appends_since_checkpoint = 0
self._fh: IO[bytes] | None = None
self._lock_fd: int | None = None
# `seq` and `prev_sha256` are read, used and advanced across several
# statements in `_append`. Two threads sharing one log — which is what
# `Engine(audit_sink=...)` invites, and what the Engine's own
# thread-safety promise implies — interleaved there and produced two
# lines claiming the same seq. Every evaluation returned normally and
# the resulting file verified as tampered.
self._write_lock = threading.Lock()
self._path.parent.mkdir(parents=True, exist_ok=True)
self._fingerprint = key_fingerprint(service.public_key)
state = _scan(self._path)
self._corruption = state.error
self._next_seq = state.count
self._head_sha256 = state.head_sha256
self.public_key_path = self._path.with_name(self._path.name + ".pub.pem")
if state.count == 0 and self._corruption is None:
# New (or empty) log: mint an identity and write the genesis
# record so every subsequent line is bound to this log.
self._log_id = uuid.uuid4().hex
self.public_key_path.write_bytes(service.public_key_pem())
self._append(
{
"type": GENESIS_RECORD_TYPE,
"log_id": self._log_id,
"key_fingerprint": self._fingerprint,
}
)
else:
self._log_id = state.log_id or ""
if self._corruption is None and state.genesis_fingerprint != self._fingerprint:
# Signing with a different key than the genesis pinned would
# produce a log no single public key can verify: fail closed.
self._corruption = (
f"signing key fingerprint {self._fingerprint} does not match "
f"log genesis key fingerprint {state.genesis_fingerprint}"
)
if state.genesis_fingerprint == self._fingerprint:
# Re-export only when this service holds the log's pinned
# key, so opening with the wrong key can never clobber the
# correct public key exported beside the log.
self.public_key_path.write_bytes(service.public_key_pem())
@property
def path(self) -> Path:
return self._path
@property
def log_id(self) -> str:
"""The log's identity from its genesis record (empty if corrupt)."""
return self._log_id
@property
def head_seq(self) -> int | None:
return self._next_seq - 1 if self._next_seq else None
@property
def head_sha256(self) -> str | None:
return self._head_sha256
@property
def corruption(self) -> str | None:
"""Description of detected corruption, if any. Appends fail-closed."""
return self._corruption
def append(self, record: dict[str, Any]) -> ChainedRecord:
"""Sign and append one record; returns the written line's metadata.
Raises AttestationError if the log is corrupt (fail closed).
"""
chained = self._append(record)
self._appends_since_checkpoint += 1
if self._checkpoint_interval > 0 and (
self._appends_since_checkpoint >= self._checkpoint_interval
):
self.checkpoint()
return chained
def write(self, record: Any) -> None:
"""Append an audit record, satisfying :class:`fathom.audit.AuditSink`.
Lets the chained log stand in for ``FileSink`` wherever a sink is
taken -- ``Engine(audit_sink=ChainedAttestationLog(...))`` -- so
evaluation records land in a chain whose lines commit to their
predecessors, rather than in a flat file where a deletion leaves no
trace. Pydantic records are dumped in JSON mode so the line stays
canonicalisable; mappings (the transports' hot-reload events) pass
through as-is.
"""
dump = getattr(record, "model_dump", None)
self.append(dump(mode="json") if dump is not None else dict(record))
def checkpoint(self) -> ChainedRecord:
"""Append a signed checkpoint record pinning the current head.
Fires ``anchor_callback`` with the new head (the checkpoint line's
own hash) and the checkpoint's portable JWS token.
"""
chained = self._append(
{
"type": CHECKPOINT_RECORD_TYPE,
"head_seq": self.head_seq,
"head_sha256": self._head_sha256,
}
)
self._appends_since_checkpoint = 0
if self._anchor_callback is not None:
self._anchor_callback(
AnchorEvent(
seq=chained.seq,
head_sha256=chained.line_sha256,
checkpoint_jws=chained.jws,
)
)
return chained
def _append(self, record: dict[str, Any]) -> ChainedRecord:
with self._write_lock:
return self._append_locked(record)
def _refuse_if_corrupt(self) -> None:
if self._corruption is not None:
raise AttestationError(
f"chained log {self._path} is corrupt; refusing append: {self._corruption}",
operation="append",
)
def _append_locked(self, record: dict[str, Any]) -> ChainedRecord:
self._refuse_if_corrupt()
try:
record_sha256 = _sha256_hex(_canonical(record))
except ValueError as exc:
raise AttestationError(
f"record is not canonical JSON: {exc}", operation="append"
) from exc
fh = self._fh if self._fh is not None else self._open_for_append()
# The open handle bypasses the rescan, so a log deleted or truncated
# under it would take lines that chain onto a head no longer on disk
# (an unlinked inode swallows them outright). Check the file is still
# linked and still exactly as long as this writer left it.
st = os.fstat(fh.fileno())
if st.st_nlink == 0 or st.st_size != fh.tell():
self._corruption = (
f"log on disk changed under the open writer (size {st.st_size}, "
f"expected {fh.tell()}, links {st.st_nlink}); it was truncated, "
"emptied, replaced or deleted"
)
self._refuse_if_corrupt()
iat = int(time.time())
seq = self._next_seq
prev = self._head_sha256
claims = {
"iss": CHAIN_ISSUER,
"iat": iat,
"seq": seq,
"prev_sha256": prev,
"record_sha256": record_sha256,
"log_id": self._log_id,
"v": FORMAT_VERSION,
}
token = self._service.sign_claims(claims, headers={"kid": self._fingerprint})
line = _canonical(
{
"iat": iat,
"jws": token,
"prev_sha256": prev,
"record": record,
"seq": seq,
"v": FORMAT_VERSION,
}
)
fh.write(line + b"\n")
fh.flush()
os.fsync(fh.fileno())
self._next_seq = seq + 1
self._head_sha256 = _sha256_hex(line)
return ChainedRecord(
seq=seq,
iat=iat,
prev_sha256=prev,
record=record,
jws=token,
line_sha256=self._head_sha256,
)
def _open_for_append(self) -> IO[bytes]:
"""Claim the writer lock, then re-read the head the lock now guards.
The head in memory came from the scan at construction, or from this
object's last append before it closed. Another writer may have
appended since, between our open and its close -- a rolling restart
does exactly that -- so the lock alone does not make that head
current. Rescanning under the lock does.
"""
lock_fd = _claim_exclusive_writer(self._path)
try:
state = _scan(self._path)
if state.error is not None:
self._corruption = state.error
elif state.count and state.log_id != self._log_id:
self._corruption = (
f"log_id {state.log_id} on disk is not this log's {self._log_id}; "
"the file was replaced"
)
elif state.count < self._next_seq:
# This writer already wrote seq `_next_seq - 1`. Fewer lines
# on disk means the tail was cut, or the whole log deleted or
# emptied; chaining onto what is left would hide the loss
# (a cut tail re-verifies ok) or write a log with no genesis.
self._corruption = (
f"log has {state.count} lines on disk but this writer's head is "
f"seq {self._next_seq - 1}; it was truncated, emptied or deleted"
)
self._refuse_if_corrupt()
fd = os.open(self._path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
except BaseException:
os.close(lock_fd)
raise
self._next_seq = state.count
self._head_sha256 = state.head_sha256
self._lock_fd = lock_fd
self._fh = os.fdopen(fd, "ab")
return self._fh
def records(self) -> list[ChainedRecord]:
"""All parseable records. Raises on a malformed line.
Chain/signature validity is the job of :meth:`verify`.
"""
result: list[ChainedRecord] = []
if not self._path.exists():
return result
with open(self._path, "rb") as fh:
for lineno, raw in enumerate(fh, start=1):
try:
result.append(_parse_line(raw.rstrip(b"\n")))
except (ValueError, KeyError, TypeError, RecursionError) as exc:
raise AttestationError(
f"malformed line {lineno} in {self._path}", operation="records"
) from exc
return result
def find_record(self, seq: int) -> ChainedRecord | None:
"""Return the record at *seq*, or None."""
for rec in self.records():
if rec.seq == seq:
return rec
return None
def verify(self) -> ChainVerification:
"""Full offline verification using this log's own public key."""
return verify_chain(self._path, self._service.public_key)
def close(self) -> None:
# Under the append lock: closing the handle mid-append leaked a raw
# "write to closed file", and closing the lock fd twice could release
# a descriptor number another open had reused.
with self._write_lock:
if self._fh is not None:
self._fh.close()
self._fh = None
if self._lock_fd is not None:
os.close(self._lock_fd)
self._lock_fd = None
def __enter__(self) -> ChainedAttestationLog:
return self
def __exit__(self, *exc: object) -> None:
self.close()