from __future__ import annotations import hashlib import io import json import os import shutil import subprocess import sys import tarfile import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BUNDLE = ROOT / "evidence/WP-400/database-image-candidate-20261003" SCAN_EVIDENCE = ROOT / "evidence/WP-400/DATABASE_REMEDIATION_SCANS.json" PUBLICATION_EVIDENCE = ROOT / "evidence/WP-400/DATABASE_CANDIDATE_PUBLICATION_20261003.json" def read_json(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def bundle_files(bundle: Path) -> set[str]: return { path.relative_to(bundle).as_posix() for path in bundle.rglob("*") if path.is_file() and path != bundle / "FILE_HASHES.json" } def run_git(cwd: Path, *arguments: str) -> bytes: result = subprocess.run( # noqa: S603 ["git", *arguments], # noqa: S607 cwd=cwd, check=True, capture_output=True, ) return result.stdout class DatabaseCandidateProvenanceTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.manifest = read_json(BUNDLE / "candidate-provenance.json") cls.scan_evidence = read_json(SCAN_EVIDENCE) cls.publication = read_json(PUBLICATION_EVIDENCE) def test_retained_recipe_sources_match_the_exact_build_recipe_hashes(self) -> None: for service in ("postgres", "redis"): with self.subTest(service=service): item = self.manifest["images"][service] build = read_json(BUNDLE / item["buildReceiptFile"]) self.assertEqual(set(item["sourceFiles"]), set(build["recipeHashes"])) plan = ( read_json(BUNDLE / "recipes" / service / "plan.json") if service == "postgres" else None ) for name, expected in item["sourceFiles"].items(): source = BUNDLE / "recipes" / service / name if name.endswith(".tar.gz"): self.assertFalse( source.exists(), "the binary upstream archive must stay excluded" ) self.assertFalse( self.manifest["trackedBundle"]["binaryGosuSourceArchiveIncluded"] ) self.assertEqual(expected, plan["source"]["archive_sha256"]) else: self.assertTrue(source.is_file(), name) self.assertEqual(sha256(source), expected, name) self.assertEqual(build["recipeHashes"][name], expected, name) def test_build_export_scan_and_publication_bind_the_same_candidates(self) -> None: self.assertFalse(self.manifest["releaseEligible"]) self.assertFalse(self.manifest["reviewBoundary"]["releaseEligible"]) self.assertEqual(self.manifest["reviewBoundary"]["humanCveRiskAcceptance"], "pending") self.assertFalse(self.manifest["reviewBoundary"]["agentMayApproveCveExceptions"]) self.assertFalse(self.manifest["reviewBoundary"]["bitForBitRebuildClaim"]) self.assertFalse(self.scan_evidence["productionDeployment"]) self.assertFalse(self.publication["productionDeployment"]) self.assertFalse(self.publication["cveRiskAccepted"]) for service in ("postgres", "redis"): with self.subTest(service=service): item = self.manifest["images"][service] build = read_json(BUNDLE / item["buildReceiptFile"]) download = read_json(BUNDLE / item["downloadReceiptFile"]) scan = read_json(BUNDLE / item["scanReceiptFile"]) summary = read_json(BUNDLE / item["vulnerabilityScan"]["summaryFile"]) secret = read_json(BUNDLE / item["secretScan"]["sanitizedReceiptFile"]) image = build["image"] published_ref = self.manifest["publishedImageRefs"][service] published = self.publication["candidates"][service] scanned = self.scan_evidence["images"][service] self.assertEqual(item["publishedImageRef"], published_ref) self.assertIsNone(item["productionImageRef"]) self.assertEqual(item["candidateOnly"], True) self.assertIsNone(item["productionImageRef"]) self.assertEqual(published["reference"], published_ref) self.assertEqual(scanned["artifact"]["publishedImage"]["reference"], published_ref) self.assertEqual(published["scannedImageId"], image["id"]) self.assertEqual(published["pulledImageId"], image["id"]) self.assertTrue(published["immutablePullConfigMatchedScannedImage"]) self.assertEqual(item["vulnerabilityScan"]["artifactRef"]["imageId"], image["id"]) self.assertEqual( item["vulnerabilityScan"]["rawReportSha256"], scan["rawVulnerabilityReportSha256"], ) self.assertEqual(download["artifactRef"]["imageId"], image["id"]) self.assertEqual(download["artifactRef"]["tag"], image["tag"]) self.assertEqual(download["localSha256"], build["archive"]["sha256"]) self.assertEqual(download["remoteSha256"], build["archive"]["sha256"]) self.assertEqual(scan["artifactRef"]["imageId"], image["id"]) self.assertEqual(scan["artifactRef"]["archiveSha256"], build["archive"]["sha256"]) self.assertEqual(scan["reportIdentity"]["ImageID"], image["id"]) self.assertEqual(scan["ignoreUnfixed"], False) self.assertEqual(scan["findingFilters"], []) self.assertEqual(scan["tlsVerification"], True) self.assertEqual(scan["packageCount"], summary["packageCount"]) self.assertGreater(summary["packageCount"], 0) inventory = summary["strictVersionInventory"] self.assertTrue(inventory["allNamesAndVersionsPresent"]) self.assertEqual(inventory["packageEntriesChecked"], summary["packageCount"]) self.assertEqual(secret["ImageID"], image["id"]) self.assertEqual(secret["FindingCount"], 0) self.assertFalse(secret["SensitiveFindingDetailsPersisted"]) self.assertEqual(scanned["humanRiskAcceptance"], "pending") self.assertFalse(scanned["productionDeployment"]) def test_bundle_hash_manifest_covers_retained_recipe_and_receipt_files(self) -> None: evidence = self.scan_evidence["recipeProvenance"] self.assertEqual( evidence["manifestPath"], "evidence/WP-400/database-image-candidate-20261003/candidate-provenance.json", ) self.assertEqual(evidence["manifestSha256"], sha256(BUNDLE / "candidate-provenance.json")) self.assertEqual( evidence["fileHashesPath"], "evidence/WP-400/database-image-candidate-20261003/FILE_HASHES.json", ) hashes_path = BUNDLE / "FILE_HASHES.json" self.assertEqual(evidence["fileHashesSha256"], sha256(hashes_path)) hashes = read_json(hashes_path) actual_files = bundle_files(BUNDLE) self.assertEqual(set(hashes["files"]), actual_files) for relative, expected in hashes["files"].items(): with self.subTest(path=relative): self.assertEqual(sha256(BUNDLE / relative), expected) def test_ci_compilation_preserves_strict_bundle_inventory(self) -> None: with tempfile.TemporaryDirectory(prefix="wp400-compiled-provenance-") as temporary: bundle = Path(temporary) / "evidence" recipe = bundle / "recipes" / "test_recipe_contract.py" recipe.parent.mkdir(parents=True) recipe.write_text("assert True\n", encoding="utf-8") source_hash = sha256(recipe) cache_root = Path(temporary) / "managed-bytecode" subprocess.run( # noqa: S603 [sys.executable, "-m", "py_compile", str(recipe)], env=dict(os.environ, PYTHONPYCACHEPREFIX=str(cache_root)), check=True, capture_output=True, ) self.assertTrue(any(cache_root.rglob("*.pyc"))) self.assertEqual(sha256(recipe), source_hash) self.assertEqual(bundle_files(bundle), {"recipes/test_recipe_contract.py"}) unexpected = recipe.parent / "__pycache__" / "unexpected.json" unexpected.parent.mkdir() unexpected.write_text("{}\n", encoding="utf-8") matching_cache = unexpected.with_name( f"test_recipe_contract.{sys.implementation.cache_tag}.pyc" ) matching_cache.write_bytes(b"unexpected cache bytes") self.assertEqual( bundle_files(bundle), { "recipes/test_recipe_contract.py", "recipes/__pycache__/unexpected.json", f"recipes/__pycache__/{matching_cache.name}", }, ) workflow = (ROOT / ".gitea/workflows/managed-validation.yml").read_text(encoding="utf-8") self.assertIn( 'PYTHONPYCACHEPREFIX="${RUNNER_TEMP}/managed-bytecode" ' "xargs -0 -r python -m py_compile", workflow, ) def test_git_objects_archives_and_autocrlf_checkouts_preserve_bundle_hashes(self) -> None: file_hashes = read_json(BUNDLE / "FILE_HASHES.json")["files"] expected_hashes = dict(file_hashes) expected_hashes["FILE_HASHES.json"] = self.scan_evidence["recipeProvenance"][ "fileHashesSha256" ] bundle_path = Path("evidence/WP-400/database-image-candidate-20261003") with tempfile.TemporaryDirectory(prefix="wp400-git-byte-provenance-") as temporary: temporary_root = Path(temporary) source = temporary_root / "source" source.mkdir() shutil.copyfile(ROOT / ".gitattributes", source / ".gitattributes") for relative in expected_hashes: destination = source / bundle_path / relative destination.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(BUNDLE / relative, destination) run_git(source, "init", "--quiet") run_git(source, "config", "user.name", "WP-400 provenance test") run_git(source, "config", "user.email", "wp400-provenance@example.invalid") run_git(source, "config", "core.autocrlf", "true") run_git(source, "config", "core.safecrlf", "false") hooks = source / ".test-hooks" hooks.mkdir() run_git(source, "config", "core.hooksPath", ".test-hooks") run_git(source, "add", "--all") run_git(source, "commit", "--quiet", "-m", "byte-preserving provenance fixture") archive = run_git(source, "archive", "--format=tar", "HEAD", str(bundle_path)) archive_prefix = bundle_path.as_posix() + "/" with tarfile.open(fileobj=io.BytesIO(archive), mode="r:") as archived: archived_files = { member.name.removeprefix(archive_prefix): member for member in archived.getmembers() if member.isfile() and member.name.startswith(archive_prefix) } self.assertEqual(set(archived_files), set(expected_hashes)) for relative, expected in expected_hashes.items(): member_file = archived.extractfile(archived_files[relative]) self.assertIsNotNone(member_file) self.assertEqual( hashlib.sha256(member_file.read()).hexdigest(), expected, relative ) for autocrlf in ("false", "true"): with self.subTest(core_autocrlf=autocrlf): checkout = temporary_root / f"checkout-{autocrlf}" run_git( temporary_root, "clone", "--quiet", "--no-checkout", str(source), str(checkout), ) run_git(checkout, "config", "core.autocrlf", autocrlf) run_git(checkout, "config", "core.safecrlf", "false") run_git(checkout, "checkout", "--force", "HEAD") for relative, expected in expected_hashes.items(): self.assertEqual(sha256(checkout / bundle_path / relative), expected) if __name__ == "__main__": unittest.main()