From 8937c6894e6220cf4531844c3c00f7469808055d Mon Sep 17 00:00:00 2001 From: loonghao Date: Fri, 9 Oct 2026 12:44:56 +0800 Subject: [PATCH] feat: verify immutable Python intake release assets --- tests/test_release_readback.py | 217 +++++++++++++++++++++++++++++++++ verify_release_readback.py | 193 +++++++++++++++++++++++++++++ 2 files changed, 410 insertions(+) create mode 100644 tests/test_release_readback.py create mode 100644 verify_release_readback.py diff --git a/tests/test_release_readback.py b/tests/test_release_readback.py new file mode 100644 index 0000000..baaa69f --- /dev/null +++ b/tests/test_release_readback.py @@ -0,0 +1,217 @@ +"""Offline contracts for the exact Python intake release publication boundary.""" + +from __future__ import annotations + +import importlib.util +import io +import json +import shutil +import tarfile +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import verify_release_readback as readback + + +# Reuse the existing provenance fixture, including its 22 pinned source receipts. +SPEC = importlib.util.spec_from_file_location( + "_intake_readback_fixtures", Path(__file__).with_name("test_intake_contract.py") +) +fixtures = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(fixtures) + + +def write_companion(directory: Path, name: str) -> None: + sha256 = readback.intake.digest(directory / name) + (directory / f"{name}.sha256").write_bytes(f"{sha256} {name}\n".encode("ascii")) + + +class ReleaseReadbackTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + base = Path(self.temporary.name) + self.root = base / "source" + self.root.mkdir() + _, raw = fixtures.fixture(self.root) + self.expected = base / "expected" + self.expected.mkdir() + self.downloaded = base / "downloaded" + for path in raw.iterdir(): + shutil.copyfile(path, self.expected / path.name) + shutil.copyfile(self.root / readback.INTAKE_MANIFEST, self.expected / readback.INTAKE_MANIFEST) + manifest = readback.intake.read_json(self.root / readback.INTAKE_MANIFEST) + (self.expected / readback.RELEASE_MANIFEST).write_text( + json.dumps({"schema_version": 1, "assets": manifest["assets"]}), encoding="utf-8" + ) + self.metadata_files = sorted( + {path.relative_to(self.root).as_posix() for path in (self.root / "metadata").rglob("*") if path.is_file()} + | readback.intake.METADATA_ROOT_FILES + ) + readback.intake.write_metadata_archive( + self.root, self.metadata_files, self.expected / readback.METADATA_ARCHIVE, + ) + for path in list(self.expected.iterdir()): + if not path.name.endswith(".sha256"): + write_companion(self.expected, path.name) + shutil.copytree(self.expected, self.downloaded) + + def verify(self): + # The shared schema's real recipe validation is exercised by repository CI. + with mock.patch.object(readback.intake, "shared_validator", return_value=lambda recipe: None): + return readback.verify_release_readback( + self.root, self.root, self.expected, self.downloaded, + ) + + def replace_json(self, name: str, value) -> None: + for directory in (self.expected, self.downloaded): + (directory / name).write_text(json.dumps(value), encoding="utf-8") + write_companion(directory, name) + + def test_exact_60_assets_and_original_metadata_verify_without_claiming_native_acceptance(self): + report = self.verify() + self.assertEqual(report["release_asset_count"], 60) + self.assertEqual(report["raw_archive_assets"], 27) + self.assertEqual(report["metadata_files"], 10) + self.assertEqual(report["native_runtime_acceptance"], "not-run") + self.assertEqual(report["github_immutable_release_state"], "not-checked-by-local-file-verifier") + self.assertEqual(set(report["ancillary_pins"]), readback.ANCILLARY_FILES) + + def test_missing_or_unexpected_download_is_rejected(self): + path = self.downloaded / "archive-0.tar.gz" + content = path.read_bytes() + path.unlink() + with self.assertRaisesRegex(readback.ReadbackError, "missing or unexpected"): + self.verify() + path.write_bytes(content) + (self.downloaded / "extra.txt").write_bytes(b"unexpected") + with self.assertRaisesRegex(readback.ReadbackError, "missing or unexpected"): + self.verify() + + def test_nonflat_download_is_rejected(self): + (self.downloaded / "nested").mkdir() + with self.assertRaisesRegex(readback.intake.IntakeError, "regular file"): + self.verify() + + def test_raw_corruption_cannot_be_hidden_by_a_changed_downloaded_checksum(self): + name = "archive-0.tar.gz" + original = (self.downloaded / name).read_bytes() + (self.downloaded / name).write_bytes(b"!" + original[1:]) + write_companion(self.downloaded, name) + with self.assertRaisesRegex(readback.intake.IntakeError, "pinned SHA256"): + self.verify() + + def test_checksum_companion_requires_exact_digest_filename_and_lf(self): + name = "archive-0.tar.gz" + path = self.downloaded / f"{name}.sha256" + original = path.read_bytes() + for content in [original.replace(b"archive-0", b"archive-1"), original.upper(), original.replace(b"\n", b"\r\n")]: + path.write_bytes(content) + with self.subTest(content=content), self.assertRaisesRegex(readback.ReadbackError, "checksum companion"): + self.verify() + + def test_release_manifest_rejects_duplicate_names_and_changed_historical_pins(self): + original = readback.intake.read_json(self.expected / readback.RELEASE_MANIFEST) + value = json.loads(json.dumps(original)) + value["assets"][1] = value["assets"][0] + self.replace_json(readback.RELEASE_MANIFEST, value) + with self.assertRaisesRegex(readback.ReadbackError, "duplicate asset names"): + self.verify() + value = json.loads(json.dumps(original)) + value["assets"][0]["sha256"] = "0" * 64 + self.replace_json(readback.RELEASE_MANIFEST, value) + with self.assertRaisesRegex(readback.ReadbackError, "historical identity"): + self.verify() + + def test_release_manifest_accepts_only_original_or_exact_organization_urls(self): + value = readback.intake.read_json(self.expected / readback.RELEASE_MANIFEST) + for asset in value["assets"]: + asset["url"] = f"{readback.ORGANIZATION_BASE}/{asset['name']}" + self.replace_json(readback.RELEASE_MANIFEST, value) + self.verify() + value["assets"][0]["url"] += "?sig=DO_NOT_PRINT" + self.replace_json(readback.RELEASE_MANIFEST, value) + with self.assertRaises(readback.ReadbackError) as caught: + self.verify() + self.assertNotIn("DO_NOT_PRINT", str(caught.exception)) + + def test_changed_intake_manifest_cannot_replace_the_local_public_trust_anchor(self): + value = readback.intake.read_json(self.expected / readback.INTAKE_MANIFEST) + value["publication_status"] = "different bytes" + self.replace_json(readback.INTAKE_MANIFEST, value) + with self.assertRaisesRegex(readback.intake.IntakeError, "pinned"): + self.verify() + + def test_duplicate_json_keys_in_release_manifest_are_rejected(self): + for directory in (self.expected, self.downloaded): + path = directory / readback.RELEASE_MANIFEST + path.write_text('{"schema_version":1,"schema_version":1,"assets":[]}', encoding="utf-8") + write_companion(directory, path.name) + with self.assertRaisesRegex(readback.intake.IntakeError, "duplicate object key"): + self.verify() + + def rewrite_metadata(self, transform) -> None: + path = self.expected / readback.METADATA_ARCHIVE + with tarfile.open(path, "w:gz", format=tarfile.PAX_FORMAT) as archive: + for name in self.metadata_files: + content = (self.root / name).read_bytes() + member = tarfile.TarInfo(name) + member.size = len(content) + member.mode = 0o644 + member.mtime = member.uid = member.gid = 0 + member.uname = member.gname = "" + content = transform(member, content) + archive.addfile(member, io.BytesIO(content)) + write_companion(self.expected, path.name) + for name in (path.name, f"{path.name}.sha256"): + shutil.copyfile(self.expected / name, self.downloaded / name) + + def test_self_consistent_metadata_archive_still_requires_original_public_bytes(self): + def corrupt(member, content): + return b"!" + content[1:] if member.name == "LICENSE" else content + self.rewrite_metadata(corrupt) + with self.assertRaisesRegex(readback.ReadbackError, "public pinned bytes"): + self.verify() + + def test_metadata_archive_cannot_smuggle_links_or_traversal_members(self): + def link(member, content): + if member.name == "LICENSE": + member.type = tarfile.SYMTYPE + member.linkname = "../../outside" + member.size = 0 + return b"" + return content + self.rewrite_metadata(link) + with self.assertRaisesRegex(readback.ReadbackError, "ordinary regular"): + self.verify() + def traversal(member, content): + if member.name == "LICENSE": + member.name = "../outside" + return content + self.rewrite_metadata(traversal) + with self.assertRaisesRegex(readback.intake.IntakeError, "stay inside"): + self.verify() + + def test_ancillary_download_corruption_is_rejected_even_with_matching_companion(self): + name = readback.METADATA_ARCHIVE + content = (self.downloaded / name).read_bytes() + (self.downloaded / name).write_bytes(content[:-1] + bytes([content[-1] ^ 1])) + write_companion(self.downloaded, name) + with self.assertRaisesRegex(readback.intake.IntakeError, "pinned SHA256"): + self.verify() + + def test_symlink_download_is_rejected(self): + path = self.downloaded / "archive-0.tar.gz" + path.unlink() + try: + path.symlink_to(self.expected / path.name) + except OSError: + self.skipTest("host does not permit unprivileged symlinks") + with self.assertRaisesRegex(readback.intake.IntakeError, "filesystem links"): + self.verify() + + +if __name__ == "__main__": + unittest.main() diff --git a/verify_release_readback.py b/verify_release_readback.py new file mode 100644 index 0000000..eecd9ee --- /dev/null +++ b/verify_release_readback.py @@ -0,0 +1,193 @@ +"""Verify the Python intake release's downloaded files against local public pins.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import stat +import tarfile +from pathlib import Path + +import verify_intake as intake + + +RELEASE_MANIFEST = "release-assets.json" +INTAKE_MANIFEST = "intake-assets.json" +METADATA_ARCHIVE = "intake-metadata.tar.gz" +ANCILLARY_FILES = {RELEASE_MANIFEST, INTAKE_MANIFEST, METADATA_ARCHIVE} +ORGANIZATION_BASE = ( + "https://github.com/vx-org/python/releases/download/python-3.7.9-pbs-20200823" +) + + +class ReadbackError(ValueError): + """Downloaded assets do not satisfy the pinned intake release contract.""" + + +def inventory(directory: Path, expected: set[str]) -> dict[str, Path]: + """Require a flat directory of exactly the expected, portable regular files.""" + attributes = directory.lstat() + if ( + not stat.S_ISDIR(attributes.st_mode) + or stat.S_ISLNK(attributes.st_mode) + or getattr(attributes, "st_file_attributes", 0) + & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400) + ): + raise ReadbackError("release asset directory must be an ordinary directory") + files = {} + folded = set() + for path in directory.iterdir(): + name = intake.safe_relative(path.name, flat=True) + if name.casefold() in folded: + raise ReadbackError("release asset names collide on a portable filesystem") + folded.add(name.casefold()) + files[name] = intake.contained_regular(directory, name) + if set(files) != expected: + raise ReadbackError("release asset directory has missing or unexpected files") + return files + + +def verify_checksum_companion(files: dict[str, Path], name: str, sha256: str) -> None: + canonical = f"{sha256} {name}\n".encode("ascii") + if files[f"{name}.sha256"].read_bytes() != canonical: + raise ReadbackError(f"release checksum companion differs from its pinned asset: {name}") + + +def validate_release_manifest(value: dict, manifest: dict) -> None: + if not isinstance(value, dict) or set(value) != {"schema_version", "assets"}: + raise ReadbackError("retained release manifest has unexpected fields") + assets = value["assets"] + if type(value["schema_version"]) is not int or value["schema_version"] != 1 or not isinstance(assets, list) or len(assets) != intake.ARCHIVE_COUNT: + raise ReadbackError("retained release manifest requires 27 schema-v1 assets") + expected = {asset["name"]: asset for asset in manifest["assets"]} + observed = set() + folded = set() + for asset in assets: + if not isinstance(asset, dict) or set(asset) != {"name", "url", "sha256", "size"}: + raise ReadbackError("retained release asset fields differ from the intake contract") + name = intake.safe_relative(asset["name"], flat=True) + if name.casefold() in folded: + raise ReadbackError("retained release manifest contains duplicate asset names") + folded.add(name.casefold()) + original = expected.get(name) + if ( + original is None + or type(asset["size"]) is not int + or asset["sha256"] != original["sha256"] + or asset["size"] != original["size"] + or asset["url"] not in {original["url"], f"{ORGANIZATION_BASE}/{name}"} + ): + raise ReadbackError("retained release asset differs from its historical identity") + intake.validate_https(asset["url"]) + observed.add(name) + if observed != set(expected): + raise ReadbackError("retained release identities do not cover the intake manifest") + intake.scan_public_json(value, RELEASE_MANIFEST) + + +def verify_metadata_archive(root: Path, files: list[str], archive_path: Path) -> None: + """Read regular TAR members without extracting or executing archive contents.""" + expected = set(files) + if len(expected) != len(files): + raise ReadbackError("public metadata inventory contains duplicate file names") + seen = [] + folded = set() + with tarfile.open(archive_path, "r|gz") as archive: + for member in archive: + name = intake.safe_relative(member.name) + if name.casefold() in folded: + raise ReadbackError("metadata archive contains duplicate portable member names") + folded.add(name.casefold()) + if name not in expected or len(seen) >= len(expected): + raise ReadbackError("metadata archive contains an unexpected member") + if not member.isreg() or member.issparse(): + raise ReadbackError("metadata archive members must be ordinary regular files") + if ( + member.mode, member.mtime, member.uid, member.gid, + member.uname, member.gname, + ) != (0o644, 0, 0, 0, "", ""): + raise ReadbackError("metadata archive member attributes are not canonical") + source = intake.contained_regular(root, name) + if member.size != source.stat().st_size: + raise ReadbackError("metadata archive member differs from its public pinned size") + stream = archive.extractfile(member) + if stream is None: + raise ReadbackError("metadata archive member cannot be read") + with stream: + observed_digest = hashlib.file_digest(stream, "sha256").hexdigest() + if observed_digest != intake.digest(source): + raise ReadbackError("metadata archive member differs from its public pinned bytes") + seen.append(name) + if seen != sorted(expected): + raise ReadbackError("metadata archive inventory or deterministic member order differs") + + +def verify_release_readback( + root: Path, + tooling: Path, + expected_directory: Path, + downloaded_directory: Path, +) -> dict: + """Verify provenance, the exact 60 asset names, all bytes and every checksum.""" + manifest, metadata_files = intake.verify_intake(root, tooling) + raw_names = {asset["name"] for asset in manifest["assets"]} + base_names = raw_names | ANCILLARY_FILES + expected_names = base_names | {f"{name}.sha256" for name in base_names} + if len(raw_names) != intake.ARCHIVE_COUNT or len(expected_names) != 60: + raise ReadbackError("Python intake release requires exactly 60 distinct asset names") + expected = inventory(expected_directory, expected_names) + downloaded = inventory(downloaded_directory, expected_names) + + # A downloaded checksum file alone is not a trust anchor. Raw archives use + # historical byte pins; ancillary files use the validated local release. + intake_source = intake.contained_regular(root, INTAKE_MANIFEST) + intake_pin = (intake.digest(intake_source), intake_source.stat().st_size) + for group in (expected, downloaded): + intake.verify_bytes(group[INTAKE_MANIFEST], *intake_pin) + validate_release_manifest(intake.read_json(group[RELEASE_MANIFEST]), manifest) + for asset in manifest["assets"]: + intake.verify_bytes(group[asset["name"]], asset["sha256"], asset["size"]) + verify_checksum_companion(group, asset["name"], asset["sha256"]) + + ancillary_pins = {} + for name in sorted(ANCILLARY_FILES): + sha256 = intake.digest(expected[name]) + size = expected[name].stat().st_size + intake.verify_bytes(downloaded[name], sha256, size) + for group in (expected, downloaded): + verify_checksum_companion(group, name, sha256) + ancillary_pins[name] = {"sha256": sha256, "size": size} + verify_metadata_archive(root, metadata_files, expected[METADATA_ARCHIVE]) + verify_metadata_archive(root, metadata_files, downloaded[METADATA_ARCHIVE]) + return { + "status": "verified-release-asset-readback", + "release_asset_count": len(expected_names), + "raw_archive_assets": len(raw_names), + "total_pinned_raw_archive_bytes": manifest["total_size"], + "metadata_files": len(metadata_files), + "ancillary_pins": ancillary_pins, + "github_immutable_release_state": "not-checked-by-local-file-verifier", + "native_runtime_acceptance": "not-run", + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, required=True) + parser.add_argument("--tooling", type=Path, required=True) + parser.add_argument("--expected-dir", type=Path, required=True) + parser.add_argument("--downloaded-dir", type=Path, required=True) + args = parser.parse_args() + report = verify_release_readback( + args.root, args.tooling, args.expected_dir, args.downloaded_dir, + ) + print(json.dumps(report, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except (ReadbackError, intake.IntakeError, OSError, ValueError, KeyError, TypeError, tarfile.TarError) as error: + raise SystemExit(f"release readback failed ({type(error).__name__}): {error}") from error