from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request
class DownloadError(RuntimeError):
pass
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): return None
ARTIFACTS = {
"upstream-test-75": (
"replacement-upstream-test-75-proof-",
"replacement/upstream-test-75",
),
"upstream-test-89-mdns": (
"replacement-upstream-test-89-mdns-proof-",
"replacement/upstream-test-89-mdns",
),
"security-suite": (
"replacement-security-proof-",
"replacement/security-proof",
),
"boot-replacement": (
"replacement-boot-proof-",
"replacement/boot-proof",
),
}
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: 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:
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": "systemd-resolved-rs-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:
options = arguments()
if not options.repository or not options.sha or not options.token:
raise DownloadError("repository, SHA, and token are required")
if len(options.sha) != 40:
raise DownloadError("SHA must be a full Git commit identifier")
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"{options.api_url}/repos/{options.repository}/commits/{options.sha}/status",
options.token,
)
latest_status: dict[str, dict[str, object]] = {}
for item in status_payload.get("statuses", []):
if not isinstance(item, dict):
continue
context = item.get("context")
if isinstance(context, str) and context not in latest_status:
latest_status[context] = item
with tempfile.TemporaryDirectory(prefix="resolved-proof-download-") as temporary_name:
temporary = Path(temporary_name)
for gate, (prefix, status_context) in ARTIFACTS.items():
name = prefix + options.sha
query = urllib.parse.urlencode({"name": name, "per_page": 100})
payload = api_request(
f"{options.api_url}/repos/{options.repository}/actions/artifacts?{query}",
options.token,
)
artifacts = [
artifact
for artifact in payload.get("artifacts", [])
if isinstance(artifact, dict)
and artifact.get("name") == name
and artifact.get("expired") is False
]
if not artifacts:
raise DownloadError(f"no unexpired proof artifact exists for {gate} at {options.sha}")
artifacts.sort(key=lambda value: str(value.get("created_at", "")), reverse=True)
artifact = artifacts[0]
workflow_run = artifact.get("workflow_run")
run_id = workflow_run.get("id") if isinstance(workflow_run, dict) else None
if not isinstance(run_id, int):
raise DownloadError(f"artifact has no workflow run for {gate}")
status = latest_status.get(status_context)
if not status or status.get("state") != "success":
raise DownloadError(
f"exact-source status {status_context} is not successful for {gate}"
)
target_url = status.get("target_url")
if not isinstance(target_url, str) or not target_url.endswith(
f"/actions/runs/{run_id}"
):
raise DownloadError(f"artifact run is not bound by {status_context}")
run = api_request(
f"{options.api_url}/repos/{options.repository}/actions/runs/{run_id}",
options.token,
)
if run.get("event") != "workflow_dispatch" or run.get("conclusion") != "success":
raise DownloadError(f"artifact workflow did not pass for {gate}")
archive_url = artifact.get("archive_download_url")
if not isinstance(archive_url, str):
raise DownloadError(f"artifact has no download URL for {gate}")
archive = temporary / f"{gate}.zip"
download(archive_url, options.token, archive)
subprocess.run(
[
sys.executable,
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