Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
25 kB · 543 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544#!/usr/bin/env python3"""Verify an already-extracted source archive against the resolved source index.
The source archive is the corresponding source for the two host binaries, but itdoes not carry a per-file hash index of its own. ``SOURCE-PROVENANCE.json``*inside* the archive records only aggregate counts and digests. The per-fileinventory — the size and SHA-256 of every member — is ``source-manifest.json``,which the packager writes into the release directory and copies into the binaryarchive.
This tool joins the two. It takes an extracted source tree and that resolvedindex, re-hashes every member the index records, requires the members thepackager generates (``SOURCE-PROVENANCE.json``, and ``vendor-sources.toml`` whenthe archive is vendored), and refuses any file, directory or symlink the indexdoes not list.
What it proves: every member the index names is present with the recorded sizeand SHA-256, the re-hashed first-party and vendored aggregates match the index,the directory membership of the tree is closed (no file, directory or symlink theindex does not list), and the packager's generated members are present by name.What it does not prove: it never sees the source archive, so it does notauthenticate the archive, index or publisher. ``sha256sum -c SHA256SUMS`` beforeextraction checks archive integrity against the supplied checksums; it is notindependent publisher authentication. Every check here assumes a quiescentextraction of those checked bytes that nothing writes to while this runs. The shared readers open a memberwithout following a symlink leaf and refuse a non-regular file, but this toolmakes no claim about a concurrent writer.
The packager's generated members — ``SOURCE-PROVENANCE.json`` and, when thearchive is vendored, ``vendor-sources.toml`` — are required by presence only:their bytes are not in the index the archive ships against. This tool cannot seea change to their content unless the archive's SHA-256 was checked first, so the``generated_members`` result is evidence of presence, not of content.``sys.dont_write_bytecode`` is set before anything is imported from the tree, sonothing this tool does leaves a ``__pycache__`` behind and mutates the bytes it isverifying; the documented command also passes ``python3 -B`` as a second guard.
It reads files only: no network, no subprocesses, no source commands, notoolchain and no device. Every step is bounded and fails closed.
Usage:
python3 -B verify_extracted_source.py --source-root <extracted-source-tree> \\ --manifest <release-dir>/source-manifest.json
``--source-root`` defaults to the current directory, so from inside theextracted source tree the index can be given by its sibling path."""
from __future__ import annotations
import sys
# Before importing anything from the extracted tree: importing a module that# lives in it would otherwise write a ``__pycache__`` directory into the very# bytes this tool is about to verify, and the tree would then fail its own# closure check.sys.dont_write_bytecode = True
import argparse # noqa: E402import hashlib # noqa: E402import json # noqa: E402import os # noqa: E402import re # noqa: E402import stat # noqa: E402from pathlib import Path # noqa: E402
_HERE = Path(__file__).resolve().parentif str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE))
import build_release_source as build # noqa: E402 - the closed-inventory helper, reused not copiedimport package_release as packaging # noqa: E402 - the shared bounded readers and schemaimport verify_release_local as verify # noqa: E402 - bounded file helpers
EXIT_OK = 0EXIT_FAILED = 1PROVENANCE_NAME = "SOURCE-PROVENANCE.json"VENDOR_CONFIG_NAME = "vendor-sources.toml"PROVENANCE_RECORD_KIND = "release_source_provenance"SOURCE_RECORD_KIND = "release_source_manifest_resolved"# The index is a JSON document bounded like every other shared reader; the real# one is a few megabytes even with the whole vendored crate tree indexed.MAX_INDEX_BYTES = verify.MAX_JSON_BYTESMAX_REPORTED_PROBLEMS = 20READ_CHUNK = 1 << 20HEX64 = re.compile(r"[0-9a-f]{64}")
def _is_int(value) -> bool: """A real integer: ``bool`` is an ``int`` subclass and must not pass.""" return isinstance(value, int) and not isinstance(value, bool)
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="verify_extracted_source.py", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--source-root", type=Path, default=Path("."), help="the extracted source tree (the directory the source archive unpacks to); default: the current directory", ) parser.add_argument( "--manifest", type=Path, required=True, help="the resolved source index (source-manifest.json in the release directory or the binary archive)", ) parser.add_argument( "--json", action="store_true", help="print the result as a JSON record instead of a human summary", ) return parser.parse_args(argv)
# ---------------------------------------------------------------------------# Reading the index: every field is type-checked before it is used# ---------------------------------------------------------------------------
def _object_list(value, label: str, problems: list[str]) -> list: """A list of JSON objects, or a recorded problem for anything else.""" if value is None: return [] if not isinstance(value, list): problems.append(f"the source index {label} is not a list") return [] entries = [] for index, entry in enumerate(value): if not isinstance(entry, dict): problems.append(f"the source index {label}[{index}] is not an object") continue entries.append(entry) return entries
def _canonical_relative(value, label: str, problems: list[str]) -> str | None: """A member path, relative and canonical, or a recorded problem.""" if not isinstance(value, str) or not value: problems.append(f"{label} is not a non-empty string: {value!r}") return None reason = packaging.unsafe_member_name(value) if reason: problems.append(f"{label} {value!r} is not a canonical member path ({reason})") return None if any(ord(char) < 0x20 or ord(char) == 0x7F for char in value): problems.append(f"{label} {value!r} contains a control character") return None if any(part in ("", ".", "..") for part in value.split("/")): problems.append(f"{label} {value!r} is not a canonical member path") return None return value
def _digest(value, label: str, problems: list[str]) -> str | None: if not isinstance(value, str) or not HEX64.fullmatch(value): problems.append(f"{label} is not a lowercase SHA-256: {value!r}") return None return value
def _size(value, label: str, problems: list[str]) -> int | None: if isinstance(value, bool) or not isinstance(value, int) or value < 0: problems.append(f"{label} is not a non-negative integer size: {value!r}") return None return value
def _indexed_members(manifest: dict, prefix: str, problems: list[str]) -> tuple[dict, list[str], list]: """Every member the resolved index names, with its recorded size and hash.
The four categories — first-party files, distribution-document aliases, vendored crate files and runtime corresponding-source members — are merged into one mapping. A name that appears twice is refused rather than last-one- wins, because an overwritten entry would silently drop a file's obligation. """ expected: dict[str, dict] = {} vendor_files: list[dict] = []
def add(full: str, entry: dict, label: str) -> None: if full in expected: problems.append(f"the source index names {full!r} more than once") return digest = _digest(entry.get("sha256"), f"{label} SHA-256", problems) size = _size(entry.get("size"), f"{label} size", problems) if digest is not None and size is not None: expected[full] = {"sha256": digest, "size": size}
for entry in _object_list(manifest.get("files"), "files", problems): path = _canonical_relative(entry.get("path"), "files[].path", problems) if path: add(f"{prefix}/{path}", entry, f"source file {path}")
for entry in _object_list(manifest.get("distribution_docs"), "distribution_docs", problems): path = _canonical_relative(entry.get("archive_path"), "distribution_docs[].archive_path", problems) if path: add(f"{prefix}/{path}", entry, f"distribution document {path}")
vendor = manifest.get("vendor") if vendor is not None and not isinstance(vendor, dict): problems.append("the source index vendor record is not an object") vendor = None if vendor: vendor_files = _object_list(vendor.get("files"), "vendor.files", problems) for entry in vendor_files: path = _canonical_relative(entry.get("path"), "vendor.files[].path", problems) if path: add(f"{prefix}/vendor/{path}", entry, f"vendored file {path}")
runtime = manifest.get("runtime_bundle") if runtime is not None and not isinstance(runtime, dict): problems.append("the source index runtime_bundle record is not an object") runtime = None if runtime: present = runtime.get("present") if not isinstance(present, bool): problems.append(f"the source index runtime_bundle present flag is not a boolean: {present!r}") elif present: label = "runtime_bundle.source_members" for entry in _object_list(runtime.get("source_members"), label, problems): path = _canonical_relative(entry.get("path"), f"{label}[].path", problems) if path: add(f"{prefix}/{path}", entry, f"runtime source member {path}")
required = [f"{prefix}/{PROVENANCE_NAME}"] if vendor_files: required.append(f"{prefix}/{VENDOR_CONFIG_NAME}") return expected, required, vendor_files
# ---------------------------------------------------------------------------# Inspecting the extracted tree# ---------------------------------------------------------------------------
def _bounded_tree(root: Path, allowed_files: set[str], max_nodes: int) -> dict: """Walk the tree with a hard node budget, refusing the first stray entry.
Entries are streamed with ``os.scandir`` one at a time and a directory listing is never materialised, so a directory holding an enormous number of entries cannot exhaust memory before the budget is checked. The pending-directory stack is capped by the same budget, directory symlinks are not followed, and the first file, directory or link the index does not list stops the walk rather than being collected. """ allowed_dirs = set() for name in allowed_files: parts = name.split("/") for index in range(1, len(parts)): allowed_dirs.add("/".join(parts[:index]))
files = 0 directories = 0 visited = 0 pending: list[tuple[Path, str]] = [(root, "")] while pending: current, prefix = pending.pop() try: entries = os.scandir(current) except OSError as error: raise verify.HarnessError(f"the extracted tree could not be read at {prefix or '.'}: {error}") from error with entries: for entry in entries: visited += 1 if visited > max_nodes: raise verify.HarnessError( f"the extracted tree has more than {max_nodes} entries; refusing to walk it further" ) relative = entry.name if not prefix else f"{prefix}/{entry.name}" if entry.is_symlink(): raise verify.HarnessError(f"the extracted tree contains a symbolic link: {relative}") if entry.is_dir(follow_symlinks=False): if relative not in allowed_dirs: raise verify.HarnessError( f"the extracted tree has a directory the source index does not list: {relative}" ) directories += 1 if len(pending) >= max_nodes: raise verify.HarnessError( f"the extracted tree has more than {max_nodes} pending directories" ) pending.append((Path(entry.path), relative)) else: if relative not in allowed_files: raise verify.HarnessError( f"the extracted tree has a file the source index does not list: {relative}" ) files += 1 return {"files": files, "directories": directories}
def _lstat_member(path: Path, relative: str, problems: list[str]) -> os.stat_result | None: """Lstat one member and refuse anything that is not a plain regular file.""" try: info = os.lstat(path) except FileNotFoundError: problems.append(f"{relative} is in the source index but missing from the extracted tree") return None except OSError as error: problems.append(f"{relative} could not be inspected: {error}") return None if stat.S_ISLNK(info.st_mode): problems.append(f"{relative} is a symbolic link; the extracted source must be plain files") return None if not stat.S_ISREG(info.st_mode): problems.append(f"{relative} is not a regular file") return None if info.st_nlink != 1: problems.append(f"{relative} has {info.st_nlink} hard links; the extracted source must be plain files") return info
def _hash_member(path: Path, relative: str, expected_size: int, remaining_budget: int, problems: list[str]) -> tuple[int, str] | None: """Hash one member, reading at most one byte past its declared size.
The read is bounded by the declared size plus one byte, so a file that grew after it was opened is refused on the extra byte instead of being streamed to EOF, and by the remaining aggregate budget, so the sum of every member's bytes stays inside the same limit the archive audit uses. """ if expected_size > remaining_budget: problems.append(f"the extracted tree is above the {verify.MAX_ARCHIVE_BYTES} byte bound") return None limit = expected_size + 1 digest = hashlib.sha256() total = 0 try: with packaging.open_regular_readonly( path, max_bytes=verify.MAX_MEMBER_BYTES, description=f"extracted member {relative}" ) as handle: while total < limit: chunk = handle.read(min(READ_CHUNK, limit - total)) if not chunk: break total += len(chunk) digest.update(chunk) except packaging.CheckFailure as error: problems.append(str(error)) return None if total > expected_size: problems.append(f"{relative} is larger than the {expected_size} bytes the source index records") return None return total, digest.hexdigest()
def _provenance_problems(source_root: Path, manifest: dict, first_party: list) -> list[str]: """Cross-check the archive's own bound record against the resolved index.""" problems: list[str] = [] path = source_root / PROVENANCE_NAME if not path.is_file(): return problems # presence is already required by the inventory try: document = verify.load_json_file(path, PROVENANCE_NAME) except verify.HarnessError as error: problems.append(str(error)) return problems if not _is_int(document.get("schema")) or document.get("schema") != packaging.SCHEMA: problems.append( f"{PROVENANCE_NAME} is schema {document.get('schema')!r}, expected the integer {packaging.SCHEMA}" ) if document.get("record_kind") != PROVENANCE_RECORD_KIND: problems.append( f"{PROVENANCE_NAME} record_kind is {document.get('record_kind')!r}, expected {PROVENANCE_RECORD_KIND!r}" ) files = document.get("files") if not isinstance(files, dict): problems.append(f"{PROVENANCE_NAME} files record is not an object") return problems count = files.get("count") if not _is_int(count): problems.append(f"{PROVENANCE_NAME} records a non-integer file count: {count!r}") elif count != len(first_party): problems.append(f"{PROVENANCE_NAME} records {count} first-party files, the index lists {len(first_party)}") if files.get("aggregate_sha256") != manifest.get("aggregate_sha256"): problems.append(f"{PROVENANCE_NAME} records a different first-party aggregate than the source index") return problems
def verify_extracted_tree(source_root: Path, manifest_path: Path) -> dict: """Re-hash an extracted source tree against the resolved index, or raise.""" source_root = build.normalize_path(source_root) build.reject_symlink_chain(source_root, "--source-root") if not source_root.is_dir(): raise verify.HarnessError(f"--source-root {source_root} is not a directory")
manifest = verify.load_json_file( build.normalize_path(manifest_path), "source index", max_bytes=MAX_INDEX_BYTES )
shape: list[str] = [] if not _is_int(manifest.get("schema")) or manifest.get("schema") != packaging.SCHEMA: shape.append(f"the source index is schema {manifest.get('schema')!r}, expected the integer {packaging.SCHEMA}") if manifest.get("record_kind") != SOURCE_RECORD_KIND: shape.append( f"the source index record_kind is {manifest.get('record_kind')!r}, expected {SOURCE_RECORD_KIND!r}; " "point --manifest at the resolved source-manifest.json, not at the static repository manifest" ) archive = manifest.get("archive") if not isinstance(archive, dict): shape.append("the source index archive record is not an object") archive = {} prefix = archive.get("prefix") if not isinstance(prefix, str) or not prefix or packaging.unsafe_member_name(prefix + "/x"): shape.append(f"the source index records no usable archive prefix: {prefix!r}") if not isinstance(manifest.get("files"), list): shape.append("the source index files record is not a list") elif not manifest.get("files"): shape.append("the source index lists no files") if not isinstance(manifest.get("aggregate_sha256"), str) or not HEX64.fullmatch(manifest.get("aggregate_sha256", "")): shape.append("the source index records no first-party aggregate digest") if shape: raise verify.HarnessError("the resolved source index cannot be used:\n" + "\n".join(f" {item}" for item in shape))
if source_root.name != prefix: raise verify.HarnessError( f"--source-root {source_root} is named {source_root.name!r}, but the source index records the archive " f"prefix {prefix!r}; extract the source archive and point --source-root at the directory it unpacks to" )
problems: list[str] = [] expected, required, vendor_files = _indexed_members(manifest, prefix, problems) first_party = manifest["files"] if not problems: # Only now, with every entry validated as an object with a canonical path # and a digest, is it safe to canonicalise the lists for the aggregate # comparison. if verify.aggregate_digest(first_party) != manifest.get("aggregate_sha256"): problems.append("the source index's own file list does not match the aggregate it records") vendor = manifest.get("vendor") or {} if vendor_files and verify.aggregate_digest(vendor_files) != vendor.get("aggregate_sha256"): problems.append("the source index's own vendored file list does not match the aggregate it records") if problems: raise verify.HarnessError( "the resolved source index is not well-formed:\n" + "\n".join(f" {item}" for item in problems[:MAX_REPORTED_PROBLEMS]) )
# The shared closed inventory the build gate uses must name exactly the same # members this tool read out of the index. shared = build.closed_inventory(manifest) if set(shared["expected"]) != set(expected) or set(shared["required"]) != set(required): raise verify.HarnessError( "the shared closed inventory disagrees with the member set this tool read from the source index" )
head = prefix + "/" expected_rel = {name[len(head):]: entry for name, entry in expected.items()} required_rel = [name[len(head):] for name in required] allowed = set(expected_rel) | set(required_rel) if len(allowed) > verify.MAX_MEMBERS: raise verify.HarnessError(f"the source index names {len(allowed)} members, above the {verify.MAX_MEMBERS} bound")
walked = _bounded_tree(source_root, allowed, verify.MAX_MEMBERS)
observed: dict[str, str] = {} consumed = 0 for relative in sorted(expected_rel): entry = expected_rel[relative] path = source_root / relative info = _lstat_member(path, relative, problems) if info is None: continue if info.st_size != entry["size"]: problems.append(f"{relative} is {info.st_size} bytes; the source index records {entry['size']}") continue hashed = _hash_member(path, relative, entry["size"], verify.MAX_ARCHIVE_BYTES - consumed, problems) if hashed is None: continue consumed += hashed[0] if hashed[1] != entry["sha256"]: problems.append(f"{relative} does not match the SHA-256 the source index records") continue observed[relative] = hashed[1]
for relative in required_rel: _lstat_member(source_root / relative, relative, problems)
# Recompute the aggregates from the *observed* bytes over the exact subsets # the index aggregates: first-party paths, and vendored paths prefixed with # ``vendor/``. Both are only meaningful once every member of that subset was # hashed; a member that failed above is already reported. first_party_paths = [entry["path"] for entry in manifest["files"]] if all(path in observed for path in first_party_paths): rehashed = [{"path": path, "sha256": observed[path]} for path in first_party_paths] if verify.aggregate_digest(rehashed) != manifest["aggregate_sha256"]: problems.append("the re-hashed first-party aggregate does not match the aggregate the source index records") vendor = manifest.get("vendor") or {} vendor_paths = {entry["path"]: f"vendor/{entry['path']}" for entry in vendor_files} if vendor_files and all(member in observed for member in vendor_paths.values()): rehashed = [ {"path": path, "sha256": observed[member]} for path, member in sorted(vendor_paths.items()) ] if verify.aggregate_digest(rehashed) != vendor.get("aggregate_sha256"): problems.append("the re-hashed vendored aggregate does not match the aggregate the source index records")
problems.extend(_provenance_problems(source_root, manifest, first_party))
if problems: raise verify.HarnessError( "the extracted source does not match the resolved source index:\n" + "\n".join(f" {item}" for item in problems[:MAX_REPORTED_PROBLEMS]) ) return { "source_root": str(source_root), "prefix": prefix, "members": len(expected_rel), "tree_files": walked["files"], "generated_members": sorted(required_rel), "hashed_bytes": consumed, "aggregate_sha256": manifest["aggregate_sha256"], "result": "verified", }
def main(argv: list[str] | None = None) -> int: args = parse_arguments(argv) try: result = verify_extracted_tree(args.source_root, args.manifest) except verify.HarnessError as error: print(f"verify_extracted_source: {error}", file=sys.stderr) return EXIT_FAILED except (OSError, ValueError, TypeError, KeyError) as error: # Defensive: a filesystem or shape surprise is reported, never a traceback. print(f"verify_extracted_source: {type(error).__name__}: {error}", file=sys.stderr) return EXIT_FAILED if args.json: print(json.dumps(result, indent=2, sort_keys=True)) else: print( f"verified {result['members']} members ({result['hashed_bytes']} bytes) against the source index " f"for {result['prefix']}" ) print("generated members required and present: " + ", ".join(result["generated_members"])) return EXIT_OK
if __name__ == "__main__": sys.exit(main())