platform-ci/platform_ci/delivery/gitops.py

"""Pin published container DIGESTS in one enrolled GitOps kustomization with one compare-and-swap commit.

This reuses the frozen GitOps reader and its safety rules (enrolled repository, protected branch, signature-verified
commits, exactly one touched file) but pins an immutable digest instead of a tag. Only the `digest` line of an existing
`images[]` entry whose `newName` already names the enrolled registry image is written. The target comes from an operator
binding, never from an artifact or run. A pin is a commit, not a rollout: the receipt always says `deployment_verified` is
false.
"""

import base64
import hashlib
import re
import urllib.error
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, cast

from platform_ci.delivery import legacy
from platform_ci.delivery.authorization import GitOpsTarget
from platform_ci.delivery.errors import PublicationError

_COMMIT_ID = re.compile(r"[0-9a-f]{40}")
_DEFAULT_ATTEMPTS = 3


@dataclass(frozen=True)
class ImagePin:
   """One image digest, already published and read back, to pin under its kustomize entry name."""

   kustomize_name: str
   digest: str


@dataclass(frozen=True)
class ReleaseIdentity:
   """The build that produced the bytes, the release being delivered, and the source run: three separate facts."""

   build_sha: str
   release_sha: str
   run_id: str


@dataclass(frozen=True)
class GitOpsClients:
   """Injected, already-scoped clients: the one-file GitOps client and a read-only registry client per enrolled image."""

   gitops: legacy.GitOpsCall
   registry_reader: Mapping[str, legacy.RegistryCall]


@dataclass(frozen=True)
class GitOpsReceipt:
   """The outcome of one pin commit. `deployment_verified` is never true: observing a rollout is a separate step."""

   repository: str
   branch: str
   path: str
   commit: str
   status: str
   pins: tuple[tuple[str, str], ...]
   build_sha: str
   release_sha: str
   deployment_verified: bool = False


def set_image_digests(text: str, updates: Mapping[str, tuple[str, str]]) -> str:
   """Set or add the `digest` of each enrolled entry and return the new text; all other bytes are kept.

   `updates` maps a kustomize entry name to (the registry name its newName must already be, the new digest).
   Anything the narrow parser does not recognise is refused.
   """
   if "\r" in text:
      raise PublicationError("unsupported line endings in the GitOps file")
   lines = text.split("\n")
   bounds = legacy.image_entry_bounds(lines)
   operations = [_plan_digest_line(lines, bounds, name, new_name, digest) for name, (new_name, digest) in updates.items()]
   for index, replace_existing, line in sorted(operations, reverse=True):
      if replace_existing:
         lines[index] = line
      else:
         lines.insert(index, line)
   return "\n".join(lines)


def pin_image_digests(target: GitOpsTarget, pins: tuple[ImagePin, ...], release: ReleaseIdentity, clients: GitOpsClients, attempts: int = _DEFAULT_ATTEMPTS) -> GitOpsReceipt:
   """Verify each digest is served by content, then pin them all in one signed commit, or write nothing."""
   enrolled = _verify_pins_are_served(target, pins, clients)
   updates = {pin.kustomize_name: (legacy.REGISTRY_PREFIX + enrolled[pin.kustomize_name], pin.digest) for pin in pins}
   policy: dict[str, Any] = {"repository": target.repository, "repository_id": target.repository_id, "path": target.path}
   gitops = clients.gitops
   base = "/repos/" + target.repository
   previous: str | None = None
   for _ in range(attempts):
      file_sha, text, head_id = legacy.read_gitops_file(gitops, target.branch, policy)
      if file_sha == previous:
         raise PublicationError("GitOps update was refused without a competing change")
      updated = set_image_digests(text, updates)
      if updated == text:
         return _receipt(target, pins, _current_verified_commit(target, gitops), "already-current", release)
      if not _COMMIT_ID.fullmatch(head_id):
         raise PublicationError("forge did not report the GitOps branch head")
      _, head_commit = gitops("GET", f"{base}/git/commits/{head_id}")
      legacy.require_verified_commit(head_commit, "GitOps branch head (server-side signing is not qualified)")
      images = "\n".join(f"Image: {enrolled[pin.kustomize_name]} {pin.digest}" for pin in pins)
      message = f"deploy: pin image digests\n\nBuild: {release.build_sha}\nRelease: {release.release_sha} run {release.run_id}\n{images}"
      body = {"content": base64.b64encode(updated.encode("utf-8")).decode(), "sha": file_sha, "branch": target.branch, "message": message}
      try:
         _, written = gitops("PUT", f"{base}/contents/{target.path}", body)
      except urllib.error.HTTPError as error:
         error.close()
         if error.code not in legacy.GITOPS_CONFLICT_STATUSES:
            raise
         previous = file_sha
         continue
      commit = str(cast(dict[str, Any], written)["commit"]["sha"])
      _verify_written_commit(target, gitops, commit, updated)
      return _receipt(target, pins, commit, "committed", release)
   raise PublicationError("GitOps branch kept changing; no pin was written")


def _plan_digest_line(lines: list[str], bounds: list[tuple[int, int]], name: str, new_name: str, digest: str) -> tuple[int, bool, str]:
   """Return (line index, replace-existing, new text) for one enrolled entry after validating its shape."""
   if not legacy.IMAGE_NAME_PATTERN.fullmatch(new_name) or not legacy.DIGEST_PATTERN.fullmatch(digest):
      raise PublicationError("invalid image pin")
   chosen = [(start, stop) for start, stop in bounds if _entry_match(lines[start]).group(2) == name]
   if len(chosen) != 1:
      raise PublicationError(f"enrolled image {name!r} is not exactly one images entry")
   start, stop = chosen[0]
   indent = " " * (len(_entry_match(lines[start]).group(1)) + 2)
   seen: dict[str, int] = {}
   for number in range(start + 1, stop):
      line = lines[number]
      if not line.strip() or line.lstrip().startswith("#"):
         continue
      match = legacy.IMAGE_KEY_PATTERN.match(line)
      if not match or match.group(1) != indent or match.group(2) not in ("newName", "newTag", "digest") or match.group(2) in seen:
         raise PublicationError(f"unsupported or duplicate keys in images entry {name!r}")
      seen[match.group(2)] = number
   name_line = legacy.IMAGE_KEY_PATTERN.match(lines[seen["newName"]]) if "newName" in seen else None
   if name_line is None or name_line.group(3) != new_name:
      raise PublicationError(f"images entry {name!r} must already point at the enrolled registry image")
   text = indent + "digest: " + digest
   if "digest" in seen:
      return seen["digest"], True, text
   return max(seen.values()) + 1, False, text


def _verify_pins_are_served(target: GitOpsTarget, pins: tuple[ImagePin, ...], clients: GitOpsClients) -> dict[str, str]:
   """Require unique enrolled pins whose digests the registry serves by content; return the enrolled image map."""
   enrolled = dict(target.images)
   if not pins or len({pin.kustomize_name for pin in pins}) != len(pins):
      raise PublicationError("pins must be a nonempty list with unique entry names")
   for pin in pins:
      if pin.kustomize_name not in enrolled:
         raise PublicationError(f"{pin.kustomize_name!r} is not an image enrolled for this GitOps destination")
      image = enrolled[pin.kustomize_name]
      _, _, manifest = clients.registry_reader[image]("GET", f"/v2/{image}/manifests/{pin.digest}")
      if "sha256:" + hashlib.sha256(manifest).hexdigest() != pin.digest:
         raise PublicationError(f"registry does not serve {pin.digest} for {image} by content")
   return enrolled


def _current_verified_commit(target: GitOpsTarget, gitops: legacy.GitOpsCall) -> str:
   """For an already-current file, return the newest commit touching it after requiring it to be signature-verified."""
   path = target.path.replace("/", "%2F")
   _, found = gitops("GET", f"/repos/{target.repository}/commits?sha={target.branch}&path={path}&limit=1&stat=false")
   commits = cast(list[dict[str, Any]], found)
   if len(commits) != 1:
      raise PublicationError("no commit found for the already-current GitOps file")
   legacy.require_verified_commit(commits[0], "GitOps commit")
   return str(commits[0]["sha"])


def _verify_written_commit(target: GitOpsTarget, gitops: legacy.GitOpsCall, commit: str, intended: str) -> None:
   """After the only write: the commit is signed, touches only the enrolled file, and holds exactly the intended bytes."""
   base = "/repos/" + target.repository
   if not _COMMIT_ID.fullmatch(commit):
      raise PublicationError("forge did not return a commit id")
   _, fetched = gitops("GET", f"{base}/git/commits/{commit}")
   legacy.require_verified_commit(fetched, "GitOps commit")
   changed_files = cast(list[dict[str, Any]], cast(dict[str, Any], fetched).get("files") or [])
   touched = [item.get("filename") for item in changed_files]
   if touched != [target.path]:
      raise PublicationError("GitOps commit touched files other than the enrolled path")
   _, after = gitops("GET", f"{base}/contents/{target.path}?ref={commit}")
   if base64.b64decode("".join(str(cast(dict[str, Any], after)["content"]).split()), validate=True).decode("utf-8") != intended:
      raise PublicationError("committed GitOps bytes differ from the intended update")


def _receipt(target: GitOpsTarget, pins: tuple[ImagePin, ...], commit: str, status: str, release: ReleaseIdentity) -> GitOpsReceipt:
   """Assemble the pin receipt."""
   return GitOpsReceipt(
      repository=target.repository,
      branch=target.branch,
      path=target.path,
      commit=commit,
      status=status,
      pins=tuple(sorted((pin.kustomize_name, pin.digest) for pin in pins)),
      build_sha=release.build_sha,
      release_sha=release.release_sha,
   )


def _entry_match(line: str) -> re.Match[str]:
   """Match an `- name:` images entry line."""
   match = legacy.IMAGE_ENTRY_PATTERN.match(line)
   if match is None:
      raise PublicationError("unexpected images entry line")
   return match