rustd-resolved 0.2.1

#!/usr/bin/python3 -I
"""Download and import replacement proof artifacts for one GitHub SHA."""

from __future__ import annotations

import argparse
import copy
from dataclasses import dataclass
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request


class DownloadError(RuntimeError):
    pass


class NoRedirect(urllib.request.HTTPRedirectHandler):
    """Expose redirect responses so cross-host credentials are never forwarded."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
        return None


@dataclass(frozen=True)
class ProofSpec:
    artifact_prefix: str
    status_context: str
    workflow_path: str
    workflow_name: str
    dispatch_ref: str = "main"


ARTIFACTS = {
    "reproducible-release": ProofSpec(
        "replacement-reproducible-release-proof-",
        "replacement/reproducible-release",
        ".github/workflows/reproducible-release.yml",
        "Reproducible replacement release",
    ),
    "upstream-test-75": ProofSpec(
        "replacement-upstream-test-75-proof-",
        "replacement/upstream-test-75",
        ".github/workflows/replacement-upstream-test-75.yml",
        "Replacement upstream TEST-75 proof",
    ),
    "upstream-test-89-mdns": ProofSpec(
        "replacement-upstream-test-89-mdns-proof-",
        "replacement/upstream-test-89-mdns",
        ".github/workflows/replacement-upstream-test-89-mdns.yml",
        "Replacement upstream TEST-89 mDNS proof",
    ),
    "security-suite": ProofSpec(
        "replacement-security-proof-",
        "replacement/security-proof",
        ".github/workflows/replacement-security-proof.yml",
        "Replacement security proof",
    ),
    "boot-replacement": ProofSpec(
        "replacement-boot-proof-",
        "replacement/boot-proof",
        ".github/workflows/replacement-boot-proof.yml",
        "Replacement boot and rollback proof",
    ),
}


def _positive_integer(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool) and value > 0


def validate_repository(repository: str) -> None:
    if re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository) is None:
        raise DownloadError("repository must be an exact owner/name")


def validate_sha(sha: str) -> None:
    if re.fullmatch(r"[0-9a-f]{40}", sha) is None:
        raise DownloadError("SHA must be a lowercase full Git commit identifier")


def validate_workflow(spec: ProofSpec, workflow: dict[str, object]) -> int:
    workflow_id = workflow.get("id")
    if not _positive_integer(workflow_id):
        raise DownloadError(f"workflow has no positive ID: {spec.workflow_path}")
    if workflow.get("path") != spec.workflow_path:
        raise DownloadError(f"workflow path does not match {spec.workflow_path}")
    if workflow.get("name") != spec.workflow_name:
        raise DownloadError(f"workflow name does not match {spec.workflow_name}")
    if workflow.get("state") != "active":
        raise DownloadError(f"workflow is not active: {spec.workflow_path}")
    return workflow_id


def _run_path(spec: ProofSpec) -> str:
    return spec.workflow_path


def _timestamp(value: object, label: str) -> datetime:
    if not isinstance(value, str):
        raise DownloadError(f"{label} is not a timestamp")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as error:
        raise DownloadError(f"{label} is not a valid timestamp") from error
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise DownloadError(f"{label} has no timezone")
    return parsed.astimezone(timezone.utc)


def select_status_run(
    payload: dict[str, object], repository: str, sha: str, context: str
) -> tuple[int, str]:
    if payload.get("sha") != sha:
        raise DownloadError("combined status is not for the requested SHA")
    status_repository = payload.get("repository")
    if not isinstance(status_repository, dict) or status_repository.get("full_name") != repository:
        raise DownloadError("combined status is not for the requested repository")
    statuses = payload.get("statuses")
    if not isinstance(statuses, list):
        raise DownloadError("combined status has no statuses list")
    matching: list[tuple[str, int, dict[str, object]]] = []
    for item in statuses:
        if not isinstance(item, dict) or item.get("context") != context:
            continue
        created_at = item.get("created_at")
        status_id = item.get("id")
        if not isinstance(created_at, str) or not created_at or not _positive_integer(status_id):
            raise DownloadError(f"status {context} has no stable ordering identity")
        matching.append((created_at, status_id, item))
    if not matching:
        raise DownloadError(f"exact-source status {context} does not exist")
    status = max(matching, key=lambda value: (value[0], value[1]))[2]
    if status.get("state") != "success":
        raise DownloadError(f"exact-source status {context} is not successful")
    target_url = status.get("target_url")
    if not isinstance(target_url, str):
        raise DownloadError(f"exact-source status {context} has no target run")
    parsed = urllib.parse.urlsplit(target_url)
    expected_prefix = f"/{repository}/actions/runs/"
    if (
        parsed.scheme != "https"
        or not parsed.netloc
        or parsed.username is not None
        or parsed.password is not None
        or parsed.query
        or parsed.fragment
        or not parsed.path.startswith(expected_prefix)
    ):
        raise DownloadError(f"exact-source status {context} has an invalid target run URL")
    run_text = parsed.path[len(expected_prefix) :]
    if re.fullmatch(r"[1-9][0-9]*", run_text) is None:
        raise DownloadError(f"exact-source status {context} has an invalid target run ID")
    return int(run_text), target_url


def validate_run(
    spec: ProofSpec,
    repository: str,
    sha: str,
    workflow_id: int,
    run_id: int,
    status_target_url: str,
    run: dict[str, object],
) -> tuple[int, int, int, str]:
    if run.get("id") != run_id:
        raise DownloadError(f"workflow run ID does not match for {spec.workflow_path}")
    if run.get("workflow_id") != workflow_id:
        raise DownloadError(f"workflow run is from the wrong workflow for {spec.workflow_path}")
    expected_run_name = f"{spec.workflow_name} {sha}"
    if run.get("path") != _run_path(spec) or run.get("name") != expected_run_name:
        raise DownloadError(f"workflow run identity does not match {spec.workflow_path}")
    if run.get("display_title") != expected_run_name:
        raise DownloadError(f"workflow run title is not bound to {sha}")
    if run.get("event") != "workflow_dispatch":
        raise DownloadError(f"workflow run was not explicitly dispatched for {spec.workflow_path}")
    if run.get("status") != "completed" or run.get("conclusion") != "success":
        raise DownloadError(f"workflow run did not complete successfully for {spec.workflow_path}")
    attempt = run.get("run_attempt")
    if not _positive_integer(attempt):
        raise DownloadError(f"workflow run has no positive attempt for {spec.workflow_path}")
    if run.get("head_branch") != spec.dispatch_ref:
        raise DownloadError(f"workflow run used the wrong dispatch ref for {spec.workflow_path}")
    head_sha = run.get("head_sha")
    if not isinstance(head_sha, str) or re.fullmatch(r"[0-9a-f]{40}", head_sha) is None:
        raise DownloadError(f"workflow run has an invalid dispatch SHA for {spec.workflow_path}")
    if run.get("html_url") != status_target_url:
        raise DownloadError(f"status target does not exactly identify the workflow run for {spec.workflow_path}")
    run_repository = run.get("repository")
    head_repository = run.get("head_repository")
    if not isinstance(run_repository, dict) or run_repository.get("full_name") != repository:
        raise DownloadError(f"workflow run is from the wrong repository for {spec.workflow_path}")
    if not isinstance(head_repository, dict) or head_repository.get("full_name") != repository:
        raise DownloadError(f"workflow run head is from the wrong repository for {spec.workflow_path}")
    repository_id = run_repository.get("id")
    head_repository_id = head_repository.get("id")
    if not _positive_integer(repository_id) or not _positive_integer(head_repository_id):
        raise DownloadError(f"workflow run has invalid repository IDs for {spec.workflow_path}")
    return attempt, repository_id, head_repository_id, head_sha


def validate_attempt(
    spec: ProofSpec,
    repository: str,
    sha: str,
    workflow_id: int,
    run_id: int,
    attempt_number: int,
    repository_id: int,
    head_repository_id: int,
    head_sha: str,
    attempt: dict[str, object],
) -> tuple[datetime, datetime]:
    expected_fields = {
        "id": run_id,
        "workflow_id": workflow_id,
        "path": _run_path(spec),
        "name": f"{spec.workflow_name} {sha}",
        "display_title": f"{spec.workflow_name} {sha}",
        "event": "workflow_dispatch",
        "status": "completed",
        "conclusion": "success",
        "run_attempt": attempt_number,
        "head_branch": spec.dispatch_ref,
        "head_sha": head_sha,
    }
    for field, expected in expected_fields.items():
        if attempt.get(field) != expected:
            raise DownloadError(f"exact workflow attempt has mismatched {field}: {spec.workflow_path}")
    attempt_repository = attempt.get("repository")
    attempt_head_repository = attempt.get("head_repository")
    if (
        not isinstance(attempt_repository, dict)
        or attempt_repository.get("full_name") != repository
        or attempt_repository.get("id") != repository_id
        or not isinstance(attempt_head_repository, dict)
        or attempt_head_repository.get("full_name") != repository
        or attempt_head_repository.get("id") != head_repository_id
    ):
        raise DownloadError(f"exact workflow attempt has mismatched repository identity: {spec.workflow_path}")
    started_at = _timestamp(attempt.get("run_started_at"), "attempt run_started_at")
    updated_at = _timestamp(attempt.get("updated_at"), "attempt updated_at")
    if updated_at < started_at:
        raise DownloadError(f"exact workflow attempt has an invalid time window: {spec.workflow_path}")
    return started_at, updated_at


def validate_run_artifact(
    payload: dict[str, object],
    artifact_name: str,
    run_id: int,
    repository: str,
    repository_id: int,
    head_repository_id: int,
    head_branch: str,
    head_sha: str,
    attempt_started_at: datetime,
    attempt_updated_at: datetime,
) -> dict[str, object]:
    """Bind one run-scoped artifact to the selected attempt's time window.

    GitHub's REST API has an exact workflow-attempt metadata endpoint, but no
    attempt-scoped artifact-list endpoint. Therefore this validation is
    deliberately fail-closed: the expected run artifact must be unique and its
    creation timestamp must fall inside the exact attempt's observed window.
    """
    artifacts = payload.get("artifacts")
    total_count = payload.get("total_count")
    if (
        not isinstance(artifacts, list)
        or not isinstance(total_count, int)
        or isinstance(total_count, bool)
        or total_count < 0
        or any(not isinstance(item, dict) for item in artifacts)
    ):
        raise DownloadError("workflow run returned an invalid artifact listing")
    if total_count != len(artifacts):
        raise DownloadError("workflow run artifact listing is incomplete")
    named = [item for item in artifacts if item.get("name") == artifact_name]
    if len(named) != 1:
        raise DownloadError(
            f"workflow run must contain exactly one artifact named {artifact_name}; "
            "GitHub exposes no attempt-scoped artifact listing"
        )
    artifact = named[0]
    artifact_id = artifact.get("id")
    if not _positive_integer(artifact_id) or artifact.get("expired") is not False:
        raise DownloadError(f"workflow run artifact is invalid or expired: {artifact_name}")
    artifact_created_at = _timestamp(artifact.get("created_at"), "artifact created_at")
    if not attempt_started_at <= artifact_created_at <= attempt_updated_at:
        raise DownloadError(
            f"artifact {artifact_name} cannot be bound to the exact attempt time window; "
            "GitHub exposes only a run-scoped artifact listing"
        )
    workflow_run = artifact.get("workflow_run")
    if not isinstance(workflow_run, dict):
        raise DownloadError(f"workflow artifact has no workflow run: {artifact_name}")
    expected_run_fields = {
        "id": run_id,
        "repository_id": repository_id,
        "head_repository_id": head_repository_id,
        "head_branch": head_branch,
        "head_sha": head_sha,
    }
    for field, expected in expected_run_fields.items():
        if workflow_run.get(field) != expected:
            raise DownloadError(f"artifact {artifact_name} has mismatched workflow-run {field}")
    archive_url = artifact.get("archive_download_url")
    if not isinstance(archive_url, str):
        raise DownloadError(f"artifact has no download URL: {artifact_name}")
    parsed = urllib.parse.urlsplit(archive_url)
    expected_suffix = f"/repos/{repository}/actions/artifacts/{artifact_id}/zip"
    if (
        parsed.scheme != "https"
        or not parsed.netloc
        or parsed.username is not None
        or parsed.password is not None
        or parsed.query
        or parsed.fragment
        or not parsed.path.endswith(expected_suffix)
    ):
        raise DownloadError(f"artifact has an invalid download URL: {artifact_name}")
    return artifact


def _expect_download_error(callback) -> None:  # noqa: ANN001
    try:
        callback()
    except DownloadError:
        return
    raise AssertionError("invalid provenance fixture was accepted")


def _self_test() -> None:
    repository = "owner/repository"
    sha = "a" * 40
    dispatch_sha = "b" * 40
    spec = ProofSpec(
        "proof-",
        "replacement/example",
        ".github/workflows/example.yml",
        "Replacement example proof",
    )
    workflow = {
        "id": 17,
        "path": spec.workflow_path,
        "name": spec.workflow_name,
        "state": "active",
    }
    target_url = "https://github.com/owner/repository/actions/runs/23"
    status_payload = {
        "sha": sha,
        "repository": {"full_name": repository},
        "statuses": [
            {
                "id": 29,
                "context": spec.status_context,
                "created_at": "2026-01-01T00:00:00Z",
                "state": "success",
                "target_url": target_url,
            }
        ],
    }
    run = {
        "id": 23,
        "workflow_id": 17,
        "path": _run_path(spec),
        "name": f"{spec.workflow_name} {sha}",
        "display_title": f"{spec.workflow_name} {sha}",
        "event": "workflow_dispatch",
        "status": "completed",
        "conclusion": "success",
        "run_attempt": 2,
        "head_branch": spec.dispatch_ref,
        "head_sha": dispatch_sha,
        "html_url": target_url,
        "repository": {"id": 31, "full_name": repository},
        "head_repository": {"id": 31, "full_name": repository},
    }
    attempt_payload = copy.deepcopy(run)
    attempt_payload.update(
        {
            "run_started_at": "2026-01-01T00:00:00Z",
            "updated_at": "2026-01-01T00:10:00Z",
        }
    )
    artifact_name = spec.artifact_prefix + sha
    artifact = {
        "id": 37,
        "name": artifact_name,
        "expired": False,
        "created_at": "2026-01-01T00:05:00Z",
        "archive_download_url": (
            "https://api.github.com/repos/owner/repository/actions/artifacts/37/zip"
        ),
        "workflow_run": {
            "id": 23,
            "repository_id": 31,
            "head_repository_id": 31,
            "head_branch": spec.dispatch_ref,
            "head_sha": dispatch_sha,
        },
    }
    run_artifacts = {"total_count": 1, "artifacts": [artifact]}

    validate_repository(repository)
    validate_sha(sha)
    assert validate_workflow(spec, workflow) == 17
    assert select_status_run(status_payload, repository, sha, spec.status_context) == (
        23,
        target_url,
    )
    assert validate_run(spec, repository, sha, 17, 23, target_url, run) == (
        2,
        31,
        31,
        dispatch_sha,
    )
    started_at, updated_at = validate_attempt(
        spec,
        repository,
        sha,
        17,
        23,
        2,
        31,
        31,
        dispatch_sha,
        attempt_payload,
    )
    assert validate_run_artifact(
        run_artifacts,
        artifact_name,
        23,
        repository,
        31,
        31,
        spec.dispatch_ref,
        dispatch_sha,
        started_at,
        updated_at,
    )["id"] == 37

    _expect_download_error(lambda: validate_sha("A" * 40))
    _expect_download_error(lambda: validate_repository("owner/repository/extra"))
    for field, invalid in (
        ("id", 0),
        ("path", ".github/workflows/other.yml"),
        ("name", "Other workflow"),
        ("state", "disabled_manually"),
    ):
        changed = copy.deepcopy(workflow)
        changed[field] = invalid
        _expect_download_error(lambda changed=changed: validate_workflow(spec, changed))

    for field, invalid in (
        ("sha", "c" * 40),
        ("repository", {"full_name": "other/repository"}),
    ):
        changed = copy.deepcopy(status_payload)
        changed[field] = invalid
        _expect_download_error(
            lambda changed=changed: select_status_run(
                changed, repository, sha, spec.status_context
            )
        )
    for field, invalid in (
        ("state", "pending"),
        ("target_url", target_url + "?attempt=2"),
    ):
        changed = copy.deepcopy(status_payload)
        changed["statuses"][0][field] = invalid
        _expect_download_error(
            lambda changed=changed: select_status_run(
                changed, repository, sha, spec.status_context
            )
        )

    for field, invalid in (
        ("workflow_id", 18),
        ("path", ".github/workflows/other.yml"),
        ("name", "Other workflow"),
        ("display_title", spec.workflow_name),
        ("event", "push"),
        ("status", "in_progress"),
        ("conclusion", "failure"),
        ("run_attempt", 0),
        ("head_branch", "feature"),
    ):
        changed = copy.deepcopy(run)
        changed[field] = invalid
        _expect_download_error(
            lambda changed=changed: validate_run(
                spec, repository, sha, 17, 23, target_url, changed
            )
        )
    changed = copy.deepcopy(run)
    changed["repository"]["full_name"] = "other/repository"
    _expect_download_error(
        lambda: validate_run(spec, repository, sha, 17, 23, target_url, changed)
    )

    for field, invalid in (
        ("run_attempt", 1),
        ("workflow_id", 18),
        ("path", ".github/workflows/other.yml"),
        ("display_title", spec.workflow_name),
        ("run_started_at", "not-a-timestamp"),
        ("updated_at", "2025-12-31T23:59:59Z"),
    ):
        changed = copy.deepcopy(attempt_payload)
        changed[field] = invalid
        _expect_download_error(
            lambda changed=changed: validate_attempt(
                spec,
                repository,
                sha,
                17,
                23,
                2,
                31,
                31,
                dispatch_sha,
                changed,
            )
        )

    def duplicate_artifact(value: dict[str, object]) -> None:
        value["total_count"] = 2
        value["artifacts"].append(copy.deepcopy(value["artifacts"][0]))

    for mutate in (
        lambda value: value.update({"total_count": 0}),
        duplicate_artifact,
        lambda value: value["artifacts"][0].update({"expired": True}),
        lambda value: value["artifacts"][0]["workflow_run"].update({"id": 24}),
        lambda value: value["artifacts"][0].update(
            {"created_at": "2025-12-31T23:59:59Z"}
        ),
        lambda value: value["artifacts"][0].update(
            {"archive_download_url": "https://example.invalid/artifact.zip"}
        ),
    ):
        changed = copy.deepcopy(run_artifacts)
        mutate(changed)
        _expect_download_error(
            lambda changed=changed: validate_run_artifact(
                changed,
                artifact_name,
                23,
                repository,
                31,
                31,
                spec.dispatch_ref,
                dispatch_sha,
                started_at,
                updated_at,
            )
        )


def api_request(url: str, token: str) -> dict[str, object]:
    request = urllib.request.Request(
        url,
        headers={
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
        },
    )
    with urllib.request.urlopen(request, timeout=60) as response:
        value = json.load(response)
    if not isinstance(value, dict):
        raise DownloadError(f"GitHub response is not an object: {url}")
    return value


def _copy_response(response, output: Path) -> None:  # noqa: ANN001
    with output.open("wb") as stream:
        while True:
            chunk = response.read(1024 * 1024)
            if not chunk:
                break
            stream.write(chunk)


def download(url: str, token: str, output: Path) -> None:
    """Download a GitHub artifact without forwarding its bearer token off-host."""
    request = urllib.request.Request(
        url,
        headers={
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
        },
    )
    opener = urllib.request.build_opener(NoRedirect())
    try:
        response = opener.open(request, timeout=60)
    except urllib.error.HTTPError as error:
        if error.code not in {301, 302, 303, 307, 308}:
            raise
        location = error.headers.get("Location")
        error.close()
        if not location:
            raise DownloadError("artifact redirect did not include a Location header") from error
        signed_url = urllib.parse.urljoin(url, location)
        signed_request = urllib.request.Request(
            signed_url,
            headers={
                "Accept": "application/octet-stream",
                "User-Agent": "rustd-resolved-proof-downloader",
            },
        )
        with urllib.request.urlopen(signed_request, timeout=120) as signed_response:
            _copy_response(signed_response, output)
        return

    with response:
        _copy_response(response, output)


def arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--repository", default=os.environ.get("GITHUB_REPOSITORY"))
    parser.add_argument("--sha", default=os.environ.get("GITHUB_SHA"))
    parser.add_argument("--token", default=os.environ.get("GH_TOKEN"))
    parser.add_argument("--api-url", default=os.environ.get("GITHUB_API_URL", "https://api.github.com"))
    parser.add_argument(
        "--proof-directory", type=Path, default=Path("target/replacement-proofs")
    )
    parser.add_argument(
        "--root", type=Path, default=Path(__file__).resolve().parents[1]
    )
    return parser.parse_args()


def main() -> int:
    _self_test()
    options = arguments()
    if not options.repository or not options.sha or not options.token:
        raise DownloadError("repository, SHA, and token are required")
    validate_repository(options.repository)
    validate_sha(options.sha)
    api_url = options.api_url.rstrip("/")
    root = options.root.resolve()
    proof_directory = options.proof_directory
    if not proof_directory.is_absolute():
        proof_directory = root / proof_directory
    proof_directory = proof_directory.resolve()
    proof_directory.mkdir(parents=True, exist_ok=True)

    status_payload = api_request(
        f"{api_url}/repos/{options.repository}/commits/{options.sha}/status",
        options.token,
    )

    with tempfile.TemporaryDirectory(prefix="resolved-proof-download-") as temporary_name:
        temporary = Path(temporary_name)
        for gate, spec in ARTIFACTS.items():
            workflow_file = urllib.parse.quote(Path(spec.workflow_path).name, safe="")
            workflow = api_request(
                f"{api_url}/repos/{options.repository}/actions/workflows/{workflow_file}",
                options.token,
            )
            workflow_id = validate_workflow(spec, workflow)
            run_id, target_url = select_status_run(
                status_payload,
                options.repository,
                options.sha,
                spec.status_context,
            )
            run = api_request(
                f"{api_url}/repos/{options.repository}/actions/runs/{run_id}",
                options.token,
            )
            attempt, repository_id, head_repository_id, head_sha = validate_run(
                spec,
                options.repository,
                options.sha,
                workflow_id,
                run_id,
                target_url,
                run,
            )
            attempt_payload = api_request(
                f"{api_url}/repos/{options.repository}/actions/runs/{run_id}"
                f"/attempts/{attempt}",
                options.token,
            )
            attempt_started_at, attempt_updated_at = validate_attempt(
                spec,
                options.repository,
                options.sha,
                workflow_id,
                run_id,
                attempt,
                repository_id,
                head_repository_id,
                head_sha,
                attempt_payload,
            )
            name = spec.artifact_prefix + options.sha
            query = urllib.parse.urlencode({"name": name, "per_page": 100})
            # GitHub has no attempt-scoped artifact-list endpoint. Query the
            # exact run and let validate_run_artifact fail closed unless the
            # unique artifact was created inside the validated attempt window.
            run_artifacts = api_request(
                f"{api_url}/repos/{options.repository}/actions/runs/{run_id}"
                f"/artifacts?{query}",
                options.token,
            )
            artifact = validate_run_artifact(
                run_artifacts,
                name,
                run_id,
                options.repository,
                repository_id,
                head_repository_id,
                spec.dispatch_ref,
                head_sha,
                attempt_started_at,
                attempt_updated_at,
            )
            archive_url = artifact.get("archive_download_url")
            assert isinstance(archive_url, str)
            archive = temporary / f"{gate}.zip"
            download(archive_url, options.token, archive)
            subprocess.run(
                [
                    "/usr/bin/python3",
                    "-I",
                    str(root / "scripts" / "import-replacement-proof.py"),
                    str(archive),
                    "--root",
                    str(root),
                    "--proof-directory",
                    str(proof_directory),
                    "--source-commit",
                    options.sha,
                ],
                check=True,
            )
    print(f"Downloaded and validated all replacement proofs for {options.sha}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, DownloadError, subprocess.CalledProcessError) as error:
        print(f"download-replacement-proofs: {error}", file=sys.stderr)
        raise SystemExit(1) from error