281 lines
8.3 KiB
Python
281 lines
8.3 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import base64
|
|
import binascii
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from enum import Enum
|
|
from typing import Any, Literal
|
|
|
|
from cryptography.exceptions import InvalidSignature
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
|
|
from cryptography.hazmat.primitives.serialization import load_pem_private_key, load_pem_public_key
|
|
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, *, include_signature: bool = False) -> dict[str, Any]:
|
|
payload = self.model_dump(mode="python")
|
|
if not include_signature:
|
|
payload.pop("signature", None)
|
|
return payload
|
|
|
|
def signing_bytes(self) -> bytes:
|
|
return canonical_json(self.canonical_payload()).encode("utf-8")
|
|
|
|
def sign(
|
|
self,
|
|
private_key: Ed25519PrivateKey | bytes | str,
|
|
) -> "BuildContract":
|
|
key = _load_private_key(private_key)
|
|
signature = base64.b64encode(key.sign(self.signing_bytes())).decode("ascii")
|
|
return self.model_copy(update={"signature": signature})
|
|
|
|
def verify_signature(self, public_key: Ed25519PublicKey | bytes | str) -> bool:
|
|
if self.signature is None:
|
|
return False
|
|
key = _load_public_key(public_key)
|
|
try:
|
|
key.verify(base64.b64decode(self.signature), self.signing_bytes())
|
|
except (ValueError, TypeError, binascii.Error, InvalidSignature):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _load_private_key(private_key: Ed25519PrivateKey | bytes | str) -> Ed25519PrivateKey:
|
|
if isinstance(private_key, Ed25519PrivateKey):
|
|
return private_key
|
|
if isinstance(private_key, str):
|
|
private_key = private_key.encode("utf-8")
|
|
key = load_pem_private_key(private_key, password=None)
|
|
if not isinstance(key, Ed25519PrivateKey):
|
|
raise TypeError("expected an Ed25519 private key")
|
|
return key
|
|
|
|
|
|
def _load_public_key(public_key: Ed25519PublicKey | bytes | str) -> Ed25519PublicKey:
|
|
if isinstance(public_key, Ed25519PublicKey):
|
|
return public_key
|
|
if isinstance(public_key, str):
|
|
public_key = public_key.encode("utf-8")
|
|
key = load_pem_public_key(public_key)
|
|
if not isinstance(key, Ed25519PublicKey):
|
|
raise TypeError("expected an Ed25519 public key")
|
|
return key
|
|
|
|
|
|
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)
|