idea33-buildledger-verifiable-oss/buildledger/models.py

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def _to_jsonable(value: Any) -> Any:
    if isinstance(value, BaseModel):
        return _to_jsonable(value.model_dump(mode="python"))
    if isinstance(value, datetime):
        return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
    if isinstance(value, Enum):
        return value.value
    if isinstance(value, dict):
        return {str(k): _to_jsonable(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_to_jsonable(v) for v in value]
    if isinstance(value, set):
        return sorted(_to_jsonable(v) for v in value)
    return value


def canonical_json(value: Any) -> str:
    return json.dumps(_to_jsonable(value), sort_keys=True, separators=(",", ":"), ensure_ascii=True)


def sha256_text(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)


class BuildEnvironment(StrictModel):
    platform: str
    operating_system: str
    architecture: str
    toolchain: dict[str, str]
    container_image: str | None = None
    env_hash: str | None = None
    variables: dict[str, str] = Field(default_factory=dict)

    @field_validator("toolchain")
    @classmethod
    def _toolchain_not_empty(cls, value: dict[str, str]) -> dict[str, str]:
        if not value:
            raise ValueError("toolchain must not be empty")
        return value


class BuildObject(StrictModel):
    source_repo: str
    commit: str
    builder: str
    target: str
    build_tool: str
    flags: list[str] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=utcnow)

    @field_validator("commit")
    @classmethod
    def _commit_not_empty(cls, value: str) -> str:
        if not value:
            raise ValueError("commit must not be empty")
        return value


class DependencyNode(StrictModel):
    name: str
    version: str
    digest: str
    kind: Literal["library", "tool", "plugin"] = "library"
    source: str | None = None


class DependencyEdge(StrictModel):
    source: str
    target: str
    relation: Literal["depends_on", "builds_with", "imports"] = "depends_on"


class DependencyGraph(StrictModel):
    nodes: list[DependencyNode] = Field(default_factory=list)
    edges: list[DependencyEdge] = Field(default_factory=list)


class Artifact(StrictModel):
    name: str
    path: str
    sha256: str
    size_bytes: int = Field(ge=0)
    media_type: str = "application/octet-stream"

    @field_validator("sha256")
    @classmethod
    def _sha256_format(cls, value: str) -> str:
        if len(value) != 64:
            raise ValueError("sha256 must be a 64-character hex digest")
        int(value, 16)
        return value


class BuildLog(StrictModel):
    stream: Literal["stdout", "stderr"]
    sha256: str
    line_count: int = Field(ge=0)
    excerpt: list[str] = Field(default_factory=list)

    @field_validator("sha256")
    @classmethod
    def _sha256_format(cls, value: str) -> str:
        if len(value) != 64:
            raise ValueError("sha256 must be a 64-character hex digest")
        int(value, 16)
        return value


class ReproManifest(StrictModel):
    source_commit: str
    deterministic_inputs_hash: str
    toolchain_versions: dict[str, str] = Field(default_factory=dict)
    env_hash: str
    entropy_seed: str | None = None
    reproducibility_score: float = Field(ge=0.0, le=1.0)
    run_count: int = Field(default=1, ge=1)
    matched_artifact_bytes: int = Field(default=0, ge=0)
    total_artifact_bytes: int = Field(default=0, ge=0)


class BuildContract(StrictModel):
    contract_id: str
    version: str = "1.0"
    build_object: BuildObject
    environment: BuildEnvironment
    dependency_graph: DependencyGraph = Field(default_factory=DependencyGraph)
    artifacts: list[Artifact] = Field(default_factory=list)
    logs: list[BuildLog] = Field(default_factory=list)
    manifest: ReproManifest
    signature: str | None = None
    notes: list[str] = Field(default_factory=list)

    def canonical_payload(self) -> dict[str, Any]:
        return self.model_dump(mode="python")


class LedgerEntry(StrictModel):
    sequence: int = Field(ge=0)
    timestamp: datetime = Field(default_factory=utcnow)
    kind: str
    payload: dict[str, Any]
    previous_hash: str
    entry_hash: str
    anchor_ref: str | None = None

    @field_validator("previous_hash", "entry_hash")
    @classmethod
    def _hash_format(cls, value: str) -> str:
        if len(value) != 64:
            raise ValueError("hash values must be 64-character hex digests")
        int(value, 16)
        return value

    @classmethod
    def compute_hash(
        cls,
        *,
        sequence: int,
        timestamp: datetime,
        kind: str,
        payload: dict[str, Any],
        previous_hash: str,
    ) -> str:
        material = canonical_json(
            {
                "sequence": sequence,
                "timestamp": timestamp,
                "kind": kind,
                "payload": payload,
                "previous_hash": previous_hash,
            }
        )
        return sha256_text(material)

    @classmethod
    def create(
        cls,
        *,
        sequence: int,
        kind: str,
        payload: dict[str, Any],
        previous_hash: str,
        timestamp: datetime | None = None,
        anchor_ref: str | None = None,
    ) -> "LedgerEntry":
        ts = timestamp or utcnow()
        entry_hash = cls.compute_hash(
            sequence=sequence,
            timestamp=ts,
            kind=kind,
            payload=payload,
            previous_hash=previous_hash,
        )
        return cls(
            sequence=sequence,
            timestamp=ts,
            kind=kind,
            payload=payload,
            previous_hash=previous_hash,
            entry_hash=entry_hash,
            anchor_ref=anchor_ref,
        )


class AnchorRecord(StrictModel):
    method: str
    reference: str
    anchored_hash: str
    created_at: datetime = Field(default_factory=utcnow)
    metadata: dict[str, str] = Field(default_factory=dict)