Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
39 kB · 809 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810#!/usr/bin/env python3"""Tests for the extracted-source verification the generated INSTALL documents.
``tools/package_release.py`` used to tell a reader to run``sha256sum --check source-manifest.json`` after extracting the source archive.That could never work: ``source-manifest.json`` is JSON, not a ``sha256sum``listing, and it is not inside the source archive at all — it ships in the releasedirectory and inside the *binary* archive. This module pins the replacement:
- the generated ``INSTALL.md`` documents ``tools/verify_extracted_source.py`` with the archive-derived paths, no longer prints the impossible command, and keeps every step's directory change inside a subshell;- the *documented block sequence itself* is executed end to end against a real source archive, a real binary archive and an executable stub CLI, so cwd leakage and a helper that leaves a ``__pycache__`` behind both fail the test;- the helper verifies a genuinely extracted tree against a genuinely resolved index — first-party files, the vendor tree, the distribution-document alias and the runtime corresponding-source members — and requires the packager's generated members;- it fails closed on a tampered file, a missing file, an added file, an added directory, a symlink, a hard link, a missing generated member, a mismatched provenance record, a wrong manifest kind, a misnamed source root, a duplicated member and a wrong-typed index entry.
The archive and index here are built with the packager's own writer and the samemember set a real ``--prepare-source`` produces, then extracted with ``tar``. Nonetwork, toolchain, container or device is touched, and a green run is evidenceabout the verifier and the instructions, never about a real release."""
from __future__ import annotations
import contextlibimport hashlibimport ioimport jsonimport osimport reimport shutilimport subprocessimport sysimport tarfileimport tempfileimport unittestfrom pathlib import Pathfrom unittest import mock
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"if str(TOOLS) not in sys.path: sys.path.insert(0, str(TOOLS))
import package_release as packaging # noqa: E402import verify_extracted_source as verifier # noqa: E402import verify_release_local as verify # noqa: E402
SOURCE_PREFIX = "ultimate-tool-0.2.0-preview.1-linux-x86_64-gnu-source"BINARY_PREFIX = "ultimate-tool-0.2.0-preview.1-linux-x86_64-gnu"RESERVED_VERSION = "0.2.0-preview.1"
# The helper and the first-party modules it imports from the tree: a real package# ships them, and running the helper out of the extracted tree needs them there.TOOLS_MODULES = ( "package_release.py", "verify_release_local.py", "build_release_source.py", "verify_extracted_source.py",)
# A withheld repository document ships as its reviewed sanitized copy, aliased to# the archive path the original would have occupied.DISTRIBUTION_COPY = "tools/release-docs/RELEASING.md"DISTRIBUTION_ARCHIVE_PATH = "docs/RELEASING.md"DISTRIBUTION_BYTES = b"# Releasing (sanitized copy)\n"VENDOR_CONFIG_BYTES = ( b'[source.crates-io]\nreplace-with = "vendored-sources"\n\n' b'[source.vendored-sources]\ndirectory = "vendor"\n')VENDOR_FILES = { "serde/Cargo.toml": b'[package]\nname = "serde"\n', "serde/src/lib.rs": b"// serde\n",}RUNTIME_MEMBERS = { "runtime-sources/bundle.tar.gz": b"runtime corresponding source\n", "runtime-sources/bundle.json": b'{"record_kind": "runtime_source_bundle"}\n',}SYNTHETIC_FILES = { "Cargo.toml": b'[workspace]\nmembers = ["crates/x"]\n', "Cargo.lock": b"# lock\n", "crates/x/Cargo.toml": b'[package]\nname = "x"\nversion = "0.2.0-preview.1"\n', DISTRIBUTION_COPY: DISTRIBUTION_BYTES,}# An executable stub so the documented ``--version``/``validate``/``doctor``/# ``test`` commands can actually run; it is a real program, not a skipped step.STUB_CLI = b"""#!/bin/shcase "$1" in --version) echo "ultimate-tool (extracted-source test stub)"; exit 0 ;; --help) echo "usage: ultimate-tool <command>"; exit 0 ;;esacecho "stub: $*"exit 0"""
def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest()
def tools_module_bytes() -> dict[str, bytes]: return {f"tools/{name}": (TOOLS / name).read_bytes() for name in TOOLS_MODULES}
def tree_files(root: Path) -> set[str]: return { path.relative_to(root).as_posix() for path in root.rglob("*") if path.is_file() or path.is_symlink() }
class Package: """A prepared-style source archive, its resolved index and the extracted tree."""
def __init__(self, work: Path, *, vendored: bool = True, runtime: bool = True, extract_tree: bool = True) -> None: self.work = work self.vendored = vendored self.runtime = runtime first_party = dict(SYNTHETIC_FILES) first_party.update(tools_module_bytes()) self.files = { "first_party": [ {"path": name, "sha256": sha256(content), "size": len(content)} for name, content in sorted(first_party.items()) ], "vendor": [ {"path": name, "sha256": sha256(content), "size": len(content)} for name, content in sorted(VENDOR_FILES.items()) ] if vendored else [], } aggregate = verify.aggregate_digest(self.files["first_party"]) provenance = { "schema": packaging.SCHEMA, "record_kind": "release_source_provenance", "files": {"count": len(self.files["first_party"]), "aggregate_sha256": aggregate}, } members: list[tuple[str, bytes, int]] = [] for name, content in sorted(first_party.items()): members.append((f"{SOURCE_PREFIX}/{name}", content, packaging.ARCHIVE_MODE_FILE)) if runtime: for name, content in sorted(RUNTIME_MEMBERS.items()): members.append((f"{SOURCE_PREFIX}/{name}", content, packaging.ARCHIVE_MODE_FILE)) if vendored: for name, content in sorted(VENDOR_FILES.items()): members.append((f"{SOURCE_PREFIX}/vendor/{name}", content, packaging.ARCHIVE_MODE_FILE)) members.append((f"{SOURCE_PREFIX}/vendor-sources.toml", VENDOR_CONFIG_BYTES, packaging.ARCHIVE_MODE_FILE)) members.append((f"{SOURCE_PREFIX}/{DISTRIBUTION_ARCHIVE_PATH}", DISTRIBUTION_BYTES, packaging.ARCHIVE_MODE_FILE)) members.append( (f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", json.dumps(provenance, indent=2, sort_keys=True).encode("utf-8") + b"\n", packaging.ARCHIVE_MODE_FILE) ) self.archive = work / f"{SOURCE_PREFIX}.tar.gz" self.archive_info = packaging.write_archive(self.archive, members) self.manifest = { "schema": packaging.SCHEMA, "record_kind": "release_source_manifest_resolved", "archive": {"name": self.archive.name, "prefix": SOURCE_PREFIX, "sha256": self.archive_info["sha256"]}, "files": self.files["first_party"], "aggregate_sha256": aggregate, "vendor": { "files": self.files["vendor"], "aggregate_sha256": verify.aggregate_digest(self.files["vendor"]), "file_count": len(self.files["vendor"]), "status": "generated_cargo_vendor" if vendored else "not_included", }, "distribution_docs": [ {"copy": DISTRIBUTION_COPY, "archive_path": DISTRIBUTION_ARCHIVE_PATH, "sha256": sha256(DISTRIBUTION_BYTES), "size": len(DISTRIBUTION_BYTES)} ], "runtime_bundle": ( {"present": True, "source_members": [ {"path": name, "sha256": sha256(content), "size": len(content)} for name, content in sorted(RUNTIME_MEMBERS.items()) ]} if runtime else {"present": False} ), } self.manifest_path = work / "source-manifest.json" self.manifest_bytes = json.dumps(self.manifest, indent=2, sort_keys=True).encode("utf-8") + b"\n" self.manifest_path.write_bytes(self.manifest_bytes) self.tree = self.extract(parent=work) if extract_tree else work / SOURCE_PREFIX
def expected_tree_files(self) -> set[str]: names = {entry["path"] for entry in self.files["first_party"]} names |= {entry["archive_path"] for entry in self.manifest["distribution_docs"]} names |= {f"vendor/{entry['path']}" for entry in self.files["vendor"]} names |= {entry["path"] for entry in self.manifest["runtime_bundle"].get("source_members", [])} names.add("SOURCE-PROVENANCE.json") if self.files["vendor"]: names.add("vendor-sources.toml") return names
def extract(self, parent: Path | None = None, name: str = SOURCE_PREFIX) -> Path: # Members already carry the archive prefix, so a real `tar -xzf` lays the # tree down under <cwd>/<prefix>; extraction goes into the parent here too. parent = parent or self.work with tarfile.open(self.archive, "r:gz") as tar: for member in tar.getmembers(): target = parent / member.name target.parent.mkdir(parents=True, exist_ok=True) source = tar.extractfile(member) assert source is not None with open(target, "wb") as handle: shutil.copyfileobj(source, handle) os.chmod(target, member.mode & 0o777) return parent / name
def write_binary_archive(path: Path, index_bytes: bytes) -> dict: members = [ (f"{BINARY_PREFIX}/bin/ultimate-tool", STUB_CLI, packaging.ARCHIVE_MODE_EXECUTABLE), (f"{BINARY_PREFIX}/bin/ultimate-tool-mcp", STUB_CLI, packaging.ARCHIVE_MODE_EXECUTABLE), (f"{BINARY_PREFIX}/examples/first-run/c64dev.toml", b"[project]\nname = \"first-run\"\n", packaging.ARCHIVE_MODE_FILE), (f"{BINARY_PREFIX}/source-manifest.json", index_bytes, packaging.ARCHIVE_MODE_FILE), ] return packaging.write_archive(path, members)
def write_sha256sums(directory: Path, names: list[str]) -> None: lines = [f"{sha256((directory / name).read_bytes())} {name}" for name in names] (directory / "SHA256SUMS").write_text("\n".join(lines) + "\n", encoding="utf-8")
def sh_blocks(document: str) -> list[str]: return re.findall(r"```sh\n(.*?)```", document, flags=re.DOTALL)
INSTALL_CONTEXT = { "release": { "product_title": "Ultimate Tool", "license": "GPL-3.0-or-later", "runtime_requirements": [], "support_boundary": {"tested_platform": "Ubuntu 24.04"}, }, "version": RESERVED_VERSION, "archive": {"name": f"{BINARY_PREFIX}.tar.gz"}, "source_archive": {"name": f"{SOURCE_PREFIX}.tar.gz"},}
class ExtractedSourceTests(unittest.TestCase): def setUp(self) -> None: self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-source-")) self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
def package(self, **kwargs) -> Package: return Package(self.root, **kwargs)
def assert_refused(self, tree: Path, manifest: Path, expected: str) -> None: with self.assertRaises(verify.HarnessError) as caught: verifier.verify_extracted_tree(tree, manifest) self.assertIn(expected, str(caught.exception))
def test_a_faithful_extraction_verifies(self) -> None: package = self.package() result = verifier.verify_extracted_tree(package.tree, package.manifest_path) self.assertEqual(result["result"], "verified") self.assertEqual(result["prefix"], SOURCE_PREFIX) self.assertEqual(result["generated_members"], ["SOURCE-PROVENANCE.json", "vendor-sources.toml"]) self.assertEqual(result["aggregate_sha256"], package.manifest["aggregate_sha256"]) self.assertEqual( result["members"], len(package.files["first_party"]) + len(package.files["vendor"]) + 1 + len(RUNTIME_MEMBERS), ) self.assertGreater(result["hashed_bytes"], 0) self.assertEqual(tree_files(package.tree), package.expected_tree_files())
def test_absent_vendor_tree_does_not_require_a_vendor_config(self) -> None: package = self.package(vendored=False) result = verifier.verify_extracted_tree(package.tree, package.manifest_path) self.assertEqual(result["generated_members"], ["SOURCE-PROVENANCE.json"]) self.assertFalse((package.tree / "vendor-sources.toml").exists())
def test_a_tampered_member_is_refused(self) -> None: package = self.package() # Same length, different bytes: the size check cannot catch this, only the hash. target = package.tree / "crates/x/Cargo.toml" target.write_bytes(b"X" * len(target.read_bytes())) self.assert_refused(package.tree, package.manifest_path, "does not match the SHA-256")
def test_a_missing_member_is_refused(self) -> None: package = self.package() (package.tree / "Cargo.lock").unlink() self.assert_refused(package.tree, package.manifest_path, "missing from the extracted tree")
def test_a_size_change_is_refused(self) -> None: package = self.package() target = package.tree / "Cargo.toml" target.write_bytes(target.read_bytes() + b"# extra\n") self.assert_refused(package.tree, package.manifest_path, "the source index records")
def test_an_added_file_is_refused(self) -> None: package = self.package() (package.tree / "crates/x/build.rs").write_bytes(b"fn main() {}\n") self.assert_refused(package.tree, package.manifest_path, "does not list")
def test_an_added_directory_is_refused(self) -> None: package = self.package() (package.tree / "unlisted").mkdir() self.assert_refused(package.tree, package.manifest_path, "does not list")
def test_a_symlink_is_refused(self) -> None: package = self.package() os.symlink("Cargo.toml", package.tree / "linked.toml") self.assert_refused(package.tree, package.manifest_path, "symbolic link")
def test_a_hard_link_is_refused(self) -> None: package = self.package() source = self.root / "outside-inode" shutil.copyfile(package.tree / "Cargo.lock", source) (package.tree / "Cargo.lock").unlink() os.link(source, package.tree / "Cargo.lock") self.assert_refused(package.tree, package.manifest_path, "hard links")
def test_a_missing_generated_member_is_refused(self) -> None: package = self.package() (package.tree / "SOURCE-PROVENANCE.json").unlink() self.assert_refused(package.tree, package.manifest_path, "SOURCE-PROVENANCE.json")
def test_a_mismatched_provenance_aggregate_is_refused(self) -> None: package = self.package() path = package.tree / "SOURCE-PROVENANCE.json" document = json.loads(path.read_text(encoding="utf-8")) document["files"]["aggregate_sha256"] = "0" * 64 path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8") self.assert_refused(package.tree, package.manifest_path, "different first-party aggregate")
def test_a_wrong_manifest_kind_is_refused(self) -> None: package = self.package() index = json.loads(package.manifest_path.read_text(encoding="utf-8")) index["record_kind"] = "release_source_manifest" package.manifest_path.write_text(json.dumps(index, indent=2, sort_keys=True) + "\n", encoding="utf-8") self.assert_refused(package.tree, package.manifest_path, "point --manifest at the resolved")
def test_a_misnamed_source_root_is_refused(self) -> None: package = self.package(extract_tree=False) alt = self.root / "alt" package.extract(alt) renamed = alt / "renamed-tree" os.rename(alt / SOURCE_PREFIX, renamed) self.assert_refused(renamed, package.manifest_path, "records the archive prefix")
def test_a_symlinked_source_root_is_refused(self) -> None: package = self.package() link = self.root / (SOURCE_PREFIX + "-link") os.symlink(package.tree, link) with self.assertRaises(verify.HarnessError): verifier.verify_extracted_tree(link, package.manifest_path)
class MalformedIndexTests(unittest.TestCase): """A wrong-typed or contradictory index is refused, never half-used."""
def setUp(self) -> None: self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-index-")) self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
def mutate(self, mutate) -> tuple[Package, Path]: package = Package(self.root) index = json.loads(package.manifest_path.read_text(encoding="utf-8")) mutate(index) path = self.root / "mutated.json" path.write_text(json.dumps(index, indent=2, sort_keys=True) + "\n", encoding="utf-8") return package, path
def refused(self, mutate, expected: str) -> None: package, path = self.mutate(mutate) with self.assertRaises(verify.HarnessError) as caught: verifier.verify_extracted_tree(package.tree, path) self.assertIn(expected, str(caught.exception))
def refused_provenance(self, mutate, expected: str) -> None: package = Package(self.root) path = package.tree / "SOURCE-PROVENANCE.json" document = json.loads(path.read_text(encoding="utf-8")) mutate(document) path.write_text(json.dumps(document), encoding="utf-8") with self.assertRaises(verify.HarnessError) as caught: verifier.verify_extracted_tree(package.tree, package.manifest_path) self.assertIn(expected, str(caught.exception))
# -- first-party files ------------------------------------------------- def test_a_non_list_files_record_is_refused(self) -> None: self.refused(lambda index: index.update({"files": "Cargo.toml"}), "is not a list")
def test_a_non_object_entry_is_refused(self) -> None: self.refused(lambda index: index["files"].append("Cargo.toml"), "is not an object")
def test_a_non_string_path_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"path": 7}), "not a non-empty string")
def test_a_parent_segment_path_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"path": "../escape"}), "parent-directory segment")
def test_a_backslash_path_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"path": "crates\\x"}), "backslash")
def test_a_non_hex_digest_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"sha256": "zz"}), "lowercase SHA-256")
def test_an_uppercase_digest_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"sha256": "A" * 64}), "lowercase SHA-256")
def test_a_boolean_size_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"size": True}), "non-negative integer size")
def test_a_string_size_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"size": "12"}), "non-negative integer size")
def test_a_float_size_is_refused(self) -> None: self.refused(lambda index: index["files"][0].update({"size": 1.5}), "non-negative integer size")
def test_a_duplicate_inside_one_category_is_refused(self) -> None: self.refused(lambda index: index["files"].append(dict(index["files"][0])), "more than once")
def test_a_duplicate_member_across_categories_is_refused(self) -> None: self.refused( lambda index: index["distribution_docs"][0].update({"archive_path": "Cargo.toml"}), "more than once", )
# -- distribution documents ------------------------------------------- def test_a_non_list_distribution_docs_record_is_refused(self) -> None: self.refused(lambda index: index.update({"distribution_docs": "x"}), "is not a list")
def test_a_non_object_distribution_doc_is_refused(self) -> None: self.refused(lambda index: index["distribution_docs"].append("x"), "is not an object")
def test_a_distribution_doc_absolute_path_is_refused(self) -> None: self.refused( lambda index: index["distribution_docs"][0].update({"archive_path": "/etc/passwd"}), "absolute member name", )
def test_a_distribution_doc_bad_digest_is_refused(self) -> None: self.refused(lambda index: index["distribution_docs"][0].update({"sha256": "nope"}), "lowercase SHA-256")
def test_a_distribution_doc_negative_size_is_refused(self) -> None: self.refused(lambda index: index["distribution_docs"][0].update({"size": -1}), "non-negative integer size")
# -- vendored crate files --------------------------------------------- def test_a_non_list_vendor_files_record_is_refused(self) -> None: self.refused(lambda index: index["vendor"].update({"files": {"a": 1}}), "is not a list")
def test_a_non_object_vendor_entry_is_refused(self) -> None: self.refused(lambda index: index["vendor"]["files"].append(3), "is not an object")
def test_a_vendor_empty_segment_path_is_refused(self) -> None: self.refused(lambda index: index["vendor"]["files"][0].update({"path": "a//b"}), "empty path segment")
def test_a_vendor_bad_digest_is_refused(self) -> None: self.refused(lambda index: index["vendor"]["files"][0].update({"sha256": "xyz"}), "lowercase SHA-256")
def test_a_vendor_bad_size_is_refused(self) -> None: self.refused(lambda index: index["vendor"]["files"][0].update({"size": True}), "non-negative integer size")
def test_a_duplicate_vendor_member_is_refused(self) -> None: self.refused( lambda index: index["vendor"]["files"].append(dict(index["vendor"]["files"][0])), "more than once", )
# -- runtime corresponding-source members ------------------------------ def test_a_non_list_runtime_members_record_is_refused(self) -> None: self.refused(lambda index: index["runtime_bundle"].update({"source_members": 5}), "is not a list")
def test_a_non_object_runtime_entry_is_refused(self) -> None: self.refused(lambda index: index["runtime_bundle"]["source_members"].append([]), "is not an object")
def test_a_runtime_parent_segment_path_is_refused(self) -> None: self.refused( lambda index: index["runtime_bundle"]["source_members"][0].update({"path": "../x"}), "parent-directory segment", )
def test_a_runtime_bad_digest_is_refused(self) -> None: self.refused( lambda index: index["runtime_bundle"]["source_members"][0].update({"sha256": "q"}), "lowercase SHA-256", )
def test_a_runtime_float_size_is_refused(self) -> None: self.refused( lambda index: index["runtime_bundle"]["source_members"][0].update({"size": 1.5}), "non-negative integer size", )
def test_a_duplicate_runtime_member_is_refused(self) -> None: self.refused( lambda index: index["runtime_bundle"]["source_members"].append( dict(index["runtime_bundle"]["source_members"][0]) ), "more than once", )
# -- strict scalar types ---------------------------------------------- def test_a_boolean_schema_is_refused(self) -> None: self.refused(lambda index: index.update({"schema": True}), "expected the integer")
def test_a_float_schema_is_refused(self) -> None: self.refused(lambda index: index.update({"schema": 1.0}), "expected the integer")
def test_a_string_runtime_present_is_refused(self) -> None: self.refused(lambda index: index["runtime_bundle"].update({"present": "true"}), "not a boolean")
def test_a_numeric_runtime_present_is_refused(self) -> None: self.refused(lambda index: index["runtime_bundle"].update({"present": 1}), "not a boolean")
def test_a_contradictory_aggregate_is_refused(self) -> None: self.refused(lambda index: index.update({"aggregate_sha256": "0" * 64}), "does not match the aggregate it records")
# -- the archive's own bound record ------------------------------------ def test_a_boolean_provenance_schema_is_refused(self) -> None: self.refused_provenance(lambda document: document.update({"schema": True}), "expected the integer")
def test_a_float_provenance_schema_is_refused(self) -> None: self.refused_provenance(lambda document: document.update({"schema": 1.0}), "expected the integer")
def test_a_boolean_provenance_count_is_refused(self) -> None: self.refused_provenance(lambda document: document["files"].update({"count": True}), "non-integer file count")
def test_a_float_provenance_count_is_refused(self) -> None: self.refused_provenance(lambda document: document["files"].update({"count": 1.0}), "non-integer file count")
class BoundedReadTests(unittest.TestCase): """The per-member read cannot run past the size the index recorded."""
def test_a_member_that_grows_after_open_is_refused(self) -> None: root = Path(tempfile.mkdtemp(prefix="ut-test-install-growth-")) self.addCleanup(shutil.rmtree, root, ignore_errors=True) package = Package(root) expected = next(entry for entry in package.files["first_party"] if entry["path"] == "Cargo.toml")
@contextlib.contextmanager def grown(path, *, max_bytes, description): # A stream that keeps yielding past the declared size, standing in for a # file another process appended to between the size check and the read. yield io.BytesIO(b"\0" * (expected["size"] + 64))
with mock.patch.object(verifier.packaging, "open_regular_readonly", grown): with self.assertRaises(verify.HarnessError) as caught: verifier.verify_extracted_tree(package.tree, package.manifest_path) self.assertIn("larger than", str(caught.exception))
def test_a_member_of_exactly_the_recorded_size_is_accepted(self) -> None: root = Path(tempfile.mkdtemp(prefix="ut-test-install-exact-")) self.addCleanup(shutil.rmtree, root, ignore_errors=True) package = Package(root) expected = next(entry for entry in package.files["first_party"] if entry["path"] == "Cargo.toml")
@contextlib.contextmanager def exact(path, *, max_bytes, description): yield io.BytesIO(b"\0" * expected["size"])
with mock.patch.object(verifier.packaging, "open_regular_readonly", exact): # Same length, different bytes: the read is bounded, so Cargo.toml itself # is a hash mismatch, not an overlong read. with self.assertRaises(verify.HarnessError) as caught: verifier.verify_extracted_tree(package.tree, package.manifest_path) self.assertIn(" Cargo.toml does not match the SHA-256", str(caught.exception))
class ExtractedHelperCliTests(unittest.TestCase): """The helper run *out of the extracted tree*, as the instructions say."""
def setUp(self) -> None: self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-cli-")) self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
def run_extracted(self, package: Package, *args: str, cwd: Path | None = None) -> subprocess.CompletedProcess: helper = package.tree / "tools" / "verify_extracted_source.py" return subprocess.run( [sys.executable, str(helper), *args], capture_output=True, text=True, cwd=str(cwd or self.root), timeout=180, )
def test_the_extracted_helper_verifies_and_exits_zero(self) -> None: package = Package(self.root) result = self.run_extracted(package, "--source-root", str(package.tree), "--manifest", str(package.manifest_path), "--json") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout)["result"], "verified")
def test_running_the_extracted_helper_leaves_no_bytecode_in_the_tree(self) -> None: package = Package(self.root) before = tree_files(package.tree) result = self.run_extracted(package, "--source-root", str(package.tree), "--manifest", str(package.manifest_path)) self.assertEqual(result.returncode, 0, result.stderr) after = tree_files(package.tree) self.assertEqual(after, before) self.assertFalse([name for name in after if "__pycache__" in name])
def test_the_default_source_root_is_the_current_directory(self) -> None: package = Package(self.root) result = self.run_extracted(package, "--manifest", str(package.manifest_path), cwd=package.tree) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("verified", result.stdout)
def test_a_tampered_tree_exits_nonzero(self) -> None: package = Package(self.root) target = package.tree / "Cargo.toml" target.write_bytes(b"X" * len(target.read_bytes())) result = self.run_extracted(package, "--source-root", str(package.tree), "--manifest", str(package.manifest_path)) self.assertEqual(result.returncode, 1) self.assertIn("does not match the SHA-256", result.stderr)
def test_a_wrong_typed_index_exits_nonzero_without_a_traceback(self) -> None: package = Package(self.root) index = json.loads(package.manifest_path.read_text(encoding="utf-8")) index["files"][0]["size"] = "12" bad = self.root / "bad.json" bad.write_text(json.dumps(index), encoding="utf-8") result = self.run_extracted(package, "--source-root", str(package.tree), "--manifest", str(bad)) self.assertEqual(result.returncode, 1) self.assertNotIn("Traceback", result.stderr) self.assertIn("non-negative integer size", result.stderr)
def test_a_wrong_typed_runtime_present_exits_nonzero_without_a_traceback(self) -> None: package = Package(self.root) index = json.loads(package.manifest_path.read_text(encoding="utf-8")) index["runtime_bundle"]["present"] = "yes" bad = self.root / "bad-runtime.json" bad.write_text(json.dumps(index), encoding="utf-8") result = self.run_extracted(package, "--source-root", str(package.tree), "--manifest", str(bad)) self.assertEqual(result.returncode, 1) self.assertNotIn("Traceback", result.stderr) self.assertIn("not a boolean", result.stderr)
def test_a_missing_manifest_is_a_usage_error(self) -> None: package = Package(self.root) helper = package.tree / "tools" / "verify_extracted_source.py" result = subprocess.run([sys.executable, str(helper), "--source-root", str(package.tree)], capture_output=True, text=True, cwd=str(self.root), timeout=60) self.assertEqual(result.returncode, 2)
def snapshot(root: Path) -> dict[str, tuple[str, str]]: """Every path under ``root``: its kind and, for a file, its content hash.
Directories are recorded too, so a verification run that creates or removes even an empty directory is visible. ``__pycache__`` would show up here. """ recorded: dict[str, tuple[str, str]] = {} for current, dirnames, filenames in os.walk(root): for name in dirnames: path = Path(current) / name relative = path.relative_to(root).as_posix() recorded[relative] = ("link", os.readlink(path)) if path.is_symlink() else ("dir", "") for name in filenames: path = Path(current) / name relative = path.relative_to(root).as_posix() if path.is_symlink(): recorded[relative] = ("link", os.readlink(path)) else: recorded[relative] = ("file", hashlib.sha256(path.read_bytes()).hexdigest()) return recorded
def verification_command(block: str) -> str: """The verification invocation from a documented block, without its extraction.
The block extracts the source archive and then runs the helper. Re-running only the helper against an already-extracted tree is what proves the check is read-only; re-extracting would overwrite whatever was tampered with first. """ return "\n".join(line for line in block.splitlines() if not line.strip().startswith("tar -xzf"))
class DocumentedSequenceTests(unittest.TestCase): """Execute the generated instructions exactly, in one shell session."""
def setUp(self) -> None: self.root = Path(tempfile.mkdtemp(prefix="ut-test-install-seq-")) self.addCleanup(shutil.rmtree, self.root, ignore_errors=True) self.shim = self.root / "shim" self.shim.mkdir() (self.shim / "python3").symlink_to(sys.executable)
def environment(self) -> dict: return {**os.environ, "PATH": f"{self.shim}{os.pathsep}{os.environ['PATH']}"}
def prepare(self) -> tuple[Package, list[str]]: package = Package(self.root, extract_tree=False) binary_archive = self.root / f"{BINARY_PREFIX}.tar.gz" write_binary_archive(binary_archive, package.manifest_bytes) write_sha256sums(self.root, [binary_archive.name, package.archive.name]) document = packaging.render_install(INSTALL_CONTEXT).decode("utf-8") relevant = [ block for block in sh_blocks(document) if "sha256sum -c SHA256SUMS" in block or "bin/ultimate-tool --version" in block or "examples/first-run" in block or "verify_extracted_source.py" in block ] self.assertEqual(len(relevant), 4, relevant) return package, relevant
def run_blocks(self, blocks: list[str], *, extra: str = "") -> subprocess.CompletedProcess: script = "\n".join(part.rstrip() for part in blocks) + "\n" + extra return subprocess.run( ["bash", "-eu", "-c", script], cwd=self.root, capture_output=True, text=True, timeout=300, env=self.environment(), )
def assert_tree_matches_index(self, package: Package) -> Path: tree = self.root / SOURCE_PREFIX self.assertEqual(tree_files(tree), package.expected_tree_files()) self.assertFalse([name for name in tree_files(tree) if "__pycache__" in name]) for entry in package.files["first_party"]: self.assertEqual(sha256((tree / entry["path"]).read_bytes()), entry["sha256"]) for entry in package.files["vendor"]: self.assertEqual(sha256((tree / "vendor" / entry["path"]).read_bytes()), entry["sha256"]) for entry in package.manifest["distribution_docs"]: self.assertEqual(sha256((tree / entry["archive_path"]).read_bytes()), entry["sha256"]) for entry in package.manifest["runtime_bundle"]["source_members"]: self.assertEqual(sha256((tree / entry["path"]).read_bytes()), entry["sha256"]) return tree
def test_the_documented_block_sequence_runs_from_one_directory(self) -> None: package, relevant = self.prepare() verify_block = next(block for block in relevant if "verify_extracted_source.py" in block) self.assertIn("-B ", verify_block) result = self.run_blocks(relevant, extra='printf "PWD=%s\\n" "$PWD"\n') self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertIn("verified", result.stdout) self.assertIn(f"PWD={self.root}", result.stdout) self.assert_tree_matches_index(package) self.assertTrue((self.root / BINARY_PREFIX / "examples" / "first-run").is_dir())
def test_the_verification_command_is_read_only_and_rerunnable(self) -> None: package, relevant = self.prepare() verify_block = next(block for block in relevant if "verify_extracted_source.py" in block) first = self.run_blocks(relevant) self.assertEqual(first.returncode, 0, first.stdout + first.stderr) tree = self.root / SOURCE_PREFIX before = snapshot(tree) self.assertIn(("dir", ""), before.values()) self.assertEqual(snapshot(tree), before) second = self.run_blocks([verification_command(verify_block)]) self.assertEqual(second.returncode, 0, second.stdout + second.stderr) self.assertIn("verified", second.stdout) self.assertEqual(snapshot(tree), before)
def test_tampering_an_indexed_file_fails_the_same_verification_command(self) -> None: package, relevant = self.prepare() verify_block = next(block for block in relevant if "verify_extracted_source.py" in block) first = self.run_blocks(relevant) self.assertEqual(first.returncode, 0, first.stdout + first.stderr) tree = self.root / SOURCE_PREFIX target = tree / "Cargo.toml" target.write_bytes(b"X" * len(target.read_bytes())) tampered = snapshot(tree) result = self.run_blocks([verification_command(verify_block)]) self.assertNotEqual(result.returncode, 0) self.assertIn("does not match the SHA-256", result.stderr) # The failed check wrote nothing: the tampered tree is byte for byte as it was. self.assertEqual(snapshot(tree), tampered)
class GeneratedInstallTests(unittest.TestCase): """The documented procedure must name the shipped helper and real inputs."""
def install(self) -> str: return packaging.render_install(INSTALL_CONTEXT).decode("utf-8")
def test_the_impossible_command_is_gone(self) -> None: document = self.install() self.assertNotIn("sha256sum --check source-manifest.json", document)
def test_the_procedure_names_the_shipped_helper_and_real_inputs(self) -> None: document = self.install() self.assertIn(f"python3 -B {SOURCE_PREFIX}/tools/verify_extracted_source.py", document) self.assertIn(f"--source-root {SOURCE_PREFIX}", document) self.assertIn(f"--manifest {BINARY_PREFIX}/source-manifest.json", document) self.assertIn(f"tar -xzf {SOURCE_PREFIX}.tar.gz", document)
def test_no_documented_step_leaves_the_download_directory(self) -> None: document = self.install() for block in sh_blocks(document): for line in block.splitlines(): stripped = line.strip() self.assertFalse( stripped.startswith("cd "), f"a documented step changes directory outside a subshell: {stripped!r}", )
def test_the_documented_helper_is_the_one_in_the_tree(self) -> None: self.assertTrue((ROOT / "tools" / "verify_extracted_source.py").is_file()) self.assertIn("tools/verify_extracted_source.py", self.install())
def test_the_source_section_still_points_at_the_checksum_chain(self) -> None: document = self.install() self.assertIn("sha256sum -c SHA256SUMS", document) self.assertIn("SOURCE-AND-LICENSES.md", document)
if __name__ == "__main__": unittest.main()