diff --git a/scripts/ci/verify_release_distribution_set.py b/scripts/ci/verify_release_distribution_set.py new file mode 100644 index 0000000000..a67f78239d --- /dev/null +++ b/scripts/ci/verify_release_distribution_set.py @@ -0,0 +1,294 @@ +#!/usr/bin/env python3 +"""Recheck a release distribution set against immutable Actions artifact bytes.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import stat +import subprocess +import tempfile +import zipfile +from contextlib import contextmanager +from datetime import datetime +from pathlib import Path +from typing import Any, BinaryIO, Callable, Iterable, Mapping + +SHA_RE = re.compile(r"^[0-9a-f]{40}$") +DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +NAME_RE = re.compile(r"^[A-Za-z0-9_.+-]+$") +REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +RECORD_MEMBERS = { + "reproducibility-record.tsv", + "release-scope-identities.json", + "release-gate-distribution-set.json", +} +ROW_KEYS = {"leg", "file", "sha256", "artifact_id", "artifact_name", "artifact_digest"} +MANIFEST_KEYS = { + "schema_version", "source_repository", "source_sha", "control_sha", + "run_id", "run_attempt", "distributions", +} +MAX_ARCHIVE_BYTES = 1024 * 1024 * 1024 +MAX_CONTROL_BYTES = 16 * 1024 * 1024 + + +class DistributionSetError(ValueError): + """Refuse an incomplete or unbound release distribution set.""" + + +def _strict_object(pairs: Iterable[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise DistributionSetError(f"duplicate JSON key: {key}") + result[key] = value + return result + + +def _reject_constant(value: str) -> Any: + raise DistributionSetError(f"non-finite JSON value: {value}") + + +def _json_bytes(data: bytes) -> Any: + if len(data) > MAX_CONTROL_BYTES: + raise DistributionSetError("control JSON is too large") + try: + return json.loads(data.decode("utf-8"), object_pairs_hook=_strict_object, + parse_constant=_reject_constant) + except (UnicodeError, json.JSONDecodeError) as error: + raise DistributionSetError("invalid control JSON") from error + + +def _timestamp(value: Any) -> datetime: + if not isinstance(value, str) or not value.endswith("Z"): + raise DistributionSetError("missing canonical UTC timestamp") + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as error: + raise DistributionSetError("invalid UTC timestamp") from error + if result.utcoffset() is None or result.utcoffset().total_seconds() != 0: + raise DistributionSetError("timestamp is not UTC") + return result + + +def _digest(value: Any) -> str: + if not isinstance(value, str) or DIGEST_RE.fullmatch(value) is None: + raise DistributionSetError("missing canonical artifact digest") + return value + + +def _artifact(artifacts: Mapping[str, Mapping[str, Any]], name: str, artifact_id: int, + digest: str, run_id: int, control_sha: str, started: datetime) -> None: + item = artifacts.get(name) + if item is None or type(artifact_id) is not int or artifact_id <= 0: + raise DistributionSetError(f"{name}: missing immutable artifact identity") + workflow_run = item.get("workflow_run") + if (type(item.get("id")) is not int or item["id"] != artifact_id + or item.get("digest") != digest + or item.get("expired") is not False + or not isinstance(workflow_run, Mapping) + or workflow_run.get("id") != run_id + or workflow_run.get("head_sha") != control_sha + or _timestamp(item.get("created_at")) <= started): + raise DistributionSetError(f"{name}: foreign, stale, or changed artifact metadata") + + +@contextmanager +def _archive(repository: str, artifact_id: int, digest: str, + fetch: Callable[[str, int, BinaryIO], None]): + with tempfile.TemporaryFile() as archive: + fetch(repository, artifact_id, archive) + if archive.tell() > MAX_ARCHIVE_BYTES: + raise DistributionSetError("artifact ZIP exceeds the size limit") + archive.seek(0) + actual = "sha256:" + hashlib.file_digest(archive, "sha256").hexdigest() + if actual != digest: + raise DistributionSetError("artifact ZIP digest mismatch") + archive.seek(0) + with zipfile.ZipFile(archive) as result: + yield result + + +def _members(archive: zipfile.ZipFile, expected: set[str]) -> dict[str, zipfile.ZipInfo]: + entries = archive.infolist() + if len(entries) != len(expected) or {entry.filename for entry in entries} != expected: + raise DistributionSetError("artifact ZIP members differ from the expected set") + for entry in entries: + mode = entry.external_attr >> 16 + if (NAME_RE.fullmatch(entry.filename) is None or entry.is_dir() + or stat.S_ISLNK(mode) or stat.S_IFMT(mode) not in (0, stat.S_IFREG) + or entry.file_size > (MAX_CONTROL_BYTES if expected == RECORD_MEMBERS else MAX_ARCHIVE_BYTES)): + raise DistributionSetError("unsafe or oversized artifact ZIP member") + return {entry.filename: entry for entry in entries} + + +def _record_rows(data: bytes, source_sha: str) -> dict[str, tuple[str, str]]: + if len(data) > MAX_CONTROL_BYTES: + raise DistributionSetError("reproducibility record is too large") + try: + lines = data.decode("utf-8").splitlines() + except UnicodeError as error: + raise DistributionSetError("invalid reproducibility record encoding") from error + header = ["target", "byte_verified", "verification", "sha256", "rebuild_sha256", "file", "build_env"] + if (len(lines) < 4 or re.fullmatch( + rf"# release [^ ]+ @ {source_sha}, SOURCE_DATE_EPOCH=[0-9]+", lines[0]) is None + or lines[1].split("\t") != header): + raise DistributionSetError("reproducibility record is not bound to the source") + rows: dict[str, tuple[str, str]] = {} + for line in lines[2:]: + fields = line.split("\t") + if (len(fields) != len(header) or fields[0] in rows + or fields[1] != "true" or fields[3] != fields[4] + or re.fullmatch(r"[0-9a-f]{64}", fields[3]) is None): + raise DistributionSetError("duplicate, unverified, or malformed record row") + rows[fields[0]] = (fields[5], fields[3]) + return rows + + +def verify_distribution_set(artifacts: Iterable[Any], attempt: Any, *, + repository: str, source_sha: str, + control_sha: str, run_id: int, run_attempt: int, + record_artifact_id: int, record_artifact_digest: str, + wheel_filename: str, sdist_filename: str, + fetch: Callable[[str, int, BinaryIO], None], + output_dir: Path) -> list[dict[str, Any]]: + """Require exact metadata and ZIP bytes for every declared distribution.""" + if (REPOSITORY_RE.fullmatch(repository) is None or SHA_RE.fullmatch(source_sha) is None + or SHA_RE.fullmatch(control_sha) is None or type(run_id) is not int or run_id <= 0 + or type(run_attempt) is not int or run_attempt <= 0): + raise DistributionSetError("invalid expected release identity") + if (not isinstance(attempt, Mapping) or type(attempt.get("id")) is not int + or attempt["id"] != run_id or type(attempt.get("run_attempt")) is not int + or attempt["run_attempt"] != run_attempt or attempt.get("head_sha") != control_sha): + raise DistributionSetError("workflow attempt differs from the caller") + started = _timestamp(attempt.get("run_started_at")) + listed: dict[str, Mapping[str, Any]] = {} + for item in artifacts: + if not isinstance(item, Mapping) or not isinstance(item.get("name"), str): + raise DistributionSetError("invalid artifact metadata entry") + name = item["name"] + if name in listed: + raise DistributionSetError(f"duplicate artifact name: {name}") + listed[name] = item + _artifact(listed, "reproducibility-record", record_artifact_id, + _digest(record_artifact_digest), run_id, control_sha, started) + with _archive(repository, record_artifact_id, record_artifact_digest, fetch) as archive: + _members(archive, RECORD_MEMBERS) + manifest = _json_bytes(archive.read("release-gate-distribution-set.json")) + record = archive.read("reproducibility-record.tsv") + if not isinstance(manifest, Mapping) or set(manifest) != MANIFEST_KEYS: + raise DistributionSetError("distribution manifest has an unknown shape") + if (type(manifest["schema_version"]) is not int or manifest["schema_version"] != 1 + or manifest["source_repository"] != repository + or manifest["source_sha"] != source_sha or manifest["control_sha"] != control_sha + or type(manifest["run_id"]) is not int or manifest["run_id"] != run_id + or type(manifest["run_attempt"]) is not int or manifest["run_attempt"] != run_attempt): + raise DistributionSetError("distribution manifest differs from the caller identity") + rows = manifest["distributions"] + if not isinstance(rows, list) or len(rows) < 2: + raise DistributionSetError("distribution manifest lacks wheel/sdist coverage") + record_rows = _record_rows(record, source_sha) + expected_names: set[str] = set() + seen_legs: set[str] = set() + seen_files: set[str] = set() + seen_ids = {record_artifact_id} + verified: list[dict[str, Any]] = [] + if output_dir.exists() or output_dir.is_symlink(): + raise DistributionSetError("distribution output already exists") + with tempfile.TemporaryDirectory(prefix=".release-set-", dir=output_dir.parent) as staging_name: + staging = Path(staging_name) + for row in rows: + if not isinstance(row, Mapping) or set(row) != ROW_KEYS: + raise DistributionSetError("invalid distribution row shape") + leg, filename, name = row["leg"], row["file"], row["artifact_name"] + artifact_id, digest, sha = row["artifact_id"], row["artifact_digest"], row["sha256"] + if (not isinstance(leg, str) or NAME_RE.fullmatch(leg) is None + or not isinstance(filename, str) or NAME_RE.fullmatch(filename) is None + or not isinstance(name, str) or name != ("dist-sdist" if leg == "sdist" else f"dist-wheel-{leg}") + or filename.endswith(".tar.gz") != (leg == "sdist") + or (leg != "sdist" and not filename.endswith(".whl")) + or not isinstance(sha, str) or re.fullmatch(r"[0-9a-f]{64}", sha) is None + or type(artifact_id) is not int or artifact_id in seen_ids + or leg in seen_legs or filename in seen_files): + raise DistributionSetError("duplicate or malformed distribution identity") + _artifact(listed, name, artifact_id, _digest(digest), run_id, control_sha, started) + if record_rows.get(leg) != (filename, sha): + raise DistributionSetError(f"{leg}: distribution differs from reproducibility record") + with _archive(repository, artifact_id, digest, fetch) as archive: + member = _members(archive, {filename})[filename] + with archive.open(member) as source, (staging / filename).open("xb") as target: + copied = hashlib.sha256() + size = 0 + for block in iter(lambda: source.read(1024 * 1024), b""): + size += len(block) + if size > MAX_ARCHIVE_BYTES: + raise DistributionSetError("distribution member exceeds the size limit") + copied.update(block) + target.write(block) + if copied.hexdigest() != sha or size != member.file_size: + raise DistributionSetError(f"{leg}: distribution bytes differ from the manifest") + expected_names.add(name) + seen_legs.add(leg) + seen_files.add(filename) + seen_ids.add(artifact_id) + verified.append(dict(row)) + if seen_legs != set(record_rows) or "sdist" not in seen_legs: + raise DistributionSetError("distribution set differs from reproducibility record") + if {name for name in listed if name.startswith("dist-")} != expected_names: + raise DistributionSetError("distribution artifact set is missing or has extras") + if (wheel_filename not in {row["file"] for row in verified if row["leg"] != "sdist"} + or sdist_filename != next(row["file"] for row in verified if row["leg"] == "sdist")): + raise DistributionSetError("selected wheel/sdist is not in the verified distribution set") + staging.rename(output_dir) + return verified + + +def fetch_artifact(repository: str, artifact_id: int, output: BinaryIO) -> None: + """Stream one immutable GitHub artifact ZIP without invoking a shell.""" + process = subprocess.Popen( + ["gh", "api", f"repos/{repository}/actions/artifacts/{artifact_id}/zip"], + stdout=subprocess.PIPE, + ) + try: + while block := process.stdout.read(1024 * 1024): + output.write(block) + if output.tell() > MAX_ARCHIVE_BYTES: + raise DistributionSetError("artifact ZIP exceeds the size limit") + if process.wait() != 0: + raise DistributionSetError("GitHub artifact download failed") + except BaseException: + if process.poll() is None: + process.kill() + process.wait() + raise + finally: + process.stdout.close() + + +def main() -> None: + parser = argparse.ArgumentParser() + for option in ("repository", "source-sha", "control-sha", "run-id", "run-attempt", + "record-artifact-id", "record-artifact-digest", "wheel-filename", + "sdist-filename", "metadata", "attempt", "output"): + parser.add_argument(f"--{option}", required=True) + args = parser.parse_args() + repository = args.repository + record_id = int(args.record_artifact_id) + record_digest = _digest(args.record_artifact_digest) + metadata = [_json_bytes(line.encode("utf-8")) for line in Path(args.metadata).read_text(encoding="utf-8").splitlines()] + attempt = _json_bytes(Path(args.attempt).read_bytes()) + verified = verify_distribution_set( + metadata, attempt, repository=repository, + source_sha=args.source_sha, control_sha=args.control_sha, + run_id=int(args.run_id), run_attempt=int(args.run_attempt), + record_artifact_id=record_id, record_artifact_digest=record_digest, + wheel_filename=args.wheel_filename, sdist_filename=args.sdist_filename, + fetch=fetch_artifact, output_dir=Path(args.output), + ) + print(json.dumps({"verified_distributions": verified}, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_verify_release_distribution_set.py b/tests/test_verify_release_distribution_set.py new file mode 100644 index 0000000000..c312df8b61 --- /dev/null +++ b/tests/test_verify_release_distribution_set.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import copy +import hashlib +import io +import json +import zipfile +from pathlib import Path + +import pytest + +from scripts.ci.verify_release_distribution_set import ( + DistributionSetError, + verify_distribution_set, +) + + +SOURCE = "a" * 40 +CONTROL = "b" * 40 +RUN = 424242 +ATTEMPT = 2 +CREATED = "2026-09-26T12:01:00Z" +STARTED = "2026-09-26T12:00:00Z" + + +def _zip(members: dict[str, bytes]) -> bytes: + output = io.BytesIO() + with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive: + for name, data in members.items(): + archive.writestr(name, data) + return output.getvalue() + + +def _sha(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _case() -> dict: + rows = [] + archives = {} + metadata = [] + record_lines = [ + f"# release v1.2.3 @ {SOURCE}, SOURCE_DATE_EPOCH=1", + "target\tbyte_verified\tverification\tsha256\trebuild_sha256\tfile\tbuild_env", + ] + for index in range(1, 14): + leg = "sdist" if index == 13 else f"target{index}-py3.12" + filename = "pkg-1.2.3.tar.gz" if index == 13 else f"pkg-1.2.3-{index}.whl" + name = "dist-sdist" if index == 13 else f"dist-wheel-{leg}" + data = f"verified bytes for {leg}".encode() + archive = _zip({filename: data}) + digest = "sha256:" + _sha(archive) + archives[index] = archive + metadata.append({"id": index, "name": name, "digest": digest, + "created_at": CREATED, "expired": False, + "workflow_run": {"id": RUN, "head_sha": CONTROL}}) + rows.append({"leg": leg, "file": filename, "sha256": _sha(data), + "artifact_id": index, "artifact_name": name, + "artifact_digest": digest}) + record_lines.append(f"{leg}\ttrue\tclean-target-repeat-same-env\t{_sha(data)}\t{_sha(data)}\t{filename}\trunner:x") + manifest = {"schema_version": 1, "source_repository": "owner/repo", + "source_sha": SOURCE, "control_sha": CONTROL, "run_id": RUN, + "run_attempt": ATTEMPT, "distributions": rows} + record = ("\n".join(record_lines) + "\n").encode() + case = {"manifest": manifest, "record": record, "archives": archives, + "metadata": metadata, "attempt": {"id": RUN, "run_attempt": ATTEMPT, + "head_sha": CONTROL, "run_started_at": STARTED}} + _repack_record(case) + return case + + +def _repack_record(case: dict) -> None: + archive = _zip({ + "reproducibility-record.tsv": case["record"], + "release-scope-identities.json": b"[]\n", + "release-gate-distribution-set.json": (json.dumps(case["manifest"]) + "\n").encode(), + }) + case["archives"][14] = archive + entry = {"id": 14, "name": "reproducibility-record", "digest": "sha256:" + _sha(archive), + "created_at": CREATED, "expired": False, + "workflow_run": {"id": RUN, "head_sha": CONTROL}} + case["metadata"] = [item for item in case["metadata"] if item["name"] != "reproducibility-record"] + [entry] + + +def _verify(case: dict, output: Path, *, wheel: str = "pkg-1.2.3-1.whl") -> list[dict]: + record_digest = next(item["digest"] for item in case["metadata"] if item["name"] == "reproducibility-record") + + def fetch(repository: str, artifact_id: int, destination) -> None: + assert repository == "owner/repo" + destination.write(case["archives"][artifact_id]) + + return verify_distribution_set( + case["metadata"], case["attempt"], repository="owner/repo", + source_sha=SOURCE, control_sha=CONTROL, run_id=RUN, run_attempt=ATTEMPT, + record_artifact_id=14, record_artifact_digest=record_digest, + wheel_filename=wheel, sdist_filename="pkg-1.2.3.tar.gz", + fetch=fetch, output_dir=output, + ) + + +def test_verifies_all_thirteen_immutable_artifact_archives(tmp_path: Path) -> None: + case = _case() + verified = _verify(case, tmp_path / "dist") + assert len(verified) == 13 + assert {path.name for path in (tmp_path / "dist").iterdir()} == { + row["file"] for row in case["manifest"]["distributions"] + } + for row in verified: + assert _sha((tmp_path / "dist" / row["file"]).read_bytes()) == row["sha256"] + + +def test_refuses_forged_missing_stale_and_tampered_sets(tmp_path: Path) -> None: + def missing(case): + case["manifest"]["distributions"].pop() + _repack_record(case) + + def wrong_source(case): + case["manifest"]["source_sha"] = "c" * 40 + _repack_record(case) + + def wrong_run(case): + case["metadata"][0]["workflow_run"]["id"] = 1 + + def earlier_attempt(case): + case["metadata"][0]["created_at"] = "2026-09-26T11:59:59Z" + + def tampered_zip(case): + case["archives"][1] = _zip({case["manifest"]["distributions"][0]["file"]: b"altered"}) + + def tampered_inner(case): + case["archives"][1] = _zip({case["manifest"]["distributions"][0]["file"]: b"altered"}) + digest = "sha256:" + _sha(case["archives"][1]) + case["metadata"][0]["digest"] = digest + case["manifest"]["distributions"][0]["artifact_digest"] = digest + _repack_record(case) + + def duplicate_id(case): + case["manifest"]["distributions"][1]["artifact_id"] = 1 + _repack_record(case) + + def extra_artifact(case): + case["metadata"].append({**copy.deepcopy(case["metadata"][0]), "id": 100, + "name": "dist-wheel-extra"}) + + def wrong_attempt(case): + case["attempt"]["run_attempt"] = 1 + + for name, mutate in ( + ("missing", missing), ("wrong-source", wrong_source), + ("wrong-run", wrong_run), ("earlier-attempt", earlier_attempt), + ("tampered-zip", tampered_zip), ("tampered-inner", tampered_inner), + ("duplicate-id", duplicate_id), + ("extra-artifact", extra_artifact), ("wrong-attempt", wrong_attempt), + ): + case = _case() + mutate(case) + output = tmp_path / name + with pytest.raises(DistributionSetError): + _verify(case, output) + assert not output.exists(), name + + with pytest.raises(DistributionSetError, match="selected wheel/sdist"): + _verify(_case(), tmp_path / "foreign-pair", wheel="../../outside.whl") + assert not (tmp_path / "foreign-pair").exists()