platform-ci/tests/test_dependency_checkout.py

"""Pinned private dependency checkout: authorization, exact commit, destination containment, credential isolation and redaction.

NOT EXECUTED when written (test execution was suspended). Every test uses a fake Git runner and temporary directories; no process is started, no
network is used, and the only "token" is an obviously fake string.
"""

import base64
import subprocess
import tempfile
import unittest
from collections.abc import Mapping
from pathlib import Path
from typing import cast, override
from unittest import mock

from platform_ci.dependency_checkout import (
   CHECKOUT_SECRET_NAME,
   TRUSTED_FORGE,
   Binding,
   DependencyCheckoutError,
   authorize,
   checkout_dependencies,
   load_bindings,
   run_git,
)
from platform_ci.plan import CheckoutTarget, Target

PINNED = "a" * 40
OTHER = "b" * 40
TOKEN = "fake-token-not-a-secret-0123"  # noqa: S105
CONSUMER = "Org/consumer"
DEPENDENCY = "Org/content"
BINDINGS = {"content-read": Binding(name="content-read", secret=CHECKOUT_SECRET_NAME, consumers=(CONSUMER,), repositories=(DEPENDENCY,))}
ENVIRONMENT = {"CI_REPOSITORY": CONSUMER, "CI_CHECKOUT_TOKEN": TOKEN}


def _target(**checkouts: CheckoutTarget) -> Target:
   """Return a target carrying only checkouts: the module under test reads nothing else."""
   return cast("Target", {"checkouts": checkouts})


def _checkout(path: str = "deps/content", ref: str = PINNED, binding: str = "content-read", repository: str = DEPENDENCY) -> CheckoutTarget:
   """Return one planned checkout."""
   return CheckoutTarget(binding=binding, repository=repository, ref=ref, path=path)


class FakeGit:
   """A GitRunner that records every call and behaves like a successful `init`, `fetch`, `checkout` and `rev-parse` of `head`."""

   def __init__(self, head: str = PINNED, config_text: str = "[core]\n\tbare = false\n") -> None:
      self.calls: list[tuple[Path, tuple[str, ...], dict[str, str]]] = []
      self.head = head
      self.config_text = config_text

   def __call__(self, directory: Path, arguments: tuple[str, ...], environment: Mapping[str, str]) -> bytes:
      """Record the call; `init` creates the Git directory and its configuration, `rev-parse` returns the configured head."""
      self.calls.append((directory, arguments, dict(environment)))
      if arguments[0] == "init":
         (directory / ".git").mkdir()
         (directory / ".git" / "config").write_text(self.config_text, encoding="utf-8")
      if arguments[0] == "rev-parse":
         return f"{self.head}\n".encode()
      return b""

   def verbs(self) -> list[str]:
      """Return the Git subcommand of each call, in order."""
      return [arguments[0] for _directory, arguments, _environment in self.calls]


class _CheckoutCase(unittest.TestCase):
   """A temporary checkout root for each test."""

   @override
   def setUp(self) -> None:
      """Create the checkout directory."""
      scratch = tempfile.TemporaryDirectory()
      self.addCleanup(scratch.cleanup)
      self.scratch = Path(scratch.name)
      self.root = self.scratch / "checkout"
      self.root.mkdir()

   def run_checkout(self, target: Target, git: FakeGit, environment: Mapping[str, str] = ENVIRONMENT) -> list[str]:
      """Run the real function with the fixed bindings, so no manifest file is needed."""
      return checkout_dependencies(target, self.root, environment, BINDINGS, git)


class AuthorizationTest(_CheckoutCase):
   """A dependency is fetched only when the operator's logical binding allows this consumer to read this repository."""

   def test_an_authorized_consumer_fetches_exactly_the_pinned_dependency(self) -> None:
      """Allow: the result names the repository and the short commit, and the work tree is under the declared path."""
      git = FakeGit()
      lines = self.run_checkout(_target(content=_checkout()), git)
      self.assertEqual(lines, [f"dependency content: {DEPENDENCY} at {PINNED[:12]} in deps/content"])
      self.assertEqual(git.verbs(), ["init", "fetch", "checkout", "rev-parse"])
      self.assertEqual({directory for directory, _arguments, _environment in git.calls}, {(self.root / "deps" / "content").resolve()})

   def test_every_denial_happens_before_anything_is_fetched_or_created(self) -> None:
      """Deny: unknown binding, a consumer the binding does not list, a repository it does not list; Git never runs and no directory appears."""
      cases = {
         "unknown binding": (_checkout(binding="other-read"), CONSUMER),
         "consumer not listed": (_checkout(), "Org/someone-else"),
         "repository not listed": (_checkout(repository="Org/secret-repo"), CONSUMER),
      }
      for label, (checkout, consumer) in cases.items():
         with self.subTest(label):
            git = FakeGit()
            with self.assertRaises(DependencyCheckoutError):
               self.run_checkout(_target(content=checkout), git, {"CI_REPOSITORY": consumer, "CI_CHECKOUT_TOKEN": TOKEN})
            self.assertEqual(git.calls, [])
            self.assertFalse((self.root / "deps").exists())

   def test_the_denial_names_the_binding_not_the_credential(self) -> None:
      """The refusal message is about the binding and repository and never contains the token."""
      with self.assertRaises(DependencyCheckoutError) as caught:
         authorize(_checkout(repository="Org/secret-repo"), CONSUMER, BINDINGS)
      self.assertIn("content-read", str(caught.exception))
      self.assertNotIn(TOKEN, str(caught.exception))

   def test_a_missing_identity_or_binding_secret_stops_before_any_work(self) -> None:
      """Without the consuming repository or the forwarded secret the step fails closed instead of fetching anonymously."""
      for label, environment in {"no consumer": {"CI_CHECKOUT_TOKEN": TOKEN}, "no token": {"CI_REPOSITORY": CONSUMER}}.items():
         with self.subTest(label):
            git = FakeGit()
            with self.assertRaises(DependencyCheckoutError):
               self.run_checkout(_target(content=_checkout()), git, environment)
            self.assertEqual(git.calls, [])

   def test_a_target_without_checkouts_needs_neither_identity_nor_token(self) -> None:
      """Most targets carry no checkouts; nothing is required and nothing runs."""
      git = FakeGit()
      self.assertEqual(checkout_dependencies(_target(), self.root, {}, BINDINGS, git), [])
      self.assertEqual(git.calls, [])


class PinnedCommitTest(_CheckoutCase):
   """The fetched commit must equal the pinned `ref` exactly."""

   def test_a_head_that_is_not_the_pinned_commit_is_refused(self) -> None:
      """Mismatch: a remote that answers with a different commit fails the step."""
      with self.assertRaisesRegex(DependencyCheckoutError, "not the pinned one"):
         self.run_checkout(_target(content=_checkout()), FakeGit(head=OTHER))

   def test_a_prefix_of_the_pinned_commit_does_not_match(self) -> None:
      """Only the full id counts; an abbreviated or longer head is a mismatch."""
      for head in (PINNED[:12], PINNED + "0"):
         with self.subTest(head=head), self.assertRaises(DependencyCheckoutError):
            self.run_checkout(_target(content=_checkout(path=f"deps/{len(head)}")), FakeGit(head=head))

   def test_the_pinned_commit_is_what_is_fetched(self) -> None:
      """The fetch asks the trusted forge for exactly the declared repository and commit, with no tags and a depth of one."""
      git = FakeGit()
      self.run_checkout(_target(content=_checkout()), git)
      fetch = next(arguments for _directory, arguments, _environment in git.calls if arguments[0] == "fetch")
      self.assertEqual(fetch[-2:], (f"{TRUSTED_FORGE}/{DEPENDENCY}.git", PINNED))
      self.assertIn("--no-tags", fetch)
      self.assertIn("--depth=1", fetch)


class DestinationTest(_CheckoutCase):
   """The destination must stay inside the checkout and must not already hold files (as the module actually enforces)."""

   def test_a_path_that_leaves_the_checkout_is_refused_before_git_runs(self) -> None:
      """`..` escapes resolve outside the root and are refused; nothing is created outside it."""
      git = FakeGit()
      with self.assertRaisesRegex(DependencyCheckoutError, "outside the checkout"):
         self.run_checkout(_target(content=_checkout(path="../escaped")), git)
      self.assertEqual(git.calls, [])
      self.assertFalse((self.scratch / "escaped").exists())

   def test_a_symlinked_parent_that_points_outside_is_refused(self) -> None:
      """A link inside the checkout aimed at another directory resolves outside the root, so the fetch never happens there."""
      outside = self.scratch / "outside"
      outside.mkdir()
      try:
         (self.root / "link").symlink_to(outside, target_is_directory=True)
      except (OSError, NotImplementedError):
         self.skipTest("this host cannot create symbolic links")
      git = FakeGit()
      with self.assertRaisesRegex(DependencyCheckoutError, "symbolic link"):
         self.run_checkout(_target(content=_checkout(path="link/dep")), git)
      self.assertEqual(git.calls, [])
      self.assertEqual(list(outside.iterdir()), [])

   def test_a_link_inside_the_checkout_to_its_own_metadata_is_refused_without_git_or_mutation(self) -> None:
      """A link that stays inside the root but points at `.git` or `.platform-ci` is refused: no Git call, and the metadata is byte-for-byte unchanged."""
      for protected in (".git", ".platform-ci"):
         with self.subTest(protected):
            self.setUp()
            target_directory = self.root / protected / "refs"
            target_directory.mkdir(parents=True)
            marker = self.root / protected / "marker"
            marker.write_text("original", encoding="utf-8")
            try:
               (self.root / "deps").symlink_to(target_directory, target_is_directory=True)
            except (OSError, NotImplementedError):
               self.skipTest("this host cannot create symbolic links")
            before = sorted(path.relative_to(self.root).as_posix() for path in self.root.rglob("*") if not path.is_symlink())
            git = FakeGit()
            with self.assertRaisesRegex(DependencyCheckoutError, "symbolic link"):
               self.run_checkout(_target(content=_checkout(path="deps/content")), git)
            self.assertEqual(git.calls, [])
            self.assertEqual(sorted(path.relative_to(self.root).as_posix() for path in self.root.rglob("*") if not path.is_symlink()), before)
            self.assertEqual(marker.read_text(encoding="utf-8"), "original")

   def test_a_link_as_the_destination_itself_is_refused(self) -> None:
      """The final component may not be a link either, even to an empty directory inside the checkout."""
      (self.root / "elsewhere").mkdir()
      try:
         (self.root / "dep").symlink_to(self.root / "elsewhere", target_is_directory=True)
      except (OSError, NotImplementedError):
         self.skipTest("this host cannot create symbolic links")
      git = FakeGit()
      with self.assertRaisesRegex(DependencyCheckoutError, "symbolic link"):
         self.run_checkout(_target(content=_checkout(path="dep")), git)
      self.assertEqual(git.calls, [])

   def test_a_valid_nested_path_still_works(self) -> None:
      """Plain nested directories, created as needed, are accepted."""
      git = FakeGit()
      self.run_checkout(_target(content=_checkout(path="vendor/deep/content")), git)
      self.assertTrue((self.root / "vendor" / "deep" / "content" / ".git").is_dir())

   def test_a_destination_that_already_holds_files_is_refused_and_left_untouched(self) -> None:
      """Existing content is never overlaid or replaced."""
      existing = self.root / "deps" / "content"
      existing.mkdir(parents=True)
      (existing / "keep.txt").write_text("mine", encoding="utf-8")
      git = FakeGit()
      with self.assertRaisesRegex(DependencyCheckoutError, "already holds files"):
         self.run_checkout(_target(content=_checkout()), git)
      self.assertEqual(git.calls, [])
      self.assertEqual((existing / "keep.txt").read_text(encoding="utf-8"), "mine")

   def test_an_existing_empty_directory_is_usable(self) -> None:
      """An empty directory is a valid destination."""
      (self.root / "deps" / "content").mkdir(parents=True)
      self.assertEqual(len(self.run_checkout(_target(content=_checkout()), FakeGit())), 1)


class CredentialIsolationTest(_CheckoutCase):
   """The token reaches only the environment of the fetch process, never a URL, a file or a later step."""

   def test_only_the_fetch_child_environment_carries_the_credential(self) -> None:
      """No argument, no local-command environment and no checked-out file holds the token; the caller's environment is not modified."""
      git = FakeGit()
      caller = dict(ENVIRONMENT)
      self.run_checkout(_target(content=_checkout()), git, caller)
      encoded = base64.b64encode(f"x-access-token:{TOKEN}".encode()).decode()
      self.assertEqual(caller, ENVIRONMENT)
      for directory, arguments, environment in git.calls:
         self.assertFalse(any(TOKEN in argument or encoded in argument for argument in arguments), "the credential must never be an argument or part of a URL")
         self.assertNotIn("CI_CHECKOUT_TOKEN", environment)
         if arguments[0] == "fetch":
            carried = [key for key, value in environment.items() if TOKEN in value or encoded in value]
            self.assertEqual(carried, ["GIT_CONFIG_VALUE_0"])
            self.assertEqual(environment["GIT_CONFIG_NOSYSTEM"], "1")
            # the header is scoped to the trusted forge's origin, and the fetch is configured not to follow redirects (configuration only, not behavior)
            self.assertEqual(environment["GIT_CONFIG_COUNT"], "2")
            self.assertEqual(environment["GIT_CONFIG_KEY_0"], f"http.{TRUSTED_FORGE}/.extraheader")
            self.assertEqual((environment["GIT_CONFIG_KEY_1"], environment["GIT_CONFIG_VALUE_1"]), ("http.followRedirects", "false"))
         else:
            self.assertFalse([key for key in environment if key.startswith(("GIT_CONFIG_KEY", "GIT_CONFIG_VALUE"))], arguments[0])
            self.assertFalse([value for value in environment.values() if TOKEN in value or encoded in value], arguments[0])
         self.assertNotIn(TOKEN, (directory / ".git" / "config").read_text(encoding="utf-8"))

   def test_a_credential_that_reaches_the_repository_configuration_fails_the_step_without_echoing_it(self) -> None:
      """If anything wrote the token or an Authorization header into `.git/config` the step fails, and the error does not repeat the token."""
      for leaked in ("[http]\n\textraheader = Authorization: Basic xyz\n", f"[remote]\n\turl = https://{TOKEN}@host/x.git\n"):
         with self.subTest(leaked=leaked[:20]):
            self.setUp()
            with self.assertRaisesRegex(DependencyCheckoutError, "reached the checkout configuration") as caught:
               self.run_checkout(_target(content=_checkout()), FakeGit(config_text=leaked))
            self.assertNotIn(TOKEN, str(caught.exception))


class UnreadableConfigurationTest(_CheckoutCase):
   """If the repository configuration cannot be read the step fails with a redacted class name, not a raw OS error."""

   def test_an_unreadable_git_config_is_a_checkout_error(self) -> None:
      """A runner that leaves no `.git/config` (here: a directory in its place) yields DependencyCheckoutError naming only the error class."""

      def git(directory: Path, arguments: tuple[str, ...], _environment: Mapping[str, str]) -> bytes:
         if arguments[0] == "init":
            (directory / ".git" / "config").mkdir(parents=True)
         return f"{PINNED}\n".encode() if arguments[0] == "rev-parse" else b""

      with self.assertRaisesRegex(DependencyCheckoutError, "cannot be read: ") as caught:
         checkout_dependencies(_target(content=_checkout()), self.root, ENVIRONMENT, BINDINGS, git)
      self.assertNotIn(str(self.root), str(caught.exception))
      self.assertNotIn(TOKEN, str(caught.exception))


class RedactionTest(unittest.TestCase):
   """What the real Git runner surfaces: a fixed message, never Git's own output, which can echo the remote or a header."""

   def test_a_failed_git_command_surfaces_no_stderr(self) -> None:
      """The child's stderr (with a URL and a token in it) is dropped; only a generic message is raised."""
      failed = subprocess.CompletedProcess(args=[], returncode=128, stdout=b"", stderr=f"fatal: could not read from https://{TOKEN}@git.example/x.git".encode())
      with mock.patch("platform_ci.dependency_checkout.subprocess.run", return_value=failed), self.assertRaises(DependencyCheckoutError) as caught:
         run_git(Path.cwd(), ("fetch",), {})
      self.assertNotIn(TOKEN, str(caught.exception))
      self.assertNotIn("git.example", str(caught.exception))

   def test_a_start_failure_or_timeout_reports_only_the_error_type(self) -> None:
      """An OS error or a timeout whose text could hold paths or arguments is reduced to its class name."""
      for error in (OSError(f"cannot run with {TOKEN}"), subprocess.TimeoutExpired(cmd=["git", TOKEN], timeout=1)):
         with self.subTest(error=type(error).__name__), mock.patch("platform_ci.dependency_checkout.subprocess.run", side_effect=error), self.assertRaises(DependencyCheckoutError) as caught:
            run_git(Path.cwd(), ("fetch",), {})
         self.assertNotIn(TOKEN, str(caught.exception))
         self.assertIn(type(error).__name__, str(caught.exception))

   def test_git_runs_with_hooks_off_and_exactly_the_given_environment(self) -> None:
      """The real runner disables hooks and passes the supplied environment, not the process's own."""
      done = subprocess.CompletedProcess(args=[], returncode=0, stdout=b"ok", stderr=b"")
      with mock.patch("platform_ci.dependency_checkout.subprocess.run", return_value=done) as runner:
         self.assertEqual(run_git(Path.cwd(), ("rev-parse", "HEAD"), {"ONLY": "this"}), b"ok")
      command = runner.call_args.args[0]
      self.assertTrue(any(part.startswith("core.hooksPath=") for part in command))
      self.assertEqual(runner.call_args.kwargs["env"], {"ONLY": "this"})


class BindingsManifestTest(unittest.TestCase):
   """The operator manifest authorizes only what it names, and a bad or missing one is an error, not an empty pass."""

   @override
   def setUp(self) -> None:
      """Create a directory for manifests."""
      scratch = tempfile.TemporaryDirectory()
      self.addCleanup(scratch.cleanup)
      self.directory = Path(scratch.name)

   def write(self, text: str) -> Path:
      """Write a manifest file and return its path."""
      path = self.directory / "bindings.yaml"
      path.write_text(text, encoding="utf-8")
      return path

   def test_a_valid_manifest_yields_sorted_bindings(self) -> None:
      """The binding keeps its secret slot, consumers and readable repositories."""
      path = self.write(f"schema: 1\nbindings:\n  content-read:\n    secret: {CHECKOUT_SECRET_NAME}\n    consumers: [Org/b, Org/a]\n    repositories: [Org/content]\n")
      binding = load_bindings(path)["content-read"]
      self.assertEqual((binding.consumers, binding.repositories), (("Org/a", "Org/b"), ("Org/content",)))

   def test_malformed_manifests_are_refused(self) -> None:
      """Wrong secret slot, an extra field, an empty list, a non owner/name entry, a wrong schema or a missing file all fail."""
      header = "schema: 1\nbindings:\n  content-read:\n"
      bad = {
         "wrong secret": f"{header}    secret: SOME_OTHER_SECRET\n    consumers: [Org/a]\n    repositories: [Org/c]\n",
         "extra field": f"{header}    secret: {CHECKOUT_SECRET_NAME}\n    consumers: [Org/a]\n    repositories: [Org/c]\n    admin: true\n",
         "empty consumers": f"{header}    secret: {CHECKOUT_SECRET_NAME}\n    consumers: []\n    repositories: [Org/c]\n",
         "not owner/name": f"{header}    secret: {CHECKOUT_SECRET_NAME}\n    consumers: [consumer]\n    repositories: [Org/c]\n",
         "wrong schema": "schema: 2\nbindings: {}\n",
      }
      for label, text in bad.items():
         with self.subTest(label), self.assertRaises(DependencyCheckoutError):
            load_bindings(self.write(text))
      with self.assertRaises(DependencyCheckoutError):
         load_bindings(self.directory / "missing.yaml")


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