platform-ci/tests/test_workflow_identity.py

"""Reuse trust: a source run counts only if the forge shows it was executed by the canonical pinned workflow, never because its bundles verify.

A manifest, its hashes and the job names are all build output a custom workflow can fabricate. Trust therefore comes from the caller file the run
names, read through the forge at the run's own commit: a trigger-only caller whose single job uses a full-SHA reference the current run also uses.
"""

import tempfile
import unittest
from pathlib import Path
from typing import override

from platform_ci.delivery.errors import DeliveryError, ReuseError
from platform_ci.delivery.forge import RepositoryScope, caller_reference, inspect_run
from tests.synthetic_services import CALLER_TEXT, CANONICAL_REFERENCE, CUSTOM_CALLER_TEXT, REPOSITORY, REPOSITORY_ID, SyntheticForge, SyntheticRepository
from tests.test_reuse_lifecycle import LifecycleTest

USES = f"    uses: {CANONICAL_REFERENCE}\n"


def _caller(job: str = USES, top: str = "") -> str:
   """A caller workflow with the given job body and extra top-level text."""
   return f"name: Q\non:\n  push:\npermissions:\n  contents: read\n{top}jobs:\n  pipeline:\n{job}"


class CallerIdentityTest(unittest.TestCase):
   """Only a trigger-only caller of a full-SHA scoped workflow reference yields a reference; every other shape yields none."""

   def _reference(self, caller_text: str | None, path: str = "qualification.yaml@refs/heads/main") -> str | None:
      with tempfile.TemporaryDirectory() as directory:
         repository = SyntheticRepository(Path(directory) / "repo")
         changes = {"README.md": "x"} | ({} if caller_text is None else {".gitea/workflows/qualification.yaml": caller_text})
         sha = repository.commit(changes)
         return caller_reference(SyntheticForge(repository), REPOSITORY, {"id": 1, "path": path}, sha)

   def test_the_canonical_caller_yields_its_reference(self) -> None:
      self.assertEqual(self._reference(_caller()), CANONICAL_REFERENCE)

   def test_every_other_shape_yields_none(self) -> None:
      cases = {
         "steps instead of uses": _caller("    runs-on: linux-small\n    steps:\n      - run: echo hi\n"),
         "uses plus steps": _caller(USES + "    steps: []\n"),
         "uses plus with": _caller(USES + "    with: {a: b}\n"),
         "uses plus secrets": _caller(USES + "    secrets: inherit\n"),
         "unpinned branch reference": _caller("    uses: self:KofTwentyTwo/platform-ci/.gitea/scoped_workflows/pipeline.yaml@main\n"),
         "short sha": _caller("    uses: self:KofTwentyTwo/platform-ci/.gitea/scoped_workflows/pipeline.yaml@4444444\n"),
         "local path": _caller("    uses: ./.gitea/workflows/other.yaml\n"),
         "another hosting scheme": _caller("    uses: https://example.invalid/o/r/.gitea/scoped_workflows/pipeline.yaml@" + "4" * 40 + "\n"),
         "extra top level env": _caller(top="env:\n  A: b\n"),
         "second job": _caller() + "  second:\n    runs-on: linux-small\n    steps: []\n",
         "duplicate keys": _caller() + "  pipeline:\n" + USES,
         "not a mapping": "- a\n- b\n",
         "no file": None,
      }
      for name, text in cases.items():
         with self.subTest(name):
            self.assertIsNone(self._reference(text))

   def test_a_run_path_cannot_name_another_directory_or_a_non_workflow(self) -> None:
      for path in ("../qualification.yaml@refs/heads/main", "sub/qualification.yaml@refs/heads/main", "qualification.txt@refs/heads/main", "@refs/heads/main", ""):
         with self.subTest(path):
            self.assertIsNone(self._reference(_caller(), path))

   def test_an_empty_scope_trusts_no_run(self) -> None:
      with tempfile.TemporaryDirectory() as directory:
         repository = SyntheticRepository(Path(directory) / "repo")
         sha = repository.commit({".gitea/workflows/unified-qualification.yaml": CALLER_TEXT})
         forge = SyntheticForge(repository)
         forge.add_run(7, sha)
         untrusted = RepositoryScope(REPOSITORY, REPOSITORY_ID, 1000, 10, frozenset({"b" * 40}))
         with self.assertRaisesRegex(DeliveryError, "trusted canonical workflow reference"):
            inspect_run(forge, untrusted, 7)
         trusted = RepositoryScope(REPOSITORY, REPOSITORY_ID, 1000, 10, frozenset({"b" * 40}), frozenset({CANONICAL_REFERENCE}))
         self.assertEqual(inspect_run(forge, trusted, 7).conclusion, "success")


class FabricatedWorkflowTest(LifecycleTest):
   """A custom workflow that uploads valid bundles under the canonical job names is never trusted; the build runs instead."""

   CUSTOM = ".gitea/workflows/custom.yaml"

   @override
   def setUp(self) -> None:
      """Commit a tree holding both the canonical caller and a custom workflow file."""
      super().setUp()
      self.head = self.commit({self.CUSTOM: CUSTOM_CALLER_TEXT})

   def test_bundles_of_a_custom_workflow_run_are_not_reused_and_the_merge_builds_once(self) -> None:
      fabricated = self.pipeline(100, caller="custom.yaml")
      self.assertEqual((fabricated.builds, fabricated.evidence), (1, ""))
      self.assertIn("does not prove this run was executed by the canonical workflow", fabricated.note)
      self.repository.commit({}, "merge commit with no content change")
      merged = self.pipeline(101)
      self.assertEqual((merged.builds, merged.packages, merged.report_code, merged.evidence), (1, 1, 0, ""))
      self.assertIn("not executed by a trusted canonical workflow reference", merged.note)
      self.assertIn("required once", merged.report_text)

   def test_the_publisher_refuses_the_custom_workflows_bundles_and_publishes_nothing(self) -> None:
      self.pipeline(100, caller="custom.yaml")
      self.repository.commit({}, "merge commit with no content change")
      with self.assertRaisesRegex(ReuseError, "not executed by a trusted canonical workflow reference"):
         self.publish()
      self.assertFalse(self.world.written_anything())

   def test_a_canonical_looking_caller_pinned_to_another_implementation_is_not_trusted(self) -> None:
      self.commit({".gitea/workflows/unified-qualification.yaml": CALLER_TEXT.replace("4" * 40, "5" * 40)})
      self.pipeline(100)
      self.repository.commit({}, "same tree")
      with self.assertRaisesRegex(ReuseError, "not executed by a trusted canonical workflow reference"):
         self.publish()