fathom.ChainedAttestationLog

fathom.ChainedAttestationLog

Append-only hash-chained JSONL log signed with an Ed25519 key.

Parameters:

Name	Type	Description	Default
path	str | Path	JSONL log file (created if missing; resumed if present).	required
service	AttestationService	Signing service. Its public key is exported beside the log as <path>.pub.pem for offline verifiers.	required
checkpoint_interval	int	If > 0, automatically append a checkpoint record after every N regular appends.	0
anchor_callback	Callable[[AnchorEvent], None] | None	Called with an :class:AnchorEvent after every checkpoint (manual or automatic). Exceptions propagate.	None

Source code in src/fathom/chained_log.py

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 _append_locked(self, record: dict[str, Any]) -> ChainedRecord:
        if self._corruption is not None:
            raise AttestationError(
                f"chained log {self._path} is corrupt; refusing append: {self._corruption}",
                operation="append",
            )
        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": _sha256_hex(_canonical(record)),
            "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,
            }
        )
        if self._fh is None:
            lock_fd = _claim_exclusive_writer(self._path)
            try:
                fd = os.open(self._path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
            except BaseException:
                os.close(lock_fd)
                raise
            self._lock_fd = lock_fd
            self._fh = os.fdopen(fd, "ab")
        self._fh.write(line + b"\n")
        self._fh.flush()
        os.fsync(self._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 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) 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:
        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()

log_id property

The log's identity from its genesis record (empty if corrupt).

corruption property

Description of detected corruption, if any. Appends fail-closed.

append(record)

Sign and append one record; returns the written line's metadata.

Raises AttestationError if the log is corrupt (fail closed).

Source code in src/fathom/chained_log.py

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

write(record)

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.

Source code in src/fathom/chained_log.py

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))

checkpoint()

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.

Source code in src/fathom/chained_log.py

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

records()

All parseable records. Raises on a malformed line.

Chain/signature validity is the job of :meth:verify.

Source code in src/fathom/chained_log.py

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) as exc:
                raise AttestationError(
                    f"malformed line {lineno} in {self._path}", operation="records"
                ) from exc
    return result

find_record(seq)

Return the record at seq, or None.

Source code in src/fathom/chained_log.py

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

verify()

Full offline verification using this log's own public key.

Source code in src/fathom/chained_log.py

def verify(self) -> ChainVerification:
    """Full offline verification using this log's own public key."""
    return verify_chain(self._path, self._service.public_key)