Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
11 kB · 241 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242#!/usr/bin/env python3"""Deterministic coherence tests for the prepared source archive.
``tools/package_release.py --prepare-source`` hashes and measures everyallowlisted file once, records the size and SHA-256 in the resolved``source-manifest.json``, and only then writes the source archive. The archivewriter reads each member from the *live* input path, so a file that changes inthe window between hashing and archiving could otherwise slip bytes into thearchive that the manifest never described — and ``--prepare-source`` would stillreport success.
These tests pin the guarantee that closes that window: each member is streamedthrough a reader that hashes exactly the bytes written to the archive andcompares them with the recorded size and digest, so a divergence fails thepreparation before anything is published. They also pin the pack-side rule thatthe archive is verified against the *recorded* prepared metadata, never againstthe current source tree.
The fault is injected with the deterministic ``set_member_read_hook`` seam, sothe race is reproduced every run instead of being timed. The synthetic projectfixture is shared with ``test_release_package`` so the allowlist, privacy rulesand archive contract stay in one place."""
from __future__ import annotations
import contextlibimport hashlibimport ioimport jsonimport sysimport tarfileimport tempfileimport unittestfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"PYTHON_TESTS = Path(__file__).resolve().parentfor entry in (str(TOOLS), str(PYTHON_TESTS)): if entry not in sys.path: sys.path.insert(0, entry)
import package_release as packaging # noqa: E402from test_release_package import PackageTestCase # noqa: E402
def prepare_argv(project) -> list[str]: """The command line for an offline ``--prepare-source`` run of ``project``.""" return [ "--source-root", str(project.root), "--source-manifest", str(project.root / "tools/release_source_manifest.json"), "--output", str(project.output), "--prepare-source", "--vendor-dir", str(project.root.parent / "vendor"), "--development", "synthetic", ]
def output_source_archive(project) -> Path: names = packaging.versioned_names(project.version, "synthetic") return project.output / names["source_archive"]
def run_prepare(project) -> int: """Run ``--prepare-source`` in process, discarding the summary it prints.""" with contextlib.redirect_stdout(io.StringIO()): return packaging.run(prepare_argv(project))
class PrepareSourceCoherenceTests(PackageTestCase): """A file that changes between hashing and archiving must fail the preparation."""
def setUp(self) -> None: super().setUp() # Always clear the hook, even when an assertion fails mid-test. self.addCleanup(packaging.set_member_read_hook, None) self.project_obj = self.project() self.project_obj.write_vendor(self.project_obj.root.parent / "vendor")
def _run_with_fault(self, mutate): """Prepare with a hook that rewrites each input just before it is read.
The hook is called immediately before a file member's bytes are read for the archive, which is exactly the window the fix has to close. It mutates every member it sees, so whichever member is read first is the one whose recorded metadata no longer describes it. """ seen: list[str] = []
def hook(name: str, path: Path) -> None: seen.append(name) path.write_bytes(mutate(path.read_bytes()))
packaging.set_member_read_hook(hook) with self.assertRaises(packaging.CheckFailure) as caught: packaging.run(prepare_argv(self.project_obj)) return seen, caught.exception
def _assert_nothing_published(self) -> None: output = self.project_obj.output # ``prepare_output_directory`` may create the directory, but a failed # preparation must leave it empty: no archive, no manifest, no staging # leftovers. self.assertTrue(output.is_dir()) self.assertEqual(sorted(entry.name for entry in output.iterdir()), []) self.assertFalse(output_source_archive(self.project_obj).exists()) self.assertFalse((output / "source-manifest.json").exists())
def test_same_size_change_is_refused_and_nothing_is_published(self) -> None: # Same byte count, different content: only the digest can catch it. seen, error = self._run_with_fault(lambda data: bytes(byte ^ 0xFF for byte in data)) self.assertTrue(seen, "the read hook was never called") message = str(error) self.assertIn("changed while the archive was being prepared", message) self.assertIn("hashes to", message) self._assert_nothing_published()
def test_grown_file_is_refused_and_nothing_is_published(self) -> None: # The reported failure mode: a file grows while it is being packaged. seen, error = self._run_with_fault(lambda data: data + b"# appended during preparation\n") self.assertTrue(seen, "the read hook was never called") message = str(error) self.assertIn("changed while the archive was being prepared", message) self.assertIn("bytes were recorded when it was hashed", message) self._assert_nothing_published()
def test_truncated_file_is_refused_and_nothing_is_published(self) -> None: seen, error = self._run_with_fault(lambda data: data[:-1]) self.assertTrue(seen, "the read hook was never called") message = str(error) self.assertIn("changed while the archive was being prepared", message) self._assert_nothing_published()
def test_stable_input_publishes_an_archive_that_matches_its_manifest(self) -> None: # Positive control: with no fault the preparation succeeds, and every # archived member's bytes equal the size and digest it recorded. self.assertEqual(run_prepare(self.project_obj), packaging.EXIT_OK) manifest = json.loads((self.project_obj.output / "source-manifest.json").read_text(encoding="utf-8")) archive = output_source_archive(self.project_obj) self.assertTrue(archive.exists()) prefix = manifest["archive"]["prefix"] checked = 0 with packaging.open_bounded_tar(archive, packaging.DEFAULT_LIMITS) as tar: members = {member.name: member for member in tar.getmembers() if member.isreg()} for entry in manifest["files"]: member = members[f"{prefix}/{entry['path']}"] digest = hashlib.sha256(tar.extractfile(member).read()).hexdigest() self.assertEqual(member.size, entry["size"]) self.assertEqual(digest, entry["sha256"]) checked += 1 self.assertGreater(checked, 0) # The packer's own narrow re-verification agrees. result = packaging.verify_prepared_source(archive, manifest, packaging.DEFAULT_LIMITS) self.assertEqual(result["verified_members"], checked + len(manifest["vendor"]["files"]))
def test_verify_uses_recorded_metadata_not_the_current_tree(self) -> None: # A source tree that moves on after preparation must not make the # prepared archive look corrupt: the check is against what the manifest # recorded, not against the live files. self.assertEqual(run_prepare(self.project_obj), packaging.EXIT_OK) manifest = json.loads((self.project_obj.output / "source-manifest.json").read_text(encoding="utf-8")) (self.project_obj.root / "Cargo.lock").write_text("changed after preparation\n", encoding="utf-8") result = packaging.verify_prepared_source( output_source_archive(self.project_obj), manifest, packaging.DEFAULT_LIMITS ) self.assertGreater(result["verified_members"], 0)
class WriteArchiveCoherenceTests(unittest.TestCase): """The archive writer's own verify-while-writing behaviour."""
def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory(prefix="ut-coherence-") self.tmp = Path(self._tmp.name) self.addCleanup(self._tmp.cleanup) self.addCleanup(packaging.set_member_read_hook, None)
def test_matching_expectation_is_written(self) -> None: payload = b"payload\n" source = self.tmp / "a.txt" source.write_bytes(payload) expected = {"m/a.txt": (len(payload), hashlib.sha256(payload).hexdigest())} info = packaging.write_archive( self.tmp / "a.tar.gz", [("m/a.txt", source, 0o644)], packaging.DEFAULT_LIMITS, expected ) self.assertEqual(info["members"], 1) with tarfile.open(self.tmp / "a.tar.gz", "r:gz") as tar: self.assertEqual(tar.extractfile("m/a.txt").read(), payload)
def test_recorded_size_divergence_is_refused(self) -> None: payload = b"payload\n" source = self.tmp / "a.txt" source.write_bytes(payload) expected = {"m/a.txt": (len(payload) + 1, hashlib.sha256(payload).hexdigest())} with self.assertRaises(packaging.CheckFailure) as caught: packaging.write_archive( self.tmp / "a.tar.gz", [("m/a.txt", source, 0o644)], packaging.DEFAULT_LIMITS, expected ) self.assertIn("bytes were recorded when it was hashed", str(caught.exception)) self.assertFalse((self.tmp / "a.tar.gz").exists()) self.assertEqual([p.name for p in self.tmp.iterdir() if p.name.startswith(".pack-")], [])
def test_recorded_digest_divergence_is_refused(self) -> None: payload = b"payload\n" source = self.tmp / "a.txt" source.write_bytes(payload) expected = {"m/a.txt": (len(payload), "0" * 64)} with self.assertRaises(packaging.CheckFailure) as caught: packaging.write_archive( self.tmp / "a.tar.gz", [("m/a.txt", source, 0o644)], packaging.DEFAULT_LIMITS, expected ) self.assertIn("hashes to", str(caught.exception)) self.assertFalse((self.tmp / "a.tar.gz").exists())
def test_in_memory_member_is_verified_too(self) -> None: expected = {"m/generated.json": (3, hashlib.sha256(b"abc").hexdigest())} with self.assertRaises(packaging.CheckFailure): packaging.write_archive( self.tmp / "b.tar.gz", [("m/generated.json", b"abd", 0o644)], packaging.DEFAULT_LIMITS, expected ) self.assertFalse((self.tmp / "b.tar.gz").exists())
def test_without_expectations_the_writer_still_writes(self) -> None: # The binary archive path passes no expectations; it must keep working. source = self.tmp / "a.txt" source.write_bytes(b"payload\n") info = packaging.write_archive( self.tmp / "c.tar.gz", [("m/a.txt", source, 0o644)], packaging.DEFAULT_LIMITS ) self.assertEqual(info["members"], 1) self.assertTrue((self.tmp / "c.tar.gz").exists())
if __name__ == "__main__": unittest.main()