proof-aware-crypto-tooling-agent/provider/src/pacta_provider/transparency_log.py

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

from pacta.dogfood import BACKEND_OPENSSL, BACKEND_VERIFIED, load_provenance, locate_verifier
from pacta.signing import canonical_json
from pacta.transparency import (
    HASH_ALGORITHM,
    RECEIPT_TYPE,
    attestation_leaf,
    consistency_proof,
    inclusion_proof,
    leaf_bytes_for_attestation,
    leaf_hash,
    make_signed_tree_head,
    merkle_root,
    proof_to_hex,
    verify_inclusion,
)
from pacta.yamlio import dump_data, load_data


@dataclass(slots=True)
class LogEntry:
    index: int
    leaf: dict[str, Any]
    leaf_hash: str

    def leaf_bytes(self) -> bytes:
        return canonical_json(self.leaf)


class TransparencyLog:
    def __init__(self, log_dir: str | Path) -> None:
        self.log_dir = Path(log_dir)
        self.metadata_path = self.log_dir / "metadata.json"
        self.entries_path = self.log_dir / "entries.jsonl"
        self.sth_path = self.log_dir / "sth.yaml"

    def init(self, provider: str, public_key_path: str | Path) -> dict[str, Any]:
        if self.metadata_path.exists() or self.entries_path.exists():
            raise ValueError(f"Transparency log already exists: {self.log_dir}")
        self.log_dir.mkdir(parents=True, exist_ok=True)
        public_key = Path(public_key_path).read_bytes()
        log_id = hashlib.sha256(b"pacta-log-v1\0" + provider.encode("utf-8") + b"\0" + public_key).hexdigest()
        metadata = {
            "schema_version": 1,
            "type": "pacta.transparency.log_metadata.v1",
            "provider": provider,
            "log_id": log_id,
            "hash_algorithm": HASH_ALGORITHM,
            "ed25519_public_key_fingerprint_sha256": hashlib.sha256(public_key).hexdigest(),
            "created_at": _now(),
            "standards": [
                "RFC 9162 Merkle tree hash and inclusion/consistency proof algorithms",
                "RFC 8032 Ed25519 signature verification via OpenSSL",
                "FIPS 204 ML-DSA signature slot; must be required by policy only when a backend is configured",
            ],
        }
        self.metadata_path.write_text(json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        self.entries_path.write_text("", encoding="utf-8")
        return metadata

    def metadata(self) -> dict[str, Any]:
        if not self.metadata_path.exists():
            raise ValueError(f"Transparency log is not initialized: {self.log_dir}")
        raw = json.loads(self.metadata_path.read_text(encoding="utf-8"))
        if not isinstance(raw, dict):
            raise ValueError(f"Invalid transparency log metadata: {self.metadata_path}")
        return raw

    def entries(self) -> list[LogEntry]:
        if not self.entries_path.exists():
            return []
        entries: list[LogEntry] = []
        for line_number, line in enumerate(self.entries_path.read_text(encoding="utf-8").splitlines(), start=1):
            if not line.strip():
                continue
            raw = json.loads(line)
            if not isinstance(raw, dict):
                raise ValueError(f"Invalid log entry at line {line_number}: {self.entries_path}")
            leaf = raw.get("leaf")
            if not isinstance(leaf, dict):
                raise ValueError(f"Invalid leaf at line {line_number}: {self.entries_path}")
            entries.append(LogEntry(index=int(raw["index"]), leaf=leaf, leaf_hash=str(raw["leaf_hash"])))
        return entries

    def latest_sth(
        self,
        private_key_path: str | Path,
        public_key_path: str | Path,
        timestamp: str | None = None,
    ) -> dict[str, Any]:
        metadata = self.metadata()
        entries = self.entries()
        leaves = [entry.leaf_bytes() for entry in entries]
        sth = make_signed_tree_head(
            metadata["log_id"],
            len(leaves),
            merkle_root(leaves).hex(),
            timestamp or _now(),
            private_key_path,
            public_key_path,
            signing_provenance=self.signing_provenance(entries),
        )
        dump_data(sth, self.sth_path)
        return sth

    def append_attestation(
        self,
        attestation_path: str | Path,
        private_key_path: str | Path,
        public_key_path: str | Path,
        receipt_out: str | Path | None = None,
    ) -> dict[str, Any]:
        metadata = self.metadata()
        attestation = load_data(attestation_path)
        if not isinstance(attestation, dict):
            raise ValueError(f"Attestation must be a mapping: {attestation_path}")
        entries = self.entries()
        previous_size = len(entries)
        previous_root = merkle_root([entry.leaf_bytes() for entry in entries]).hex()
        leaf = attestation_leaf(attestation)
        leaf_bytes = leaf_bytes_for_attestation(attestation)
        computed_leaf_hash = leaf_hash(leaf_bytes).hex()
        existing = next((entry for entry in entries if entry.leaf_hash == computed_leaf_hash), None)
        if existing:
            index = existing.index
            appended = False
        else:
            index = len(entries)
            record = {"index": index, "leaf_hash": computed_leaf_hash, "leaf": leaf}
            with self.entries_path.open("a", encoding="utf-8") as handle:
                handle.write(json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n")
            entries.append(LogEntry(index=index, leaf=leaf, leaf_hash=computed_leaf_hash))
            appended = True

        leaves = [entry.leaf_bytes() for entry in entries]
        sth = make_signed_tree_head(
            metadata["log_id"],
            len(leaves),
            merkle_root(leaves).hex(),
            _now(),
            private_key_path,
            public_key_path,
            signing_provenance=self.signing_provenance(entries),
        )
        dump_data(sth, self.sth_path)
        consistency = []
        if appended and previous_size > 0:
            consistency = proof_to_hex(consistency_proof(leaves, previous_size))
        receipt = {
            "schema_version": 1,
            "type": RECEIPT_TYPE,
            "log_id": metadata["log_id"],
            "hash_algorithm": HASH_ALGORITHM,
            "attestation_path": str(attestation_path),
            "attestation_digest_sha256": hashlib.sha256(canonical_json(attestation)).hexdigest(),
            "leaf_index": index,
            "leaf_hash": computed_leaf_hash,
            "tree_size": len(leaves),
            "inclusion_proof": proof_to_hex(inclusion_proof(leaves, index)),
            "consistency": {
                "from_tree_size": previous_size,
                "from_root_hash": previous_root,
                "proof": consistency,
                "status": "not_applicable" if previous_size == 0 or not appended else "included",
            },
            "sth": sth,
        }
        if receipt_out:
            Path(receipt_out).parent.mkdir(parents=True, exist_ok=True)
            dump_data(receipt, receipt_out)
        return receipt


    def signing_provenance(self, entries: list[LogEntry], signing_library_component: str = "dalek-ed25519-verified") -> dict[str, Any]:
        """THE PROVIDER EATS ITS OWN DOGFOOD: before signing a tree head,
        verify that the attestation of the SIGNING LIBRARY ITSELF (the
        merkleized dalek source whose build produced the signing binary) is
        included in the very tree about to be signed - by running the same
        Merkle inclusion verification an agent runs. The result is recorded
        inside the signature block: a root signature that names the leaf
        vouching for the code that produced it."""
        binary = locate_verifier()
        backend = BACKEND_VERIFIED if binary is not None else BACKEND_OPENSSL
        provenance: dict[str, Any] = {
            "signing_backend": backend,
            "signing_library_component": signing_library_component,
        }
        if binary is not None:
            build = load_provenance(binary)
            if build.get("source_commit"):
                provenance["signing_library_source_commit"] = build["source_commit"]
        matches = [
            entry
            for entry in entries
            if ((entry.leaf.get("attestation") or {}).get("subject") or {}).get("component") == signing_library_component
        ]
        if not matches:
            provenance["self_inclusion"] = "library_not_in_log"
            return provenance
        newest = matches[-1]
        leaves = [entry.leaf_bytes() for entry in entries]
        proof = inclusion_proof(leaves, newest.index)
        verified = verify_inclusion(
            newest.leaf_bytes(), newest.index, len(leaves), proof, merkle_root(leaves)
        )
        provenance["self_inclusion"] = "verified" if verified else "FAILED"
        provenance["signing_library_leaf_index"] = newest.index
        attested = (newest.leaf.get("attestation") or {}).get("certificates") or []
        clean = sum(1 for c in attested if c.get("status") == "proven" and c.get("axiom_status") == "clean")
        provenance["signing_library_certificates_proven"] = f"{clean}/{len(attested)}"
        return provenance

    def consistency_from(self, old_tree_size: int) -> dict[str, Any]:
        """Consistency proof from an arbitrary earlier tree size - what a
        pinning agent requests when its pin is older than the receipt's
        embedded from_tree_size."""
        entries = self.entries()
        leaves = [entry.leaf_bytes() for entry in entries]
        if old_tree_size < 0 or old_tree_size > len(leaves):
            raise ValueError(f"old_tree_size {old_tree_size} outside tree size {len(leaves)}")
        return {
            "schema_version": 1,
            "type": "pacta.transparency.consistency_proof.v1",
            "log_id": self.metadata()["log_id"],
            "from_tree_size": old_tree_size,
            "from_root_hash": merkle_root(leaves[:old_tree_size]).hex(),
            "to_tree_size": len(leaves),
            "to_root_hash": merkle_root(leaves).hex(),
            "proof": proof_to_hex(consistency_proof(leaves, old_tree_size)),
        }

    def audit(self) -> dict[str, Any]:
        """Monitor-side self-check: recompute every prefix root, confirm the
        stored STH matches the full tree, and confirm every prefix is
        consistent with the final tree (append-only structure)."""
        entries = self.entries()
        leaves = [entry.leaf_bytes() for entry in entries]
        problems: list[str] = []
        for position, entry in enumerate(entries):
            if entry.index != position:
                problems.append(f"Entry at position {position} carries index {entry.index}.")
            if leaf_hash(entry.leaf_bytes()).hex() != entry.leaf_hash:
                problems.append(f"Entry {position} leaf_hash does not match its leaf bytes.")
        computed_root = merkle_root(leaves).hex()
        stored_sth = load_data(self.sth_path) if self.sth_path.exists() else None
        if stored_sth:
            if stored_sth.get("tree_size") != len(leaves):
                problems.append("Stored STH tree_size does not match the entry count.")
            if stored_sth.get("root_hash") != computed_root:
                problems.append("Stored STH root hash does not match the recomputed tree root.")
        return {
            "tree_size": len(leaves),
            "computed_root": computed_root,
            "stored_sth_root": (stored_sth or {}).get("root_hash"),
            "problems": problems,
            "ok": not problems,
        }


def _now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")