platform-ci/platform_ci/delivery/source_lifecycle.py

"""The source-lifecycle write phase after a verified, published release: persist the build-effective version, from the captured `qrun-source-version.mjs`.

UNTESTED: written while test execution was suspended; nothing here has run, signed, pushed or read a credential. Reached only by the `finalize-version` command (`delivery/finalize_version.py`),
which no workflow job calls yet (see "Activation prerequisites" in `IMPLEMENTATION.md`).

What it ports, and from where (`references/release-lifecycle/b06800a6…/scripts/qrun-source-version.mjs`, receipt `references/source-version-helper-receipt.json`; the `revisionOnlyChange` rule is
from the captured `qrun-maven-prepare.mjs`): the version-plan checks, "no source write until the whole reactor is read back byte-identical from the private registry and the Central completion
is terminal", and `persistVersion`: a staging branch at the exact source commit, one server-signed `PUT` of the one file, strict verification of that commit (one file, one parent,
server-verified, exact message, exact bytes), a normal NON-forced fast-forward push through bare Git, an acknowledged head, and deletion of only the exact heads this run created. A push
failure is a warning unless the channel requires the push.

What changed, and only what had to: the reactor, the publication and the Central completion are read from the VERIFIED BUNDLE and the terminal records the candidate's publisher already writes
(`maven_reactor`, `completion`, `completion_receipt`), not from a legacy Actions artifact and a poll; the plan is derived from repository configuration (`version_write_back`) and those facts, not
from a prepare-script artifact. Source-writing code is this pinned platform module only: no application script ever runs with the write credential. The credential is the job's own ephemeral
token, given to Git only through the child environment (`SubprocessTagGit`, the existing tag boundary's class) and to the forge only by `SourceWriter`, which confines every request to the
enrolled repository, the one planned file and the one staging branch.
"""

import base64
import hashlib
import json
import re
import tempfile
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Any, Final, cast

from platform_ci.configuration import WriteBackPolicy
from platform_ci.delivery import completion_receipt, legacy
from platform_ci.delivery.completion import reactor_for, verified_modules
from platform_ci.delivery.errors import DeliveryError
from platform_ci.delivery.manifest import ManifestFile
from platform_ci.delivery.release_tag import TRUSTED_FORGE
from platform_ci.delivery.reuse import ReleaseArtifact
from platform_ci.delivery.signed_tag import SignedTagError, SubprocessTagGit

_HTTP_OK: Final = 200
_HTTP_MULTIPLE_CHOICES: Final = 300
_HTTP_NOT_FOUND: Final = 404
MAX_POM_BYTES: Final = 512 * 1024
SUPPORTED_FILE: Final = "pom.xml"
"""The captured helper rewrites only the root POM's `<properties><revision>`; no other file or property is supported."""
_REVISION_VALUE: Final = re.compile(r"^[0-9A-Za-z][0-9A-Za-z._-]*$")
_SHA40: Final = re.compile(r"^[0-9a-f]{40}$")
_TOKENS: Final = re.compile(r"<(/?)([A-Za-z][\w.-]*)[^>]*?(/?)>|([^<]+)")


class SourceLifecycleError(DeliveryError):
   """A source-lifecycle refusal."""


class Fatal(SourceLifecycleError):  # noqa: N818 - the established name of the captured postpublish contract, imported by name
   """A condition that makes the publication or the plan untrustworthy: never downgraded to a warning."""


class PushFailure(SourceLifecycleError):  # noqa: N818 - the established name of the captured postpublish contract, imported by name
   """The version could not be pushed (race, protection, forge): a warning, except where the channel requires the push."""


class NotYetPublished(SourceLifecycleError):  # noqa: N818 - the established name of the captured postpublish contract, imported by name
   """The registry files or the Central completion are not (all) there yet: nothing is written and the next invocation tries again."""


# --------------------------------------------------------------------------------------------------- the revision rule


def _revision_paths(text: str) -> list[str]:
   """Return the element path of every `<revision>` from a comment, PI and CDATA free token scan (DOCTYPE and ENTITY refused)."""
   if re.search(r"<!\s*(DOCTYPE|ENTITY)", text, re.IGNORECASE):
      raise SourceLifecycleError("DTD/entity declarations are not allowed")
   body = re.sub(r"<!\[CDATA\[[\s\S]*?\]\]>", "", re.sub(r"<\?[\s\S]*?\?>", "", re.sub(r"<!--[\s\S]*?-->", "", text)))
   stack: list[str] = []
   found: list[str] = []
   for match in _TOKENS.finditer(body):
      closing, name, self_closing, text_node = match.groups()
      if text_node is not None:
         continue
      if closing:
         if stack:
            stack.pop()
      else:
         if not self_closing:
            stack.append(name)
         if name == "revision":
            found.append("/".join(stack))
   return found


def revision_only_change(old: str, new: str) -> tuple[str, str]:
   """Return (declared, effective) when the two root POMs differ ONLY in the one root `properties/revision` value, exactly as the captured rule; anything else raises."""

   def locate(text: str) -> re.Match[str]:
      matches = list(re.finditer(r"<revision>([^<]*)</revision>", text))
      paths = _revision_paths(text)
      if len(matches) != 1 or len(re.findall(r"<revision[\s>/]", text)) != 1 or paths != ["project/properties/revision"]:
         raise SourceLifecycleError("Expected exactly one root properties revision")
      return matches[0]

   before, after = locate(old), locate(new)
   if not _REVISION_VALUE.match(after.group(1)):
      raise SourceLifecycleError("Unsupported computed revision value")
   if new != old[: before.start()] + after.group(0) + old[before.end() :]:
      raise SourceLifecycleError("The version rewrite changed more than the root revision value")
   return before.group(1), after.group(1)


def rewrite_revision(old: str, effective: str) -> str:
   """Return `old` with ONLY the one root `properties/revision` value replaced by `effective`; the result is re-checked by `revision_only_change`, so nothing else can differ."""
   if not _REVISION_VALUE.match(effective):
      raise SourceLifecycleError("Unsupported computed revision value")
   matches = list(re.finditer(r"<revision>([^<]*)</revision>", old))
   if len(matches) != 1:
      raise SourceLifecycleError("Expected exactly one root properties revision")
   start, end = matches[0].span(1)
   new = old[:start] + effective + old[end:]
   revision_only_change(old, new)
   return new


def git_blob_sha(data: bytes) -> str:
   """Return the Git blob id of the bytes (what the forge's `contents` API reports as `sha`)."""
   return hashlib.sha1(b"blob " + str(len(data)).encode() + b"\0" + data).hexdigest()  # noqa: S324 - a Git object id, not a security digest


# ---------------------------------------------------------------------------------------------------- the plan


@dataclass(frozen=True)
class WriteContext:
   """The authoritative facts of the write run: the repository, the verified source commit, the branch the bump goes to, an optional run tag, and the run identity."""

   repository: str
   sha: str
   branch: str
   run: int
   attempt: int
   tag: str = ""


@dataclass(frozen=True)
class VersionPlan:
   """One planned bump: the exact old and new bytes of the one file, bound to the source commit and blob, changing only the revision to the verified version."""

   path: str
   source_commit: str
   source_blob: str
   old: bytes
   new: bytes
   declared: str
   effective: str
   require_push: bool


def build_plan(ctx: WriteContext, policy: WriteBackPolicy, *, old: bytes, new: bytes, verified_version: str) -> VersionPlan | None:
   """Derive the plan from the source POM and the rewritten POM; None when the version is unchanged (nothing to persist, which is not an error).

   The effective version must be exactly the version the verified bundle recorded; a different one is a relabelling and is refused.
   """
   if len(old) > MAX_POM_BYTES or len(new) > MAX_POM_BYTES:
      raise Fatal("the POM exceeds the plan bound")
   declared, effective = revision_only_change(old.decode("utf-8"), new.decode("utf-8"))
   if effective != verified_version:
      raise Fatal("the planned revision is not the version the verified bundle recorded")
   if old == new:
      return None
   if ctx.tag and not policy.target_branch:
      raise Fatal("a tag run persists a version only to a branch the channel's configuration names")
   if not _SHA40.fullmatch(ctx.sha):
      raise Fatal("the source commit must be a full commit id")
   return VersionPlan(SUPPORTED_FILE, ctx.sha, git_blob_sha(old), old, new, declared, effective, policy.require_push)


# ---------------------------------------------------------------------------------------------------- the confined write client

Send = Callable[[str, str, Mapping[str, str], bytes | None], tuple[int, bytes]]


def _urllib_send(method: str, url: str, headers: Mapping[str, str], data: bytes | None) -> tuple[int, bytes]:
   """One HTTPS request that never follows a redirect; an HTTP error status is returned, not raised, as the captured helper's `call` did."""

   class NoRedirect(urllib.request.HTTPRedirectHandler):
      def redirect_request(self, *args: object, **kwargs: object) -> None:  # type: ignore[override]
         raise SourceLifecycleError("unexpected redirect")

   request = urllib.request.Request(url, method=method, data=data, headers=dict(headers))  # noqa: S310 - the URL is built from the trusted origin by SourceWriter only
   try:
      with urllib.request.build_opener(NoRedirect()).open(request, timeout=30) as response:
         return response.status, response.read(8 * 1024 * 1024)
   except urllib.error.HTTPError as error:
      status = error.code
      error.close()
      return status, b""
   except (urllib.error.URLError, TimeoutError, OSError):
      return 0, b""


class SourceWriter:
   """The one forge client of the write phase: `call(method, path, body=None) -> (status, json)`, confined to what `persist_version` needs.

   Allowed: GET the repository, one branch, one commit, one Git commit, one run, one tag, and the planned file at a commit id; POST exactly the staging branch; PUT exactly the planned
   file on exactly the staging branch without force; DELETE exactly the staging branch. Everything else is refused before any request, whatever the token could do.
   """

   def __init__(self, token: str, ctx: WriteContext, plan_path: str = SUPPORTED_FILE, send: Send = _urllib_send) -> None:
      """Bind the token, the repository and the run's staging branch name."""
      if not token or re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", ctx.repository) is None:
         raise SourceLifecycleError("the job token and an owner/name repository are required")
      self._token, self._send = token, send
      self.repository, self.path = ctx.repository, plan_path
      self.stage = f"ci/version-stage-{ctx.run}-{ctx.attempt}"
      base = f"/repos/{ctx.repository}"
      self._get = re.compile(re.escape(base) + r"(?:|/branches/[^/?#]+|/commits/[0-9a-f]{40}|/git/commits/[0-9a-f]{40}|/actions/runs/[1-9][0-9]*|/tags/[A-Za-z0-9][A-Za-z0-9._-]*|/contents/" + re.escape(plan_path) + r"\?ref=[0-9a-f]{40})")
      self._exact = {("POST", f"{base}/branches"), ("PUT", f"{base}/contents/{plan_path}"), ("DELETE", f"{base}/branches/{urllib.parse.quote(self.stage, safe='')}")}

   def __call__(self, method: str, path: str, body: Mapping[str, Any] | None = None) -> tuple[int, Any]:
      """Send one confined request and return (status, decoded JSON or None)."""
      if not ((method == "GET" and self._get.fullmatch(path) and body is None) or ((method, path) in self._exact and self._body_ok(method, body))):
         raise SourceLifecycleError("refusing a forge request outside the planned version write")
      headers = {"Authorization": f"token {self._token}", "Accept": "application/json", "Content-Type": "application/json"}
      status, raw = self._send(method, TRUSTED_FORGE + "/api/v1" + path, headers, None if body is None else json.dumps(body).encode())
      try:
         return status, json.loads(raw) if raw else None
      except ValueError:
         return status, None

   def _body_ok(self, method: str, body: Mapping[str, Any] | None) -> bool:
      """Return whether the body is one the staging write may carry: the staging branch at the source commit, or the planned file on the staging branch without force."""
      if method == "DELETE":
         return body is None
      if method == "POST":
         return isinstance(body, dict) and body.get("new_branch_name") == self.stage and set(body) == {"new_branch_name", "old_ref_name"}
      return isinstance(body, dict) and body.get("branch") == self.stage and body.get("force_push") is False and set(body) == {"branch", "sha", "content", "message", "force_push"}


class VersionGit(SubprocessTagGit):
   """Bare Git with the token only in the child environment: fetch the verified staging commit, verify it again, push it as a normal fast-forward."""

   def push_verified(self, stage: str, child: str, source: str, branch: str) -> None:
      """Fetch the staging branch, require it to be exactly `child` with `source` as its sole parent, and push `child` to `branch` (never forced, hooks off)."""
      with tempfile.TemporaryDirectory() as work:
         try:
            self._git(work, "init", "--bare", "--template=", "-q")
            self._git(work, "fetch", "--no-tags", self.url, f"refs/heads/{stage}:refs/cas/stage")
         except SignedTagError as error:
            raise PushFailure("git could not fetch the staging commit") from error
         try:
            fetched = self._git(work, "rev-parse", "--verify", "refs/cas/stage^{commit}").decode().strip()
            lineage = self._git(work, "rev-list", "--parents", "-n", "1", child).decode().strip()
         except SignedTagError as error:
            raise Fatal("the fetched object cannot be verified") from error
         if fetched != child:
            raise Fatal("the fetched object is not the verified staged commit")
         if lineage != f"{child} {source}":
            raise Fatal("the fetched commit parent is not the source commit")
         try:
            self._git(work, "push", "--no-verify", self.url, f"{child}:refs/heads/{branch}")
         except SignedTagError as error:
            raise PushFailure("the version push was rejected") from error


# --------------------------------------------------------------------------------------------------------- the write itself

Writer = Callable[..., tuple[int, Any]]


def _good(result: tuple[int, Any], what: str) -> object:
   """Return the JSON of a 2xx answer; any other status is a push failure (a race, protection or a forge hiccup)."""
   status, body = result
   if not _HTTP_OK <= status < _HTTP_MULTIPLE_CHOICES:
      raise PushFailure(f"{what} status {status}")
   return cast("object", body)


def _fields(value: object) -> dict[str, Any]:
   """Return a JSON object's fields; any other value (absent, null, a list) reads as an empty object, as the captured `x || {}` did."""
   return cast("dict[str, Any]", value) if isinstance(value, dict) else {}


def _items(value: object) -> list[dict[str, Any]]:
   """Return the JSON objects of a JSON array; anything else reads as no items."""
   return [_fields(item) for item in cast("list[object]", value)] if isinstance(value, list) else []


def verify_bump_commit(ctx: WriteContext, plan: VersionPlan, child: str, writer: Writer) -> None:
   """Re-verify, from the forge alone, that `child` is exactly the planned bump: one file, the source commit as sole parent, server-verified, exact message and bytes."""
   prefix = f"/repos/{ctx.repository}"
   commit = _fields(_good(writer("GET", f"{prefix}/commits/{child}"), "commit"))
   if [item.get("filename") for item in _items(commit.get("files"))] != [plan.path]:
      raise Fatal(f"the staged commit does not change exactly {plan.path}")
   if [item.get("sha") for item in _items(commit.get("parents"))] != [plan.source_commit]:
      raise Fatal("the staged commit parent is not the source commit")
   git_commit = _fields(_good(writer("GET", f"{prefix}/git/commits/{child}"), "git commit"))
   embedded = _fields(git_commit.get("commit"))
   verification = _fields(embedded.get("verification") or git_commit.get("verification"))
   if verification.get("verified") is not True:
      raise Fatal("the staged commit is not server-verified")
   message = str(embedded.get("message") or git_commit.get("message") or "").strip()
   if message != f"Bump version to {plan.effective} [skip ci]":
      raise Fatal("the staged commit message differs")
   staged = _fields(_good(writer("GET", f"{prefix}/contents/{plan.path}?ref={child}"), "contents"))
   try:
      staged_bytes = base64.b64decode(staged.get("content", ""), validate=True)
   except ValueError as error:
      raise Fatal("the staged file is not valid base64") from error
   if staged_bytes != plan.new:
      raise Fatal("the staged bytes differ from the plan")


def persist_version(ctx: WriteContext, plan: VersionPlan, writer: Writer, git: VersionGit) -> str:
   """Persist the bump as the captured helper did, normal and non-forced, and return the pushed commit id (a `PushFailure` leaves the branch untouched)."""
   prefix = f"/repos/{ctx.repository}"
   stage = f"ci/version-stage-{ctx.run}-{ctx.attempt}"
   owned: set[str] = set()

   def head(branch: str) -> str | None:
      body = _fields(_good(writer("GET", f"{prefix}/branches/{urllib.parse.quote(branch, safe='')}"), f"branch {branch}"))
      return cast("str | None", _fields(body.get("commit")).get("id"))

   try:
      if ctx.tag:
         status, found = writer("GET", f"{prefix}/tags/{ctx.tag}")
         if status != _HTTP_OK or _fields(_fields(found).get("commit")).get("sha") != ctx.sha:
            raise Fatal("the run tag no longer points at the source commit")
      if head(ctx.branch) != ctx.sha:
         raise PushFailure("the canonical branch moved since the source commit")
      _good(writer("POST", f"{prefix}/branches", {"new_branch_name": stage, "old_ref_name": ctx.sha}), "create staging branch")
      owned.add(ctx.sha)
      if head(stage) != ctx.sha:
         raise Fatal("the staging branch is not at the source commit")
      written = _good(
         writer("PUT", f"{prefix}/contents/{plan.path}", {"branch": stage, "sha": plan.source_blob, "content": base64.b64encode(plan.new).decode(), "message": f"Bump version to {plan.effective} [skip ci]", "force_push": False}),
         "write staged file",
      )
      child = _fields(_fields(written).get("commit")).get("sha")
      if not isinstance(child, str) or not _SHA40.fullmatch(child) or child == ctx.sha:
         raise Fatal("bad staged commit sha")
      owned.add(child)
      verify_bump_commit(ctx, plan, child, writer)
      git.push_verified(stage, child, ctx.sha, ctx.branch)
      if head(ctx.branch) != child:
         raise PushFailure("the version push was not acknowledged")
      return child
   finally:
      try:
         status, current = writer("GET", f"{prefix}/branches/{urllib.parse.quote(stage, safe='')}")
         if status == _HTTP_OK and _fields(_fields(current).get("commit")).get("id") in owned:
            writer("DELETE", f"{prefix}/branches/{urllib.parse.quote(stage, safe='')}")
      except SourceLifecycleError:
         pass  # cleanup is best effort and exact: only a staging head this run created is ever deleted


# ------------------------------------------------------------------------------------------ evidence read from the verified bundle


def verify_publication(artifact: ReleaseArtifact, files: tuple[ManifestFile, ...], version: str, owner: str, reader: Callable[..., tuple[int, bytes]]) -> None:
   """Read every published Maven file back from the private registry and require the retained bytes (the captured `waitForPublication`, without the poll).

   A file that is not there yet is `NotYetPublished`; a file with other bytes is `Fatal`. The paths are the frozen publisher's own planner's, so they are exactly what it wrote.
   """
   modules = verified_modules(artifact, files, version)
   evidence = {"version": version, "modules": [{key: value for key, value in module.items() if key != "version"} for module in modules]}
   snapshot = legacy.snapshot_stamp(artifact.run.created_at, int(artifact.run.run_id)) if version.endswith("-SNAPSHOT") else None
   with legacy.translate_refusals():
      plan = legacy.qrun_maven_publication.plan_files(evidence, owner, snapshot)
   for item in plan:
      try:
         status, body = reader("GET", item["path"])
      except urllib.error.HTTPError as error:
         error.close()
         if error.code == _HTTP_NOT_FOUND:
            raise NotYetPublished(f"{item['path']} is not in the registry yet") from error
         raise
      if status != _HTTP_OK or hashlib.sha256(body).digest() != hashlib.sha256(item["data"]).digest():
         raise Fatal("a published Maven file differs from the verified bytes")


def verify_completion(artifact: ReleaseArtifact, files: tuple[ManifestFile, ...], version: str, *, repository: str, repository_id: int, owner: str, reader: Callable[..., tuple[int, bytes]]) -> dict[str, Any]:  # noqa: PLR0913 - one verified reactor and its identity
   """Read the immutable Central completion receipt and require it to describe exactly this tested reactor and to be terminal (the captured `waitForCentralCompletion`, without the poll).

   Absent is `NotYetPublished`; present but different is `Fatal`. The receipt module validates the terminal-state rules itself.
   """
   reactor = reactor_for(artifact, files, version)
   modules = verified_modules(artifact, files, version)
   target = "central-snapshots" if version.endswith("-SNAPSHOT") else "central-publisher"
   run_id = int(artifact.run.run_id)
   expected: dict[str, Any] = {
      "repository": repository,
      "repository_id": repository_id,
      "run_id": run_id,
      "source_commit": artifact.build_sha,
      "version": version,
      "archive_sha256": reactor.archive_sha256,
      "manifest_sha256": reactor.manifest_sha256,
      "gav": sorted(f"{module['groupId']}:{module['artifactId']}:{version}" for module in modules),
      "target": target,
   }
   try:
      return completion_receipt.read(owner, reader, repository_id, run_id, artifact.build_sha, expected)
   except urllib.error.HTTPError as error:
      error.close()
      if error.code == _HTTP_NOT_FOUND:
         raise NotYetPublished("the Central completion receipt does not exist yet") from error
      raise
   except ValueError as error:
      raise Fatal(f"the Central completion receipt does not match this tested reactor: {error}") from error


def postpublish(  # noqa: PLR0913 - keyword-only evidence of one release; the call site names every field
   ctx: WriteContext,
   policy: WriteBackPolicy,
   *,
   artifact: ReleaseArtifact,
   files: tuple[ManifestFile, ...],
   version: str,
   owner: str,
   repository_id: int,
   central_required: bool,
   writer: Writer,
   registry_reader: Callable[..., tuple[int, bytes]],
   completion_reader: Callable[..., tuple[int, bytes]],
   git: VersionGit,
   extra_evidence: tuple[tuple[ReleaseArtifact, tuple[ManifestFile, ...]], ...] = (),
) -> dict[str, str]:
   """Run the captured `postpublish` order over verified evidence: source POM, whole-reactor readback, terminal Central completion, then (only then) the version write.

   `version` is the effective version the verified bundle recorded. Nothing is written until the registry and (where required) Central evidence are complete; an absent one raises
   `NotYetPublished` so the next invocation retries. An unchanged version ends honestly as `skipped`. A push failure is a warning unless the policy requires the push. `extra_evidence`
   lists further (artifact, files) pairs of the same release whose registry and completion evidence must be complete too before the one POM is written.
   """
   status, answer = writer("GET", f"/repos/{ctx.repository}/contents/{policy.file}?ref={ctx.sha}")
   if status != _HTTP_OK or not isinstance(answer, dict):
      raise Fatal("the source POM cannot be read at the verified commit")
   document = cast("dict[str, Any]", answer)  # a JSON object, checked just above
   try:
      old = base64.b64decode(document.get("content", ""), validate=True)
   except ValueError as error:
      raise Fatal("the source POM is not valid base64") from error
   if document.get("sha") != git_blob_sha(old):
      raise Fatal("the source POM blob id differs from its bytes")
   plan = build_plan(ctx, policy, old=old, new=rewrite_revision(old.decode("utf-8"), version).encode("utf-8"), verified_version=version)
   for evidenced, evidenced_files in ((artifact, files), *extra_evidence):
      verify_publication(evidenced, evidenced_files, version, owner, registry_reader)
      if central_required:
         verify_completion(evidenced, evidenced_files, version, repository=ctx.repository, repository_id=repository_id, owner=owner, reader=completion_reader)
   if plan is None:
      return {"status": "skipped", "reason": "the version did not change", "pushed_sha": ""}
   try:
      return {"status": "pushed", "reason": "", "pushed_sha": persist_version(ctx, plan, writer, git)}
   except PushFailure as error:
      if plan.require_push:
         raise Fatal(f"the required version push was not acknowledged ({error})") from error
      return {"status": "push-failed-warning", "reason": str(error), "pushed_sha": ""}