#!/usr/bin/env python3 """Verify the complete metadata-only codec backport; Python 3.11+, stdlib only. Usage: python3 vendor/verify.py --root vendor [--self-test] Trust anchors are the published archive SHA256 entries in baseline e4cfd4c. Verification never extracts archives or repairs files. """ import argparse import copy import hashlib import json from pathlib import Path, PurePosixPath import shutil import tarfile import tempfile import tomllib BASELINE = "e4cfd4c" ARCHIVES = { "bincode": "36eaf5d7b090263e8150820482d5d93cd964a81e4019913c972f4edcc6edb740", "bincode_derive": "bf95709a440f45e986983918d0e8a1f30a9b1df04918fc828670606804ac3c09", } SOURCE = "registry+https://github.com/rust-lang/crates.io-index" UPSTREAM = "https://github.com/bincode-org/bincode" def require(condition, message): if not condition: raise ValueError(message) def digest(data): return hashlib.sha256(data).hexdigest() def deltas(name): changes = [{"path": ["package", "rust-version"], "before": "1.85.0", "after": "1.75"}] if name == "bincode": changes.append({"path": ["dependencies", "bincode_derive", "version"], "before": "2.0.1", "after": "=2.0.1"}) return changes def verify_archive_checksums(root): require(not root.is_symlink(), "vendor root is a symlink") require(not (root / "provenance.json").is_symlink(), "provenance symlink") require(not (root / "upstream").is_symlink(), "archive directory symlink") provenance = json.loads((root / "provenance.json").read_text()) require(set(provenance) == {"baseline", "packages"}, "unexpected provenance fields") require(provenance["baseline"] == BASELINE, "baseline mismatch") records = provenance["packages"] require(len(records) == 2 and {p["name"] for p in records} == set(ARCHIVES), "package inventory mismatch") expected_archives = {name + "-2.0.1.crate" for name in ARCHIVES} require({p.name for p in (root / "upstream").iterdir()} == expected_archives, "archive file set mismatch") originals = {} for record in records: name = record["name"] stem = name + "-2.0.1" require(set(record) == {"name", "version", "archive", "sha256", "source", "license", "upstream", "files", "allowed_manifest_deltas"}, "unexpected package provenance fields") require(record["version"] == "2.0.1" and record["archive"] == "upstream/" + stem + ".crate", "package identity mismatch") require(record["sha256"] == ARCHIVES[name], "inventory archive checksum mismatch") require(record["source"] == SOURCE and record["license"] == "MIT" and record["upstream"] == UPSTREAM, "source/license provenance mismatch") require(record["allowed_manifest_deltas"] == deltas(name), "manifest delta inventory mismatch") archive = root / record["archive"] require(archive.is_file() and not archive.is_symlink(), "unsafe archive file") require(digest(archive.read_bytes()) == ARCHIVES[name], "archive checksum mismatch: " + name) files = {} with tarfile.open(archive, "r:gz") as tar: for member in tar.getmembers(): path = PurePosixPath(member.name) require(member.isfile() and not path.is_absolute() and ".." not in path.parts and "\\" not in member.name and len(path.parts) > 1 and path.parts[0] == stem and str(path) == member.name, "unsafe archive member: " + member.name) relative = str(path.relative_to(stem)) require(relative not in files, "duplicate archive member") files[relative] = tar.extractfile(member).read() require(record["files"] == {path: digest(data) for path, data in files.items()}, "archive member inventory/checksum mismatch: " + name) require({"LICENSE.md", "Cargo.toml", "Cargo.toml.orig", ".cargo_vcs_info.json"} <= set(files), "missing upstream release/license metadata") originals[name] = files return originals def verify_complete_package_trees(root, originals): for name, files in originals.items(): directory = root / (name + "-2.0.1") require(directory.is_dir() and not directory.is_symlink(), "unsafe package directory") actual = set() actual_dirs = set() expected_dirs = set() for relative in files: expected_dirs.update(str(p) for p in PurePosixPath(relative).parents if str(p) != ".") for path in directory.rglob("*"): require(not path.is_symlink(), "symlink in package tree") relative = path.relative_to(directory).as_posix() if path.is_dir(): actual_dirs.add(relative) else: require(path.is_file(), "special file in package tree") actual.add(relative) require(actual == set(files) and actual_dirs == expected_dirs, "complete package file/directory set mismatch: " + name) for relative, original in files.items(): if relative != "Cargo.toml": require((directory / relative).read_bytes() == original, "upstream byte mismatch: " + name + "/" + relative) def verify_manifest_allowlist(root, originals): for name, files in originals.items(): original = files["Cargo.toml"] before = b'rust-version = "1.85.0"' require(original.count(before) == 1, "unexpected original rust-version text") expected = original.replace(before, b'rust-version = "1.75"') if name == "bincode": before = b'[dependencies.bincode_derive]\nversion = "2.0.1"' require(expected.count(before) == 1, "unexpected original derive text") expected = expected.replace(before, b'[dependencies.bincode_derive]\nversion = "=2.0.1"') actual = (root / (name + "-2.0.1") / "Cargo.toml").read_bytes() require(actual == expected, "textual manifest allowlist mismatch: " + name) parsed = tomllib.loads(original.decode()) require(parsed["package"]["name"] == name and parsed["package"]["version"] == "2.0.1" and parsed["package"]["license"] == "MIT", "upstream manifest identity mismatch") modified = copy.deepcopy(parsed) for change in deltas(name): table = modified for key in change["path"][:-1]: table = table[key] key = change["path"][-1] require(table[key] == change["before"], "original parsed manifest mismatch") table[key] = change["after"] require(tomllib.loads(actual.decode()) == modified, "parsed manifest allowlist mismatch: " + name) # The original author manifest is retained byte-for-byte, not rewritten. tomllib.loads(files["Cargo.toml.orig"].decode()) def verify(root): originals = verify_archive_checksums(root) verify_complete_package_trees(root, originals) verify_manifest_allowlist(root, originals) return originals def self_test(root): def source_edit(tree): path = tree / "bincode-2.0.1/src/lib.rs" path.write_bytes(path.read_bytes() + b"\n// altered\n") def license_remove(tree): (tree / "bincode-2.0.1/LICENSE.md").unlink() def source_extra(tree): (tree / "bincode_derive-2.0.1/src/extra.rs").write_text("// extra\n") def manifest_edit(tree): path = tree / "bincode-2.0.1/Cargo.toml" path.write_bytes(path.read_bytes().replace(b'edition = "2021"', b'edition = "2018"')) def archive_edit(tree): path = tree / "upstream/bincode-2.0.1.crate" path.write_bytes(path.read_bytes() + b"altered") def checksum_edit(tree): path = tree / "provenance.json" data = json.loads(path.read_text()) data["packages"][0]["sha256"] = "0" * 64 path.write_text(json.dumps(data)) def inventory_edit(tree): path = tree / "provenance.json" data = json.loads(path.read_text()) data["packages"][0]["files"]["src/lib.rs"] = "0" * 64 path.write_text(json.dumps(data)) with tempfile.TemporaryDirectory(prefix="hexwork-vendor-verifier-") as temp: for mutate in (None, source_edit, license_remove, source_extra, manifest_edit, archive_edit, checksum_edit, inventory_edit): tree = Path(temp) / "vendor" shutil.copytree(root, tree) if mutate: mutate(tree) try: verify(tree) except ValueError: print("self-test rejected:", mutate.__name__) else: raise ValueError("negative control accepted: " + mutate.__name__) else: verify(tree) print("self-test accepted: pristine copy") shutil.rmtree(tree) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, required=True) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() try: originals = verify(args.root) print("PASS verify_archive_checksums, verify_complete_package_trees, verify_manifest_allowlist:", sum(map(len, originals.values())), "authenticated files") if args.self_test: self_test(args.root) except (ValueError, OSError, KeyError, TypeError, tarfile.TarError) as error: parser.exit(1, "FAIL vendor verification: " + str(error) + "\n") if __name__ == "__main__": main()