platform-ci/platform_ci/dependency_checkout.py

"""Fetch a component's pinned private dependencies before its build, with an operator-authorized read binding.

UNTESTED: written while test execution was suspended; no credential was read and no network call was made.

The consumer's `.ci.yaml` owns the dependency identity (`checkouts.<name>`: a logical `binding`, a `repository`, a full commit `ref`, a `path`). The
operator-owned authorization manifest `runtime/checkout-bindings.yaml`, part of the reviewed platform revision, owns the read scope: which consuming
repositories may use which binding to read which repositories. Nothing here branches on a repository name; an empty manifest authorizes nothing.

The one credential is the existing read-only repository secret the workflow forwards to this step alone (`CI_CHECKOUT_TOKEN`). It is given to Git only
through the environment of the fetch process, never in a URL or a file, so no credential is written into the checkout, and no later build step receives it.
The fetched commit must equal the pinned `ref`.
"""

import base64
import os
import subprocess
import tempfile
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Final, cast

import yaml

from platform_ci.configuration import parse_yaml_rejecting_duplicate_keys
from platform_ci.plan import CheckoutTarget, Target
from platform_ci.workspace import PipelineError

BINDINGS_PATH: Final = Path(__file__).resolve().parent.parent / "runtime" / "checkout-bindings.yaml"
BINDINGS_SCHEMA: Final = 1
CHECKOUT_SECRET_NAME: Final = "PUBLISHER_CHECKOUT_TOKEN"  # noqa: S105 - the name of the existing repository secret the workflow forwards, not a credential
"""The one existing read binding the workflow forwards to this step; a binding in the manifest must name it, so the manifest authorizes who may use the slot."""
TRUSTED_FORGE: Final = "https://git.kof22.com"
_GIT_TIMEOUT_SECONDS: Final = 300

GitRunner = Callable[[Path, tuple[str, ...], Mapping[str, str]], bytes]


class DependencyCheckoutError(PipelineError):
   """A pinned private dependency was not authorized, could not be fetched, or is not exactly the pinned commit."""


@dataclass(frozen=True)
class Binding:
   """One authorization: the secret slot it uses, the consuming repositories it serves and the repositories it may read."""

   name: str
   secret: str
   consumers: tuple[str, ...]
   repositories: tuple[str, ...]


def load_bindings(path: Path = BINDINGS_PATH) -> dict[str, Binding]:
   """Read the operator's manifest; a missing or malformed one authorizes nothing and is an error, never an empty pass."""
   try:
      loaded = parse_yaml_rejecting_duplicate_keys(path.read_text(encoding="utf-8"))
   except (OSError, UnicodeDecodeError, yaml.YAMLError) as error:
      raise DependencyCheckoutError(f"the checkout authorization manifest cannot be read: {type(error).__name__}") from error
   malformed = f"the checkout authorization manifest must be a mapping with schema {BINDINGS_SCHEMA} and a `bindings` mapping"
   if not isinstance(loaded, dict):
      raise DependencyCheckoutError(malformed)
   raw = cast("dict[str, object]", loaded)
   if raw.get("schema") != BINDINGS_SCHEMA or not isinstance(raw.get("bindings"), dict):
      raise DependencyCheckoutError(malformed)
   bindings: dict[str, Binding] = {}
   for name, entry in cast("dict[object, object]", raw["bindings"]).items():
      if not isinstance(name, str) or not isinstance(entry, dict):
         raise DependencyCheckoutError("every binding must be a named mapping")
      fields = cast("dict[str, object]", entry)
      if set(fields) != {"secret", "consumers", "repositories"} or fields["secret"] != CHECKOUT_SECRET_NAME:
         raise DependencyCheckoutError(f"binding {name!r} must have exactly secret ({CHECKOUT_SECRET_NAME}), consumers and repositories")
      lists: list[tuple[str, ...]] = []
      for key in ("consumers", "repositories"):
         value = fields[key]
         if not isinstance(value, list) or not value or not all(isinstance(item, str) and "/" in item for item in cast("list[object]", value)):
            raise DependencyCheckoutError(f"binding {name!r} {key} must be a non-empty list of owner/name")
         lists.append(tuple(sorted(cast("list[str]", value))))
      bindings[name] = Binding(name=name, secret=CHECKOUT_SECRET_NAME, consumers=lists[0], repositories=lists[1])
   return bindings


def authorize(checkout: CheckoutTarget, consumer: str, bindings: Mapping[str, Binding]) -> Binding:
   """Require the named binding to exist and to allow this consuming repository to read this dependency repository."""
   binding = bindings.get(checkout["binding"])
   if binding is None:
      raise DependencyCheckoutError(f"no authorization binding named {checkout['binding']!r} exists in the operator's manifest")
   if consumer not in binding.consumers:
      raise DependencyCheckoutError(f"binding {binding.name!r} does not authorize this repository to use it")
   if checkout["repository"] not in binding.repositories:
      raise DependencyCheckoutError(f"binding {binding.name!r} does not authorize reading {checkout['repository']!r}")
   return binding


def run_git(directory: Path, arguments: tuple[str, ...], environment: Mapping[str, str]) -> bytes:
   """Run Git in a directory with hooks off and an explicit environment; stderr is never surfaced, because it can echo the remote."""
   try:
      result = subprocess.run(["git", "-c", f"core.hooksPath={os.devnull}", *arguments], cwd=directory, env=dict(environment), capture_output=True, timeout=_GIT_TIMEOUT_SECONDS, check=False)  # noqa: S603, S607
   except (OSError, subprocess.TimeoutExpired) as error:
      raise DependencyCheckoutError(f"git could not run: {type(error).__name__}") from error
   if result.returncode != 0:
      raise DependencyCheckoutError("a git command failed while fetching a pinned dependency")
   return result.stdout


def _local_environment() -> dict[str, str]:
   """Return the environment of a Git command that never touches the network: no credential, no global or system configuration."""
   return {"PATH": os.environ.get("PATH", ""), "HOME": tempfile.gettempdir(), "GIT_TERMINAL_PROMPT": "0", "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull}


def _authenticated_environment(token: str) -> dict[str, str]:
   """Return the environment of one fetch process: the credential as a command-scope header, no global or system configuration, no prompt."""
   auth = base64.b64encode(f"x-access-token:{token}".encode()).decode()
   return {
      "PATH": os.environ.get("PATH", ""),
      "HOME": tempfile.gettempdir(),
      "GIT_TERMINAL_PROMPT": "0",
      "GIT_CONFIG_NOSYSTEM": "1",
      "GIT_CONFIG_GLOBAL": os.devnull,
      "GIT_CONFIG_COUNT": "2",
      # origin-scoped: Git applies this header only to requests whose URL is under the trusted forge, and the pinned fetch does not follow redirects
      "GIT_CONFIG_KEY_0": f"http.{TRUSTED_FORGE}/.extraheader",
      "GIT_CONFIG_VALUE_0": f"Authorization: Basic {auth}",
      "GIT_CONFIG_KEY_1": "http.followRedirects",
      "GIT_CONFIG_VALUE_1": "false",
   }


_PROTECTED_DIRECTORIES: Final = frozenset({".git", ".platform-ci"})


def _destination(root: Path, relative: str, name: str) -> Path:
   """Return the resolved destination under the checkout root; no path component may be a symbolic link and it may not be inside `.git` or `.platform-ci`.

   A link is refused rather than followed, so nothing is ever written through a link, including one that stays inside the checkout (for example one
   aimed at the source's own Git metadata). A plain nested path such as `a/b/c` is unchanged.
   """
   walked = root
   for part in Path(relative).parts:
      walked /= part
      if walked.is_symlink():
         raise DependencyCheckoutError(f"dependency {name!r}: its path passes through a symbolic link")
   destination = (root / relative).resolve()
   inside = destination.relative_to(root).parts if destination.is_relative_to(root) else None
   if inside is None or (inside[:1] and inside[0] in _PROTECTED_DIRECTORIES):
      raise DependencyCheckoutError(f"dependency {name!r}: its path is outside the checkout or already holds files")
   return destination


def checkout_dependencies(
   target: Target,
   source_directory: Path,
   environment: Mapping[str, str],
   bindings: Mapping[str, Binding] | None = None,
   git: GitRunner = run_git,
) -> list[str]:
   """Fetch every pinned dependency of the target to its path under the checkout and return one line per dependency.

   Each is authorized against the operator's manifest first (before the credential is used), fetched at exactly its commit, and verified to be that
   commit. The credential exists only in the environment of the fetch process. A target without checkouts does nothing.
   """
   if not target["checkouts"]:
      return []
   consumer = environment.get("CI_REPOSITORY", "")
   token = environment.get("CI_CHECKOUT_TOKEN", "")
   if not consumer:
      raise DependencyCheckoutError("the consuming repository identity is required to authorize a private checkout")
   if not token:
      raise DependencyCheckoutError(f"the read binding {CHECKOUT_SECRET_NAME} is not available to this job, so the private dependency cannot be fetched")
   table = bindings if bindings is not None else load_bindings()
   root = source_directory.resolve()
   lines: list[str] = []
   for name, checkout in sorted(target["checkouts"].items()):
      authorize(checkout, consumer, table)
      destination = _destination(root, checkout["path"], name)
      if destination.exists() and any(destination.iterdir()):
         raise DependencyCheckoutError(f"dependency {name!r}: its path is outside the checkout or already holds files")
      destination.mkdir(parents=True, exist_ok=True)
      url = f"{TRUSTED_FORGE}/{checkout['repository']}.git"
      local = _local_environment()
      git(destination, ("init", "-q", "--template="), local)
      git(destination, ("fetch", "-q", "--depth=1", "--no-tags", url, checkout["ref"]), _authenticated_environment(token))
      git(destination, ("checkout", "-q", "--detach", "FETCH_HEAD"), local)
      head = git(destination, ("rev-parse", "HEAD"), local).decode().strip()
      if head != checkout["ref"]:
         raise DependencyCheckoutError(f"dependency {name!r}: the fetched commit is not the pinned one")
      try:
         configuration = (destination / ".git" / "config").read_text(encoding="utf-8", errors="replace")
      except OSError as error:
         raise DependencyCheckoutError(f"dependency {name!r}: the checkout configuration cannot be read: {type(error).__name__}") from error
      if "Authorization" in configuration or token in configuration:
         raise DependencyCheckoutError(f"dependency {name!r}: a credential reached the checkout configuration")
      lines.append(f"dependency {name}: {checkout['repository']} at {head[:12]} in {checkout['path']}")
   return lines