platform-ci/tests/test_package_prefetch.py

"""The private Maven package prefetch: authorized, trusted-push only, and the credential never outlives the one Maven process."""

import json
import tempfile
import unittest
from collections.abc import Mapping
from pathlib import Path
from typing import cast

import yaml

from platform_ci.caching import save_allowed
from platform_ci.configuration import parse_project_configuration_text
from platform_ci.execution import INHERITED_VARIABLES
from platform_ci.package_prefetch import (
   PACKAGE_READ_SECRET_NAME,
   TOKEN_VARIABLE,
   PackageBinding,
   PackagePrefetchError,
   authorize,
   load_package_bindings,
   prefetch_packages,
)
from platform_ci.plan import Target

CONSUMER = "Example/consumer"
REVIEWED = "a" * 40
OTHER = "b" * 40
WORKFLOW = Path(__file__).resolve().parent.parent / ".gitea" / "scoped_workflows" / "pipeline.yaml"
PLACEHOLDER_TOKEN = "placeholder-not-a-credential"  # noqa: S105
BINDINGS = {"example-read": PackageBinding(name="example-read", secret=PACKAGE_READ_SECRET_NAME, username="example-reader", consumers=(CONSUMER,))}


def trusted_environment(home: str) -> dict[str, str]:
   """A push to the default branch of the consumer, with the token present."""
   return {
      "HOME": home,
      "CI_EVENT_NAME": "push",
      "CI_EVENT_REF": "refs/heads/main",
      "CI_DEFAULT_BRANCH": "main",
      "CI_REPOSITORY": CONSUMER,
      TOKEN_VARIABLE: PLACEHOLDER_TOKEN,
   }


def target(directory: str = ".") -> Target:
   """The one field set the prefetch reads."""
   return cast("Target", {"name": "host", "directory": directory, "environment": {"REPOSITORY_DECLARED": "x"}, "package_reads": "example-read"})


class PackagePrefetchTest(unittest.TestCase):
   """Nothing here reads a real credential or starts Maven."""

   def test_the_shipped_manifest_is_valid_and_declares_the_host_binding(self) -> None:
      """The manifest parses under its own schema and names the one slot; the host config can reference its binding."""
      bindings = load_package_bindings()
      self.assertEqual(PACKAGE_READ_SECRET_NAME, bindings["kof22-qrun-read"].secret)
      self.assertIn("KofTwentyTwo/Website-Backend", bindings["kof22-qrun-read"].consumers)
      configuration = parse_project_configuration_text("schema: 1\ncomponents:\n  host:\n    build:\n      tool: maven\n      arguments: [verify]\n    toolchains:\n      java: '21'\n    package_reads:\n      binding: kof22-qrun-read\n")
      self.assertEqual("kof22-qrun-read", configuration.components[0].package_reads)

   def test_only_a_maven_component_may_declare_package_reads(self) -> None:
      """A non-Maven tool is refused at configuration time."""
      with self.assertRaises(Exception) as caught:  # noqa: PT027 - ConfigurationError is the specific type; the message is what matters here
         parse_project_configuration_text("schema: 1\ncomponents:\n  c:\n    build:\n      tool: cargo\n      commands: [[test]]\n    toolchains:\n      rust: '1'\n    package_reads:\n      binding: kof22-qrun-read\n")
      self.assertIn("package_reads", str(caught.exception))

   def test_an_unknown_binding_or_other_repository_is_refused(self) -> None:
      """The manifest, not the consumer, decides who may use a binding."""
      with self.assertRaises(PackagePrefetchError):
         authorize("missing", CONSUMER, BINDINGS)
      with self.assertRaises(PackagePrefetchError):
         authorize("example-read", "Other/repo", BINDINGS)

   def test_only_a_push_to_the_default_branch_may_read(self) -> None:
      """Pull requests, tags, feature branches, missing default-branch metadata, other repositories and a missing token never start Maven."""
      with tempfile.TemporaryDirectory() as home:
         base = trusted_environment(home)
         refused = [
            {**base, "CI_EVENT_NAME": "pull_request"},
            {**base, "CI_EVENT_REF": "refs/heads/feature"},
            {**base, "CI_EVENT_REF": "refs/tags/v1"},
            {**base, "CI_DEFAULT_BRANCH": ""},
            {**base, "CI_REPOSITORY": "Other/repo"},
            {**base, TOKEN_VARIABLE: ""},
         ]
         for environment in refused:
            with self.subTest(environment=environment), self.assertRaises(PackagePrefetchError):
               prefetch_packages(target(), Path(home), environment, BINDINGS, lambda *_arguments: self.fail("Maven must not start"))

   def test_the_credential_reaches_only_the_maven_process_and_the_settings_are_removed(self) -> None:
      """The settings hold a placeholder, not the token; the token is in that process's environment alone; the scratch settings are gone afterwards."""
      seen: dict[str, object] = {}

      def run(command: list[str], directory: Path, environment: Mapping[str, str]) -> int:
         settings = Path(command[command.index("-s") + 1])
         seen["settings_text"] = settings.read_text(encoding="utf-8")
         seen["settings_path"] = settings
         seen["token_in_environment"] = environment.get(TOKEN_VARIABLE)
         seen["command"] = command
         return 0

      with tempfile.TemporaryDirectory() as home:
         lines = prefetch_packages(target(), Path(home), trusted_environment(home), BINDINGS, run)
         settings_text = cast("str", seen["settings_text"])
         self.assertNotIn(PLACEHOLDER_TOKEN, settings_text)
         self.assertIn("${env." + TOKEN_VARIABLE + "}", settings_text)
         self.assertIn("example-reader", settings_text)
         self.assertEqual(PLACEHOLDER_TOKEN, seen["token_in_environment"])
         self.assertIn("dependency:go-offline", cast("list[str]", seen["command"]))
         self.assertFalse(cast("Path", seen["settings_path"]).exists())
         self.assertFalse((Path(home) / ".m2" / "settings.xml").exists())
         self.assertNotIn(PLACEHOLDER_TOKEN, "\n".join(lines))

   def test_a_failed_prefetch_removes_the_settings_and_stops_the_verification(self) -> None:
      """A non-zero Maven exit raises, after the scratch settings were removed."""
      seen: dict[str, Path] = {}

      def run(command: list[str], _directory: Path, _environment: Mapping[str, str]) -> int:
         seen["settings"] = Path(command[command.index("-s") + 1])
         return 1

      with tempfile.TemporaryDirectory() as home, self.assertRaises(PackagePrefetchError):
         prefetch_packages(target(), Path(home), trusted_environment(home), BINDINGS, run)
      self.assertFalse(seen["settings"].exists())

   def test_the_registry_id_is_pinned_to_the_constant_url_and_repository_environment_is_not_passed(self) -> None:
      """A project POM cannot redirect the credential (a mirror binds the registry id to the constant URL) and its declared tool environment is not merged."""
      seen: dict[str, object] = {}

      def run(command: list[str], _directory: Path, environment: Mapping[str, str]) -> int:
         seen["settings"] = Path(command[command.index("-s") + 1]).read_text(encoding="utf-8")
         seen["environment"] = dict(environment)
         seen["command"] = command
         return 0

      with tempfile.TemporaryDirectory() as home:
         prefetch_packages(target(), Path(home), trusted_environment(home), BINDINGS, run)
      self.assertIn("<mirrorOf>kof22-qrun</mirrorOf>", cast("str", seen["settings"]))
      self.assertNotIn("REPOSITORY_DECLARED", cast("dict[str, str]", seen["environment"]))
      self.assertIn("-Dmaven.ext.class.path=", cast("list[str]", seen["command"]))

   def test_a_repository_with_maven_launch_files_is_refused(self) -> None:
      """.mvn/extensions.xml, jvm.config and maven.config can load code or options into the credentialed process, so Maven is not started."""
      for name in ("extensions.xml", "jvm.config", "maven.config"):
         with self.subTest(name=name), tempfile.TemporaryDirectory() as home:
            (Path(home) / ".mvn").mkdir()
            (Path(home) / ".mvn" / name).write_text("x", encoding="utf-8")
            with self.assertRaises(PackagePrefetchError):
               prefetch_packages(target(), Path(home), trusted_environment(home), BINDINGS, lambda *_arguments: self.fail("Maven must not start"))

   def test_unsafe_poms_are_refused_before_any_process_starts(self) -> None:
      """Build extensions, a reference to the credential variable and a registry-id collision are refused; ordinary environment references, other repositories and the registry pass."""
      pom = "<project>{}</project>"
      cases = {
         "extension": pom.format("<build><extensions><extension><groupId>g</groupId></extension></extensions></build>"),
         "credential variable": pom.format("<properties><x>${env.CI_PACKAGE_READ_TOKEN}</x></properties>"),
         "credential variable hidden by a character reference": pom.format("<properties><x>&#36;{env.CI_PACKAGE_READ_TOKEN}</x></properties>"),
         "namespaced extension": '<project xmlns="http://maven.apache.org/POM/4.0.0"><build><extensions><extension/></extensions></build></project>',
         "namespaced registry id with another url": '<project xmlns="http://maven.apache.org/POM/4.0.0"><repositories><repository><id>kof22-qrun</id><url>https://attacker.example/m</url></repository></repositories></project>',
         "registry id with another url": pom.format("<repositories><repository><id>kof22-qrun</id><url>https://attacker.example/maven</url></repository></repositories>"),
      }
      for label, text in cases.items():
         with self.subTest(label=label), tempfile.TemporaryDirectory() as home:
            (Path(home) / "pom.xml").write_text(text, encoding="utf-8")
            with self.assertRaises(PackagePrefetchError):
               prefetch_packages(target(), Path(home), trusted_environment(home), BINDINGS, lambda *_arguments: self.fail("Maven must not start"))
      # what the real host POM contains and must keep working: database settings read from the environment and the Central snapshot repository
      ordinary = pom.format("<repositories><repository><id>central-portal-snapshots</id><url>https://central.sonatype.com/repository/maven-snapshots/</url></repository></repositories><build><plugins><plugin><configuration><url>${env.RDBMS_URL}</url></configuration></plugin></plugins></build>")
      with tempfile.TemporaryDirectory() as home:
         (Path(home) / "pom.xml").write_text(ordinary, encoding="utf-8")
         prefetch_packages(target(), Path(home), trusted_environment(home), BINDINGS, lambda *_arguments: 0)
      allowed = pom.format("<repositories><repository><id>kof22-qrun</id><url>https://git.kof22.com/api/packages/QRun-IO/maven</url></repository></repositories>")
      with tempfile.TemporaryDirectory() as home:
         (Path(home) / "pom.xml").write_text(allowed, encoding="utf-8")
         prefetch_packages(target(), Path(home), trusted_environment(home), BINDINGS, lambda *_arguments: 0)

   def test_maven_launch_files_above_the_component_are_refused_too(self) -> None:
      """A `.mvn/maven.config` in a parent directory of the component, up to the source root, is refused as well."""
      with tempfile.TemporaryDirectory() as home:
         (Path(home) / ".mvn").mkdir()
         (Path(home) / ".mvn" / "maven.config").write_text("x", encoding="utf-8")
         (Path(home) / "app").mkdir()
         with self.assertRaises(PackagePrefetchError):
            prefetch_packages(target("app"), Path(home), trusted_environment(home), BINDINGS, lambda *_arguments: self.fail("Maven must not start"))


REVIEWED_BINDINGS = {"example-read": PackageBinding(name="example-read", secret=PACKAGE_READ_SECRET_NAME, username="example-reader", consumers=(CONSUMER,), reviewed_commits=(REVIEWED,))}


def pull_request_environment(home: str, head: str = REVIEWED, head_repository: str = CONSUMER) -> dict[str, str]:
   """The forge-supplied fields of a pull request: the event name, the head and base repositories and the head commit id."""
   return {
      "HOME": home,
      "CI_EVENT_NAME": "pull_request",
      "CI_EVENT_REF": "refs/pull/7/merge",
      "CI_DEFAULT_BRANCH": "main",
      "CI_REPOSITORY": CONSUMER,
      "CI_PR_HEAD_SHA": head,
      "CI_PR_HEAD_REPOSITORY": head_repository,
      "CI_PR_BASE_REPOSITORY": CONSUMER,
      TOKEN_VARIABLE: PLACEHOLDER_TOKEN,
   }


class ReviewedPullRequestTest(unittest.TestCase):
   """A same-repository pull request may prefetch only when its forge-supplied head commit is an exact reviewed commit; everything else is refused first."""

   def test_the_exact_reviewed_head_commit_may_prefetch_and_the_token_stays_in_the_maven_child(self) -> None:
      """An authorized head (the checkout is the head or a merge of it) runs Maven once, with the token only in that child's environment."""
      calls: list[Mapping[str, str]] = []
      with tempfile.TemporaryDirectory() as home:
         lines = prefetch_packages(target(), Path(home), pull_request_environment(home), REVIEWED_BINDINGS, lambda _c, _d, environment: calls.append(dict(environment)) or 0, lambda *_a: ["m" * 40, REVIEWED, "c" * 40])
      self.assertEqual(1, len(calls))
      self.assertEqual(PLACEHOLDER_TOKEN, calls[0][TOKEN_VARIABLE])
      self.assertNotIn(PLACEHOLDER_TOKEN, "\n".join(lines))

   def test_other_commits_forks_an_empty_list_and_a_different_checkout_are_refused_before_any_child_process(self) -> None:
      """A different head commit of the same repository, a fork, an empty allowlist, a malformed id and a checkout that is not the head never start Maven."""
      def never(*_arguments: object) -> int:
         self.fail("Maven must not start")

      with tempfile.TemporaryDirectory() as home:
         head_ok = lambda *_a: [REVIEWED]  # noqa: E731
         cases = [
            (pull_request_environment(home, head=OTHER), REVIEWED_BINDINGS, head_ok),
            (pull_request_environment(home, head_repository="Fork/consumer"), REVIEWED_BINDINGS, head_ok),
            (pull_request_environment(home), BINDINGS, head_ok),
            (pull_request_environment(home, head="short"), REVIEWED_BINDINGS, head_ok),
            (pull_request_environment(home), REVIEWED_BINDINGS, lambda *_a: [OTHER, "c" * 40]),
            ({**pull_request_environment(home), "CI_EVENT_NAME": "pull_request_target"}, REVIEWED_BINDINGS, head_ok),
         ]
         for environment, bindings, head in cases:
            with self.subTest(environment=environment.get("CI_PR_HEAD_SHA"), repository=environment.get("CI_PR_HEAD_REPOSITORY")), self.assertRaises(PackagePrefetchError):
               prefetch_packages(target(), Path(home), environment, bindings, never, head)

   def test_the_manifest_allowlist_must_be_full_commit_ids_and_defaults_empty(self) -> None:
      """The shipped list is empty (inert); a malformed entry is refused at load."""
      self.assertEqual((), load_package_bindings()["kof22-qrun-read"].reviewed_commits)
      with tempfile.TemporaryDirectory() as directory:
         path = Path(directory) / "bindings.yaml"
         path.write_text("schema: 1\nbindings:\n  b:\n    secret: PACKAGE_READ_TOKEN\n    username: u\n    consumers: [Example/consumer]\n    reviewed_commits: [main]\n", encoding="utf-8")
         with self.assertRaises(PackagePrefetchError):
            load_package_bindings(path)

   def test_a_pull_request_never_writes_the_cache_and_the_token_is_forwarded_to_the_prefetch_step_alone(self) -> None:
      """No cache write for a pull request; only the prefetch steps carry the secret, and only for a default-branch push or a same-repository pull request."""
      self.assertFalse(save_allowed("pull_request", "refs/pull/7/merge", "main"))
      workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
      carriers = 0
      for name, job in workflow["jobs"].items():
         for step in job.get("steps", []):
            text = json.dumps(step)
            if "PACKAGE_READ_TOKEN" not in text:
               continue
            carriers += 1
            self.assertEqual("Prefetch private Maven packages", step["name"], name)
            condition = step["if"]
            self.assertIn("github.event.pull_request.head.repo.full_name == github.event.pull_request.base.repo.full_name", condition)
            self.assertIn("github.event_name == 'push'", condition)
            self.assertNotIn("always()", condition)
         for step in job.get("steps", []):
            if step.get("name", "").startswith("Save dependency cache"):
               self.assertIn("github.event_name == 'push'", step["if"])
      self.assertEqual(3, carriers)
      self.assertNotIn(TOKEN_VARIABLE, INHERITED_VARIABLES)


if __name__ == "__main__":
   unittest.main()