platform-ci/tests/test_secret_scan.py

"""History secret scan with a fake command runner: exact revision, full ancestry, sanitized environment, no leaked output."""

import json
import os
import re
import sys
import tempfile
import unittest
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import override

from platform_ci.secret_scan import (
   GITLEAKS_GO_IMAGE,
   GITLEAKS_INSTALL_PATH,
   GITLEAKS_SOURCE_COMMIT,
   GITLEAKS_SOURCE_SHA256,
   GITLEAKS_TEXT_VERSION,
   GITLEAKS_VERSION,
   CommandOutput,
   SecretScanError,
   SecretScanResult,
   run_scan_command,
   scan_history,
)
from tests.support import SOURCE_REVISION

PARENT_ENVIRONMENT = {
   "PATH": "/usr/bin",
   "HOME": "/home/runner",
   "GITEA_TOKEN": "release-secret",
   "GIT_DIR": "/elsewhere/.git",
   "GITLEAKS_CONFIG": "/elsewhere/gitleaks.toml",
   "GIT_CONFIG_COUNT": "9",
}
REPORT = Path("ci-reports/gitleaks.json")
DOCKERFILE = (Path(__file__).resolve().parent.parent / "runtime" / "security.Dockerfile").read_text(encoding="utf-8")


class FakeGit:
   """A scripted runner: answers Git and Gitleaks commands, writes the report like Gitleaks, and records every call."""

   def __init__(
      self,
      head: str = SOURCE_REVISION,
      shallow: str = "false",
      version: str = GITLEAKS_VERSION,
      scan_exit_code: int = 0,
      report: str | None = "[]",
   ) -> None:
      """Script the answers; a `None` report means Gitleaks writes no report file."""
      self.head = head
      self.shallow = shallow
      self.version = version
      self.scan_exit_code = scan_exit_code
      self.report = report
      self.calls: list[tuple[list[str], Path, Mapping[str, str]]] = []

   def __call__(self, command: Sequence[str], working_directory: Path, environment: Mapping[str, str]) -> CommandOutput:
      """Record the call and answer it."""
      self.calls.append((list(command), working_directory, environment))
      if "--is-shallow-repository" in command:
         return CommandOutput(0, self.shallow + "\n")
      if "rev-parse" in command:
         return CommandOutput(0, self.head + "\n")
      if command[1:] == ["version"]:
         return CommandOutput(0, self.version + "\n")
      if self.report is not None:
         report_path = Path(command[command.index("--report-path") + 1])
         report_path.write_text(self.report, encoding="utf-8")
      return CommandOutput(self.scan_exit_code, "RAW-MATCHED-SECRET-IN-OUTPUT")

   def scan_calls(self) -> list[tuple[list[str], Path, Mapping[str, str]]]:
      """Return only the Gitleaks scan invocations."""
      return [call for call in self.calls if "detect" in call[0]]


class SecretScanTest(unittest.TestCase):
   """Every precondition and tool problem fails; only a clean, complete, exact scan passes."""

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

   def _scan(self, fake: FakeGit, revision: str = SOURCE_REVISION, report: Path = REPORT) -> SecretScanResult:
      return scan_history(self.source, revision, report, fake, PARENT_ENVIRONMENT)

   def test_clean_scan_passes_with_report_metadata(self) -> None:
      """A matching, complete checkout with no findings passes and records the report and version."""
      result = self._scan(FakeGit())
      self.assertTrue(result.passed)
      self.assertIsNone(result.failure)
      self.assertEqual((result.findings, result.exit_code, result.gitleaks_version), (0, 0, GITLEAKS_VERSION))
      self.assertEqual(result.report_path, self.source / REPORT)
      self.assertEqual(result.source_revision, SOURCE_REVISION)

   def test_scan_arguments_cover_exactly_the_source_history_with_full_redaction(self) -> None:
      """Only `--full-history <sha>` is scanned, redacted fully, as a JSON report, with no ignore or download option."""
      fake = FakeGit()
      self._scan(fake)
      ((command, working_directory, _environment),) = fake.scan_calls()
      self.assertEqual(command[:2], [GITLEAKS_INSTALL_PATH, "detect"])
      self.assertIn(f"--log-opts=--full-history {SOURCE_REVISION}", command)
      self.assertIn("--redact=100", command)
      self.assertEqual(command[command.index("--report-format") + 1], "json")
      self.assertEqual(command[command.index("--source") + 1], str(self.source))
      self.assertEqual(working_directory, self.source)
      self.assertFalse([token for token in command if token in {"--all", "--no-git", "--verbose", "--gitleaks-ignore-path", "--baseline-path"}])

   def test_environment_is_sanitized_and_safe_directory_is_scoped_to_the_exact_source(self) -> None:
      """Credentials and Git or Gitleaks overrides are dropped; Git is told only about this source directory."""
      fake = FakeGit()
      self._scan(fake)
      for command, _directory, environment in fake.calls:
         self.assertNotIn("GITEA_TOKEN", environment)
         self.assertNotIn("HOME", environment)
         self.assertNotIn("GIT_DIR", environment)
         self.assertNotIn("GITLEAKS_CONFIG", environment)
         self.assertEqual(environment["PATH"], "/usr/bin")
         self.assertEqual(environment["GIT_CONFIG_NOSYSTEM"], "1")
         self.assertEqual(environment["GIT_NO_REPLACE_OBJECTS"], "1")
         self.assertEqual(
            (environment["GIT_CONFIG_COUNT"], environment["GIT_CONFIG_KEY_0"], environment["GIT_CONFIG_VALUE_0"]),
            ("1", "safe.directory", str(self.source)),
         )
         if command[0] == "git":
            self.assertEqual(command[1:3], ["-c", f"safe.directory={self.source}"])
         self.assertNotIn("*", environment["GIT_CONFIG_VALUE_0"])

   def test_head_other_than_source_revision_fails_before_scanning(self) -> None:
      """A checkout of another commit is refused and Gitleaks never runs."""
      fake = FakeGit(head="c" * 40)
      result = self._scan(fake)
      self.assertFalse(result.passed)
      self.assertIn("HEAD is not the source revision", result.failure or "")
      self.assertEqual(fake.scan_calls(), [])

   def test_shallow_or_unverifiable_clone_fails_before_scanning(self) -> None:
      """Anything but an explicit `false` from Git is refused, because a partial ancestry would pass silently."""
      for answer in ("true", "", "unknown"):
         with self.subTest(answer=answer):
            fake = FakeGit(shallow=answer)
            result = self._scan(fake)
            self.assertFalse(result.passed)
            self.assertIn("shallow", result.failure or "")
            self.assertEqual(fake.scan_calls(), [])

   def test_source_revision_must_be_an_exact_lowercase_40_hex_hash(self) -> None:
      """Branch names, short, uppercase, padded and 64-digit values fail without running anything."""
      for revision in ("main", "a" * 7, "A" * 40, "a" * 41, "a" * 64, f"{SOURCE_REVISION}\n", "", "--all"):
         with self.subTest(revision=revision):
            fake = FakeGit()
            self.assertFalse(self._scan(fake, revision).passed)
            self.assertEqual(fake.calls, [])

   def test_only_the_installed_binary_with_the_pinned_version_is_run(self) -> None:
      """Every command uses the fixed install path; a different version fails before scanning, nothing is downloaded."""
      fake = FakeGit()
      self._scan(fake)
      self.assertEqual({call[0][0] for call in fake.calls}, {"git", GITLEAKS_INSTALL_PATH})
      wrong = FakeGit(version="8.21.1")
      result = self._scan(wrong)
      self.assertFalse(result.passed)
      self.assertEqual(result.gitleaks_version, "8.21.1")
      self.assertEqual(wrong.scan_calls(), [])
      self.assertTrue(self._scan(FakeGit(version=f"v{GITLEAKS_VERSION}")).passed)

   def test_any_finding_fails_and_raw_output_never_reaches_the_result_or_logs(self) -> None:
      """Findings fail with a count; neither the redacted report nor the tool output is copied into the failure."""
      report = json.dumps([{"RuleID": "generic", "Secret": "REDACTED", "Match": "REDACTED"}, {"RuleID": "aws", "Secret": "REDACTED"}])
      with self.assertLogs("platform_ci.secret_scan", level="INFO") as logs:
         result = self._scan(FakeGit(scan_exit_code=2, report=report))
      self.assertFalse(result.passed)
      self.assertEqual((result.findings, result.exit_code), (2, 2))
      self.assertEqual(result.report_path, self.source / REPORT)
      visible = f"{result.failure} {logs.output}"
      self.assertNotIn("RAW-MATCHED-SECRET", visible)
      self.assertNotIn("generic", visible)

   def test_findings_fail_even_when_the_tool_exits_zero(self) -> None:
      """A report with findings is authoritative over a zero exit code."""
      self.assertFalse(self._scan(FakeGit(report='[{"RuleID": "x"}]')).passed)

   def test_tool_errors_fail(self) -> None:
      """A tool error, a missing report, a malformed report and a non-list report are all failures."""
      cases = {
         "tool error with report": FakeGit(scan_exit_code=1),
         "tool error without report": FakeGit(scan_exit_code=1, report=None),
         "no report on exit zero": FakeGit(report=None),
         "malformed report": FakeGit(report="{not json"),
         "object report": FakeGit(report="{}"),
      }
      for description, fake in cases.items():
         with self.subTest(description):
            result = self._scan(fake)
            self.assertFalse(result.passed)
            self.assertTrue(result.failure)

   def test_stale_report_cannot_stand_in_for_a_failed_run(self) -> None:
      """A clean report left by an earlier run is removed before scanning, so a run that writes nothing fails."""
      stale = self.source / REPORT
      stale.parent.mkdir()
      stale.write_text("[]", encoding="utf-8")
      self.assertFalse(self._scan(FakeGit(report=None)).passed)
      self.assertFalse(stale.exists())

   def test_report_path_must_stay_inside_the_checkout_and_away_from_git_metadata(self) -> None:
      """Escaping, Git-internal and platform-checkout report paths fail without scanning."""
      for report in (Path("../outside.json"), Path(".git/report.json"), Path(".platform-ci/report.json"), Path()):
         with self.subTest(report=str(report)):
            fake = FakeGit()
            self.assertFalse(self._scan(fake, report=report).passed)
            self.assertEqual(fake.scan_calls(), [])

   def test_runner_that_cannot_start_a_command_fails_the_scan(self) -> None:
      """A start failure is reported as a failed result rather than an exception."""

      def unstartable(command: Sequence[str], working_directory: Path, environment: Mapping[str, str]) -> CommandOutput:
         raise SecretScanError(f"cannot start {command[0]!r}")

      result = scan_history(self.source, SOURCE_REVISION, REPORT, unstartable, PARENT_ENVIRONMENT)
      self.assertFalse(result.passed)
      self.assertIn("cannot start", result.failure or "")

   def test_default_runner_captures_output_and_reports_missing_executables(self) -> None:
      """The real runner returns the exit code and standard output and fails visibly for an absent program."""
      output = run_scan_command([sys.executable, "-c", "import sys; print('out'); print('err', file=sys.stderr); raise SystemExit(3)"], self.source, dict(os.environ))
      self.assertEqual(output, CommandOutput(3, "out\n"))
      with self.assertRaises(SecretScanError):
         run_scan_command(["definitely-not-an-installed-scanner"], self.source, {})


class SecurityRuntimeRecipeTest(unittest.TestCase):
   """The image recipe rebuilds the pinned Gitleaks source with the patched Go and x/text at build time and ships only the binary."""

   def test_pins_match_the_scanner_constants_and_are_immutable_references(self) -> None:
      """The checked-in Dockerfile args equal the scanner constants and are full hashes or digests; no operator file is read."""
      for name, value in {
         "GITLEAKS_VERSION": GITLEAKS_VERSION,
         "GITLEAKS_SOURCE_COMMIT": GITLEAKS_SOURCE_COMMIT,
         "GITLEAKS_SOURCE_SHA256": GITLEAKS_SOURCE_SHA256,
         "GO_IMAGE": GITLEAKS_GO_IMAGE,
         "TEXT_VERSION": GITLEAKS_TEXT_VERSION,
      }.items():
         self.assertIn(f"ARG {name}={value}\n", DOCKERFILE)
      self.assertRegex(GITLEAKS_SOURCE_COMMIT, r"[0-9a-f]{40}")
      self.assertRegex(GITLEAKS_SOURCE_SHA256, r"[0-9a-f]{64}")
      self.assertRegex(GITLEAKS_GO_IMAGE, r".+@sha256:[0-9a-f]{64}")
      self.assertRegex(GITLEAKS_TEXT_VERSION, r"v\d+\.\d+\.\d+")

   def test_the_legacy_prebuilt_binary_checksum_is_no_longer_a_build_input(self) -> None:
      self.assertNotIn("5bc41815076e6ed6ef8fbecc9d9b75bcae31f39029ceb55da08086315316e3ba", DOCKERFILE)
      self.assertNotIn("linux_x64.tar.gz", DOCKERFILE)

   def test_binary_is_built_from_source_with_the_local_toolchain_and_patched_text(self) -> None:
      build_stage = DOCKERFILE.split("AS gitleaks-build", 1)[1].split("# Runtime stage", 1)[0]
      self.assertIn("ENV GOTOOLCHAIN=local", build_stage)
      self.assertIn("ENV CGO_ENABLED=0", build_stage)
      self.assertIn('go get "golang.org/x/text@${TEXT_VERSION}"', build_stage)
      self.assertIn("go build -trimpath", build_stage)
      self.assertIn("-s -w -X ${module}/cmd.Version=${GITLEAKS_VERSION}", build_stage)
      self.assertRegex(DOCKERFILE, r"(?m)^FROM \$\{GO_IMAGE\} AS gitleaks-build$")
      self.assertNotIn("urllib", build_stage)

   def test_runtime_derives_the_published_control_image_by_digest(self) -> None:
      self.assertRegex(DOCKERFILE, r"ARG CONTROL_IMAGE=\S+platform-ci-control@sha256:11f8cb328e7fc7aca8e0c39ed979b33cb1db411a59460baae3e0d48281a15e3c\n")
      self.assertRegex(DOCKERFILE, r"(?m)^FROM \$\{CONTROL_IMAGE\}")

   def test_download_is_build_time_only_and_the_runtime_stage_copies_just_the_binary(self) -> None:
      source_stage = DOCKERFILE.split("AS gitleaks-source", 1)[1].split("# Build stage", 1)[0]
      final_stage = DOCKERFILE.rsplit("FROM ${CONTROL_IMAGE}", 1)[1]
      self.assertIn("urllib.request.urlopen", source_stage)
      self.assertIn("hashlib.sha256", source_stage)
      self.assertIn('filter="data"', source_stage)
      self.assertNotIn("urllib", final_stage)
      self.assertNotIn("apt-get", final_stage)
      self.assertNotIn("go build", final_stage)
      # The qualified image copies exactly two things into its runtime: the pinned Node binary the artifact-upload action needs
      # (copied from a digest-pinned official Node stage, never downloaded here) and the Gitleaks binary built from verified source.
      self.assertEqual(re.findall(r"(?m)^COPY .*", final_stage), ["COPY --from=actions-node /usr/local/bin/node /usr/local/bin/node", "COPY --from=gitleaks-build /out/gitleaks /usr/local/bin/gitleaks"])
      self.assertRegex(DOCKERFILE, r"(?m)^FROM node:22\.20\.0-bookworm@sha256:[0-9a-f]{64} AS actions-node$")
      self.assertIn('test "$(node --version)" = "v22.20.0"', final_stage)
      self.assertEqual(
         re.findall(r"(?m)^(?:COPY|ADD) .*", DOCKERFILE),
         ["COPY --from=gitleaks-source /src /src", "COPY --from=actions-node /usr/local/bin/node /usr/local/bin/node", "COPY --from=gitleaks-build /out/gitleaks /usr/local/bin/gitleaks"],
      )