orchestrata_ultimate.factory.enforcement.attestation_store

Reference › orchestrata_ultimate › factory › enforcement › attestation_store

attestation_store is the append-only store of signed in-toto attestations (SPEC-033 section 7, row B20, ADR-0030). It follows the locked JSONL pattern of the enforcement ledger: _LockedJsonlFile holds an exclusive fcntl.flock on a sidecar lock file around each append, so concurrent writers serialise. Entries are immutable once written; there is no update or rewrite.

SignedAttestation bundles an InTotoLink, its SlsaProvenance, the signer's key id, a signature and a signing time. AttestationStore.attest takes a GovernedWriteRecord, maps it with link_from_record and provenance_for from attestation, signs the payload with a Signer and appends it. _signing_payload defines the exact bytes signed: the canonical link followed by the canonical provenance. verify (a static method) re-canonicalises an attestation and checks its signature, so tampering with the stored link or provenance is detected. attestations_for returns the attestations whose link environment matches a worktree and commit, and all returns every entry in insertion order. The _*_to_dict and _*_from_dict helpers convert between the dataclasses and JSON lines. The scripts.factory_enforce CLI imports it; tests are in tests/factory/test_attestation_store.py.

At a glance

Class diagram

Dependency diagram

API

Classes

SignedAttestation (dataclass)

Decorators: dataclass(frozen=True) · Line 43

One signed in-toto attestation: a link + its SLSA provenance, signed by the (Phase-1 HMAC, see :mod:.signing) platform key.

_LockedJsonlFile

Line 110

Locked/atomic append-only JSONL primitive (mirrors the append/ read-lines half of ledger._LockedFile; this store never rewrites the whole file, so the JSON-document half is not needed here).

AttestationStore

Line 153

Append-only JSONL store of signed in-toto attestations (B20).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/enforcement/attestation_store.py

"""Append-only, signed in-toto attestation store (ADR-0030, SPEC-033 §7, B20).

Mirrors ``orchestrata_ultimate.factory.enforcement.ledger``'s locked/atomic
JSONL pattern: an exclusive ``fcntl.flock`` on a sidecar lock file guards
every read-modify-write, and appends go straight to the file under that
lock (there is no whole-file rewrite here, unlike the ledger's JSON-document
half, since every entry is append-only). Two threads (or processes)
attesting concurrently serialize on the lock instead of corrupting the
file. ``SignedAttestation`` records are immutable once written -- there is
no update/rewrite, only append, same as :class:`.ledger.EventLedger`.
"""

from __future__ import annotations

import fcntl
import json
import os
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import IO, Any

from orchestrata_ultimate.factory.enforcement.attestation import (
    InTotoLink,
    SlsaProvenance,
    canonical_json,
    link_from_record,
    provenance_for,
)
from orchestrata_ultimate.factory.enforcement.models import GovernedWriteRecord
from orchestrata_ultimate.factory.enforcement.signing import Signer

# The lock-file handle type shared by _acquire/_release below (mirrors
# ledger.py's `_LockFile`).
_LockFile = IO[str]


def _utc_now_iso() -> str:
    return datetime.now(UTC).isoformat()


@dataclass(frozen=True)
class SignedAttestation:
    """One signed in-toto attestation: a link + its SLSA provenance, signed
    by the (Phase-1 HMAC, see :mod:`.signing`) platform key."""

    link: InTotoLink
    provenance: SlsaProvenance
    key_id: str
    signature: str
    signed_at: str


def _signing_payload(link: InTotoLink, provenance: SlsaProvenance) -> bytes:
    """The exact bytes that get signed: the canonical link followed by the
    canonical provenance, in that fixed order, so ``attest()`` and
    ``verify()`` always agree on what was signed. Any change to either --
    including a single tampered product hash -- changes these bytes and
    therefore the signature check."""
    return canonical_json(link) + canonical_json(provenance)


def _link_to_dict(link: InTotoLink) -> dict[str, Any]:
    return asdict(link)


def _link_from_dict(data: dict[str, Any]) -> InTotoLink:
    return InTotoLink(
        name=data["name"],
        materials=dict(data.get("materials", {})),
        products=dict(data.get("products", {})),
        byproducts=dict(data.get("byproducts", {})),
        environment=dict(data.get("environment", {})),
    )


def _provenance_to_dict(provenance: SlsaProvenance) -> dict[str, Any]:
    return asdict(provenance)


def _provenance_from_dict(data: dict[str, Any]) -> SlsaProvenance:
    return SlsaProvenance(
        builder_id=data["builder_id"],
        build_type=data["build_type"],
        invocation=dict(data.get("invocation", {})),
        slsa_level=data["slsa_level"],
    )


def _attestation_to_dict(attestation: SignedAttestation) -> dict[str, Any]:
    return {
        "link": _link_to_dict(attestation.link),
        "provenance": _provenance_to_dict(attestation.provenance),
        "key_id": attestation.key_id,
        "signature": attestation.signature,
        "signed_at": attestation.signed_at,
    }


def _dict_to_attestation(data: dict[str, Any]) -> SignedAttestation:
    return SignedAttestation(
        link=_link_from_dict(data["link"]),
        provenance=_provenance_from_dict(data["provenance"]),
        key_id=data["key_id"],
        signature=data["signature"],
        signed_at=data["signed_at"],
    )


class _LockedJsonlFile:
    """Locked/atomic append-only JSONL primitive (mirrors the append/
    read-lines half of ``ledger._LockedFile``; this store never rewrites
    the whole file, so the JSON-document half is not needed here)."""

    def __init__(self, path: Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.lock_path = self.path.with_name(self.path.name + ".lock")

    def _acquire(self) -> _LockFile:
        # 'a' (not 'w') so we never truncate a lock file a peer may hold.
        fp = open(self.lock_path, "a")
        fcntl.flock(fp.fileno(), fcntl.LOCK_EX)
        return fp

    @staticmethod
    def _release(fp: _LockFile) -> None:
        try:
            fcntl.flock(fp.fileno(), fcntl.LOCK_UN)
        finally:
            fp.close()

    def read_lines(self) -> list[dict[str, Any]]:
        """Read+parse all JSONL records. Caller must hold the lock."""
        if not self.path.exists():
            return []
        records: list[dict[str, Any]] = []
        with open(self.path, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    records.append(json.loads(line))
        return records

    def append_line(self, record: dict[str, Any]) -> None:
        """Append one JSON line. Caller must hold the lock."""
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(record) + "\n")
            f.flush()
            os.fsync(f.fileno())


class AttestationStore:
    """Append-only JSONL store of signed in-toto attestations (B20)."""

    def __init__(self, path: Path):
        self._store = _LockedJsonlFile(Path(path))

    def attest(
        self,
        record: GovernedWriteRecord,
        signer: Signer,
        builder_id: str,
    ) -> SignedAttestation:
        """Map ``record`` to an in-toto link + SLSA provenance, sign it,
        append the signed attestation to the store, and return it."""
        link = link_from_record(record)
        provenance = provenance_for(record, builder_id)
        signature = signer.sign(_signing_payload(link, provenance))
        attestation = SignedAttestation(
            link=link,
            provenance=provenance,
            key_id=signer.key_id,
            signature=signature,
            signed_at=_utc_now_iso(),
        )
        lock = self._store._acquire()
        try:
            self._store.append_line(_attestation_to_dict(attestation))
        finally:
            self._store._release(lock)
        return attestation

    @staticmethod
    def verify(attestation: SignedAttestation, signer: Signer) -> bool:
        """Re-canonicalize ``attestation``'s link + provenance and check the
        signature against ``signer``.

        A mismatched ``key_id`` fails closed before even checking the
        signature. A tampered field anywhere in the link or provenance
        (e.g. a mutated product hash) changes the canonical bytes and the
        signature check MUST then fail. A module-level function in all but
        name -- kept as a method for a stable ``store.verify(...)`` call
        site alongside ``attest``/``attestations_for``/``all``.
        """
        if attestation.key_id != signer.key_id:
            return False
        payload = _signing_payload(attestation.link, attestation.provenance)
        return signer.verify(payload, attestation.signature)

    def attestations_for(self, worktree: str, commit: str) -> list[SignedAttestation]:
        """Every attestation whose link's ``environment`` correlates to
        ``(worktree, commit)`` (see ``attestation.link_from_record``),
        insertion order."""
        lock = self._store._acquire()
        try:
            raw = self._store.read_lines()
        finally:
            self._store._release(lock)
        out: list[SignedAttestation] = []
        for r in raw:
            attestation = _dict_to_attestation(r)
            env = attestation.link.environment
            if env.get("worktree") == worktree and env.get("commit") == commit:
                out.append(attestation)
        return out

    def all(self) -> list[SignedAttestation]:
        """Return every attestation in the store, insertion order."""
        lock = self._store._acquire()
        try:
            raw = self._store.read_lines()
        finally:
            self._store._release(lock)
        return [_dict_to_attestation(r) for r in raw]