diff --git a/actions/restore-release-artifact/action.yml b/actions/restore-release-artifact/action.yml new file mode 100644 index 0000000..9e98a1a --- /dev/null +++ b/actions/restore-release-artifact/action.yml @@ -0,0 +1,58 @@ +# SPDX-FileCopyrightText: 2026 LibreCode coop and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +name: Restore release artifact +description: Restore a deterministic release-state artifact from a previous workflow run. + +inputs: + repository: + description: Repository in owner/name form. + required: true + artifact-name: + description: Exact Actions artifact name. + required: true + expected-head-sha: + description: Optional originating workflow head SHA that the artifact must match. + required: false + default: '' + destination: + description: Destination directory. + required: true + github-token: + description: Token with Actions artifact read access. + required: true + +outputs: + artifact-id: + description: Restored Actions artifact id. + value: ${{ steps.restore.outputs.artifact-id }} + workflow-run-id: + description: Workflow run id that produced the artifact. + value: ${{ steps.restore.outputs.workflow-run-id }} + +runs: + using: composite + steps: + - id: restore + name: Restore exact release artifact + shell: bash + env: + GITHUB_TOKEN: ${{ inputs.github-token }} + RELEASE_REPOSITORY: ${{ inputs.repository }} + RELEASE_ARTIFACT_NAME: ${{ inputs.artifact-name }} + RELEASE_EXPECTED_HEAD_SHA: ${{ inputs.expected-head-sha }} + RELEASE_ARTIFACT_DESTINATION: ${{ inputs.destination }} + run: | + set -euo pipefail + result_file="${RUNNER_TEMP}/release-artifact-restore.json" + python3 "${GITHUB_ACTION_PATH}/../../scripts/restore_release_artifact.py" \ + --repository "${RELEASE_REPOSITORY}" \ + --name "${RELEASE_ARTIFACT_NAME}" \ + --expected-head-sha "${RELEASE_EXPECTED_HEAD_SHA}" \ + --destination "${RELEASE_ARTIFACT_DESTINATION}" \ + > "${result_file}" + + artifact_id="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["artifact_id"])' "${result_file}")" + workflow_run_id="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["workflow_run_id"])' "${result_file}")" + echo "artifact-id=${artifact_id}" >> "${GITHUB_OUTPUT}" + echo "workflow-run-id=${workflow_run_id}" >> "${GITHUB_OUTPUT}" diff --git a/scripts/restore_release_artifact.py b/scripts/restore_release_artifact.py new file mode 100755 index 0000000..58d79f0 --- /dev/null +++ b/scripts/restore_release_artifact.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2026 LibreCode coop and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +from __future__ import annotations + +import argparse +import io +import json +import os +from pathlib import Path +from urllib.parse import quote +from urllib.request import Request, urlopen +from zipfile import ZipFile + + +def select_artifact(payload: object, name: str, expected_head_sha: str | None) -> dict[str, object]: + if not isinstance(payload, dict) or not isinstance(payload.get("artifacts"), list): + raise RuntimeError("GitHub returned an invalid artifact listing") + + candidates: list[dict[str, object]] = [] + for item in payload["artifacts"]: + if not isinstance(item, dict): + continue + if item.get("name") != name or item.get("expired") is True: + continue + workflow_run = item.get("workflow_run") + if expected_head_sha: + if not isinstance(workflow_run, dict) or workflow_run.get("head_sha") != expected_head_sha: + continue + candidates.append(item) + + if not candidates: + suffix = f" for head {expected_head_sha}" if expected_head_sha else "" + raise RuntimeError(f"Actions artifact {name!r}{suffix} was not found") + + candidates.sort( + key=lambda item: (str(item.get("created_at", "")), int(item.get("id", 0))), + reverse=True, + ) + return candidates[0] + + +def safe_extract_zip(data: bytes, destination: Path) -> None: + destination.mkdir(parents=True, exist_ok=True) + root = destination.resolve() + + with ZipFile(io.BytesIO(data)) as archive: + for entry in archive.infolist(): + relative = Path(entry.filename) + if relative.is_absolute() or ".." in relative.parts: + raise RuntimeError(f"unsafe artifact path: {entry.filename}") + target = (destination / relative).resolve() + try: + target.relative_to(root) + except ValueError as error: + raise RuntimeError(f"unsafe artifact path: {entry.filename}") from error + + archive.extractall(destination) + + +def request_json(url: str, token: str) -> object: + request = Request(url, headers={ + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {token}", + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "LibreCodeCoop/github-workflows", + }) + with urlopen(request, timeout=30) as response: + return json.load(response) + + +def request_bytes(url: str, token: str) -> bytes: + request = Request(url, headers={ + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {token}", + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "LibreCodeCoop/github-workflows", + }) + with urlopen(request, timeout=60) as response: + return response.read() + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--repository", required=True) + parser.add_argument("--name", required=True) + parser.add_argument("--expected-head-sha", default="") + parser.add_argument("--destination", required=True, type=Path) + parser.add_argument("--api-url", default=os.environ.get("GITHUB_API_URL", "https://api.github.com")) + args = parser.parse_args() + + token = os.environ.get("GITHUB_TOKEN", "") + if token.strip() == "": + parser.error("GITHUB_TOKEN must not be empty") + + listing_url = ( + f"{args.api_url.rstrip('/')}/repos/{args.repository}/actions/artifacts" + f"?name={quote(args.name, safe='')}&per_page=100" + ) + artifact = select_artifact( + request_json(listing_url, token), + args.name, + args.expected_head_sha or None, + ) + archive_url = artifact.get("archive_download_url") + if not isinstance(archive_url, str) or archive_url == "": + parser.error("GitHub returned an artifact without archive_download_url") + + safe_extract_zip(request_bytes(archive_url, token), args.destination) + + workflow_run = artifact.get("workflow_run") + run_id = workflow_run.get("id") if isinstance(workflow_run, dict) else None + print(json.dumps({ + "artifact_id": artifact.get("id"), + "workflow_run_id": run_id, + "name": args.name, + "created_at": artifact.get("created_at"), + }, separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_restore_release_artifact.py b/tests/test_restore_release_artifact.py new file mode 100644 index 0000000..4f7363c --- /dev/null +++ b/tests/test_restore_release_artifact.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: 2026 LibreCode coop and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +from __future__ import annotations + +import importlib.util +import io +import tempfile +import unittest +from pathlib import Path +from zipfile import ZipFile + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "restore_release_artifact.py" + +spec = importlib.util.spec_from_file_location("restore_release_artifact", SCRIPT) +assert spec is not None and spec.loader is not None +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) + + +class RestoreReleaseArtifactTest(unittest.TestCase): + def test_selects_latest_non_expired_artifact_for_expected_head(self) -> None: + payload = {"artifacts": [ + {"id": 1, "name": "release-preparation-pr-10", "expired": False, "created_at": "2026-01-01T00:00:00Z", "workflow_run": {"id": 11, "head_sha": "a" * 40}}, + {"id": 3, "name": "release-preparation-pr-10", "expired": False, "created_at": "2026-01-03T00:00:00Z", "workflow_run": {"id": 13, "head_sha": "b" * 40}}, + {"id": 2, "name": "release-preparation-pr-10", "expired": False, "created_at": "2026-01-02T00:00:00Z", "workflow_run": {"id": 12, "head_sha": "a" * 40}}, + ]} + selected = module.select_artifact(payload, "release-preparation-pr-10", "a" * 40) + self.assertEqual(2, selected["id"]) + + def test_ignores_expired_artifacts(self) -> None: + payload = {"artifacts": [{"id": 1, "name": "state", "expired": True, "created_at": "2026-01-01T00:00:00Z", "workflow_run": {"head_sha": "a" * 40}}]} + with self.assertRaisesRegex(RuntimeError, "was not found"): + module.select_artifact(payload, "state", "a" * 40) + + def test_rejects_path_traversal_archive(self) -> None: + buffer = io.BytesIO() + with ZipFile(buffer, "w") as archive: + archive.writestr("../escape.json", "{}") + with tempfile.TemporaryDirectory() as directory: + with self.assertRaisesRegex(RuntimeError, "unsafe artifact path"): + module.safe_extract_zip(buffer.getvalue(), Path(directory)) + + def test_extracts_safe_archive(self) -> None: + buffer = io.BytesIO() + with ZipFile(buffer, "w") as archive: + archive.writestr("release-plan.json", "{}") + with tempfile.TemporaryDirectory() as directory: + destination = Path(directory) + module.safe_extract_zip(buffer.getvalue(), destination) + self.assertEqual("{}", (destination / "release-plan.json").read_text(encoding="utf-8")) + + +if __name__ == "__main__": + unittest.main()