160 lines
6.0 KiB
Python
160 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
from buildledger.adapters import GitHubActionsAdapter
|
|
from buildledger.anchors import AnchorStore
|
|
from buildledger.cli import main as cli_main
|
|
from buildledger.ledger import SQLiteLedger
|
|
from buildledger.models import Artifact, BuildContract, BuildEnvironment, BuildLog, BuildObject, DependencyGraph, DependencyNode, ReproManifest
|
|
from buildledger.schemas import SchemaRegistry
|
|
from buildledger.sync import SyncCursor, apply_delta, prepare_delta
|
|
|
|
|
|
def make_contract() -> BuildContract:
|
|
build_object = BuildObject(
|
|
source_repo="https://example.com/repo.git",
|
|
commit="abc123",
|
|
builder="gha",
|
|
target="artifact.tar.gz",
|
|
build_tool="python",
|
|
flags=["--locked"],
|
|
)
|
|
environment = BuildEnvironment(
|
|
platform="linux",
|
|
operating_system="ubuntu-24.04",
|
|
architecture="x86_64",
|
|
toolchain={"python": "3.11.8"},
|
|
container_image="ghcr.io/example/build:1",
|
|
env_hash="f" * 64,
|
|
)
|
|
manifest = ReproManifest(
|
|
source_commit="abc123",
|
|
deterministic_inputs_hash="a" * 64,
|
|
toolchain_versions={"python": "3.11.8"},
|
|
env_hash="f" * 64,
|
|
reproducibility_score=1.0,
|
|
run_count=2,
|
|
matched_artifact_bytes=100,
|
|
total_artifact_bytes=100,
|
|
)
|
|
return BuildContract(
|
|
contract_id="contract-1",
|
|
build_object=build_object,
|
|
environment=environment,
|
|
dependency_graph=DependencyGraph(nodes=[DependencyNode(name="dep", version="1.0.0", digest="d" * 64)]),
|
|
artifacts=[Artifact(name="artifact", path="dist/artifact.tar.gz", sha256="b" * 64, size_bytes=100)],
|
|
logs=[BuildLog(stream="stdout", sha256="c" * 64, line_count=1, excerpt=["ok"])],
|
|
manifest=manifest,
|
|
)
|
|
|
|
|
|
def write_ed25519_keypair(tmp_path: Path) -> tuple[Path, Path]:
|
|
private_key = Ed25519PrivateKey.generate()
|
|
public_key = private_key.public_key()
|
|
private_key_path = tmp_path / "private.pem"
|
|
public_key_path = tmp_path / "public.pem"
|
|
private_key_path.write_text(
|
|
private_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
encryption_algorithm=serialization.NoEncryption(),
|
|
).decode("utf-8"),
|
|
encoding="utf-8",
|
|
)
|
|
public_key_path.write_text(
|
|
public_key.public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
|
).decode("utf-8"),
|
|
encoding="utf-8",
|
|
)
|
|
return private_key_path, public_key_path
|
|
|
|
|
|
def test_schema_registry_validates_contract() -> None:
|
|
registry = SchemaRegistry.with_defaults()
|
|
contract = make_contract()
|
|
loaded = registry.validate("BuildContract", "1.0", contract.model_dump(mode="python"))
|
|
assert isinstance(loaded, BuildContract)
|
|
assert loaded.contract_id == "contract-1"
|
|
assert registry.manifest()["BuildContract"] == ["1.0"]
|
|
|
|
|
|
def test_ledger_chain_and_anchor_round_trip(tmp_path: Path) -> None:
|
|
ledger = SQLiteLedger(tmp_path / "ledger.sqlite")
|
|
first = ledger.append("build.contract", make_contract().model_dump(mode="python"))
|
|
second = ledger.append("build.proof", {"status": "verified", "contract": first.entry_hash})
|
|
|
|
ok, issues = ledger.verify_chain()
|
|
assert ok
|
|
assert issues == []
|
|
assert second.sequence == 1
|
|
|
|
anchor = AnchorStore(ledger).record("cloud-anchor", "s3://bucket/buildledger")
|
|
assert anchor.anchored_hash == second.entry_hash
|
|
assert ledger.anchors()[0]["anchored_hash"] == second.entry_hash
|
|
|
|
|
|
def test_delta_sync_applies_to_fresh_ledger(tmp_path: Path) -> None:
|
|
source = SQLiteLedger(tmp_path / "source.sqlite")
|
|
source.append("build.contract", make_contract().model_dump(mode="python"))
|
|
source.append("build.proof", {"status": "verified"})
|
|
|
|
packet = prepare_delta(source, SyncCursor(sequence=-1, entry_hash="0" * 64))
|
|
target = SQLiteLedger(tmp_path / "target.sqlite")
|
|
cursor = apply_delta(target, packet)
|
|
|
|
assert cursor.sequence == 1
|
|
assert target.verify_chain()[0]
|
|
assert len(target.all_entries()) == 2
|
|
|
|
|
|
def test_ci_adapter_generates_contract() -> None:
|
|
context = {
|
|
"repo": "https://example.com/repo.git",
|
|
"commit": "abc123",
|
|
"builder": "gha",
|
|
"target": "artifact.tar.gz",
|
|
"build_tool": "python",
|
|
"platform": "github-actions",
|
|
"os": "ubuntu-24.04",
|
|
"arch": "x86_64",
|
|
"toolchain.python": "3.11.8",
|
|
"env.PYTHONHASHSEED": "0",
|
|
"flags": "--locked",
|
|
"run_count": "2",
|
|
"reproducibility_score": "1.0",
|
|
}
|
|
contract = GitHubActionsAdapter().contract_from_context(context)
|
|
assert contract.build_object.commit == "abc123"
|
|
assert contract.environment.variables["env.PYTHONHASHSEED"] == "0"
|
|
assert contract.manifest.run_count == 2
|
|
|
|
|
|
def test_contract_signing_and_cli_round_trip(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
private_key_path, public_key_path = write_ed25519_keypair(tmp_path)
|
|
contract = make_contract()
|
|
|
|
signed = contract.sign(private_key_path.read_text(encoding="utf-8"))
|
|
assert signed.signature is not None
|
|
assert signed.verify_signature(public_key_path.read_text(encoding="utf-8"))
|
|
|
|
tampered = signed.model_copy(update={"notes": ["tampered"]})
|
|
assert not tampered.verify_signature(public_key_path.read_text(encoding="utf-8"))
|
|
|
|
contract_path = tmp_path / "contract.json"
|
|
signed_path = tmp_path / "signed.json"
|
|
contract_path.write_text(contract.model_dump_json(indent=2) + "\n", encoding="utf-8")
|
|
|
|
assert cli_main(["sign-contract", str(contract_path), str(private_key_path), "--output", str(signed_path)]) == 0
|
|
assert cli_main(["verify-contract", str(signed_path), str(public_key_path)]) == 0
|
|
|
|
out = capsys.readouterr().out
|
|
assert json.loads(out)["ok"] is True
|