Archived
85 lines
5.3 KiB
Python
85 lines
5.3 KiB
Python
"""Same-repository scoped postpublication callback; no publisher credential enters its job."""
|
|
|
|
import json
|
|
import urllib.request
|
|
from collections.abc import Callable
|
|
from http import HTTPStatus
|
|
from pathlib import Path
|
|
from typing import Any, cast, override
|
|
|
|
from platform_ci.delivery.enrollment import Enrollment
|
|
from platform_ci.delivery.errors import DeliveryError
|
|
from platform_ci.delivery.forge import ForgeReader, inspect_run, list_runs
|
|
from platform_ci.delivery.release_tag import TRUSTED_FORGE
|
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
@override
|
|
def redirect_request(self, *args: object, **kwargs: object) -> None:
|
|
return None
|
|
|
|
|
|
def callback_dispatcher(token: str, forge: ForgeReader) -> Callable[[Enrollment, str, str, Path, str], object]:
|
|
"""Use only the existing API credential. A missing ActionsWRITE is an explicit refusal, never a new grant."""
|
|
opener = urllib.request.build_opener(_NoRedirect())
|
|
|
|
def dispatch(enrollment: Enrollment, ref: str, sha: str, receipts: Path, build_revision: str) -> object:
|
|
if build_revision not in enrollment.platform_revisions:
|
|
raise DeliveryError("selected build helper is outside the operator-owned enrollment")
|
|
# Handoff is the successful release PUSH, not the reused PR bundle's producer run.
|
|
candidates = [run for run in list_runs(forge, enrollment.scope) if run.get("event") == "push" and run.get("head_sha") == sha and str(run.get("path", "")).partition("@")[2] == ref]
|
|
if not candidates:
|
|
raise DeliveryError("no original push at the exact release commit/ref exists for a source-version callback")
|
|
original = candidates[0]
|
|
run_id = original.get("id")
|
|
if type(run_id) is not int:
|
|
raise DeliveryError("original push has no numeric run identity")
|
|
facts = inspect_run(forge, enrollment.scope, run_id)
|
|
if facts.status != "completed" or facts.conclusion != "success" or facts.head_sha != sha:
|
|
raise DeliveryError("version callback requires the exact completed successful original push")
|
|
source_ids = {item.source_repository_id for item in enrollment.scoped_workflows}
|
|
if len(source_ids) != 1:
|
|
raise DeliveryError("version callback requires one existing canonical scoped source registration")
|
|
source_id = next(iter(source_ids))
|
|
marker = receipts / enrollment.repository / f"{sha}.{run_id}.{facts.attempt}.version-callback.json"
|
|
if marker.exists():
|
|
prior = json.loads(marker.read_text(encoding="utf-8"))
|
|
if prior.get("state") != "accepted" or prior.get("original_run") != run_id or prior.get("original_attempt") != facts.attempt or prior.get("source_sha") != sha or prior.get("ref") != ref or prior.get("build_revision") != build_revision:
|
|
raise DeliveryError("prior version callback dispatch is uncertain; reconcile its run before any retry")
|
|
return prior
|
|
endpoint = f"/repos/{enrollment.repository}/actions/workflows/finalize-version.yaml/dispatches?scoped_workflow_source_repo_id={source_id}&return_run_details=true"
|
|
document: dict[str, object] = {"original_run": run_id, "original_attempt": facts.attempt, "source_sha": sha, "ref": ref, "state": "dispatching", "build_revision": build_revision}
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
# A lost HTTP response cannot cause a second source-write callback on the next Cron invocation.
|
|
with marker.open("x", encoding="utf-8", newline="\n") as handle:
|
|
handle.write(json.dumps(document, sort_keys=True) + "\n")
|
|
request = urllib.request.Request( # noqa: S310 - fixed trusted HTTPS origin; redirects are disabled
|
|
TRUSTED_FORGE + "/api/v1" + endpoint, data=json.dumps({"ref": ref, "inputs": {"original_run": str(run_id), "original_attempt": facts.attempt, "build_revision": build_revision}}).encode(), headers={"Authorization": "token " + token, "Content-Type": "application/json"}, method="POST"
|
|
)
|
|
try:
|
|
with opener.open(request, timeout=30) as response:
|
|
if response.status != HTTPStatus.OK:
|
|
raise DeliveryError("scoped version callback did not return its run details")
|
|
result = json.loads(response.read(65537))
|
|
except (OSError, ValueError) as error:
|
|
raise DeliveryError("version callback dispatch failed or is uncertain; check existing ActionsWRITE and reconcile before retry") from error
|
|
if not isinstance(result, dict):
|
|
raise DeliveryError("version callback response has no run identity; reconcile before retry")
|
|
details = cast("dict[str, Any]", result)
|
|
if type(details.get("workflow_run_id")) is not int or details["workflow_run_id"] <= 0:
|
|
raise DeliveryError("version callback response has no run identity; reconcile before retry")
|
|
document |= {"state": "accepted", "callback_run": details["workflow_run_id"]}
|
|
marker.write_text(json.dumps(document, sort_keys=True) + "\n", encoding="utf-8", newline="\n")
|
|
return document
|
|
|
|
return dispatch
|
|
|
|
|
|
def callback_recorded(receipts: Path, repository: str, sha: str) -> bool:
|
|
"""Do not retire a completed tag's publication before its separate callback was actually accepted."""
|
|
for path in (receipts / repository).glob(f"{sha}.*.version-callback.json"):
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
|
if document.get("state") == "accepted" and document.get("source_sha") == sha:
|
|
return True
|
|
return False
|