Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
93 kB · 2068 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069#!/usr/bin/env python3"""Independently re-verify a release directory written by tools/package_release.py.
This checker trusts nothing the packager asserted. It re-reads every archivemember, re-measures both binaries from the bytes inside the archive, re-derivesthe privacy findings, re-validates the recorded build environment against therecorded policy, and re-reads the Git state. A claimed check is only accepted ifthis run reproduces it.
Two check severities:
``integrity`` A fact about the bytes. Any failure fails the check in both modes, because a release directory whose hashes or contents do not match its manifest is broken whatever the release intends to be.
``policy`` A release-process requirement: annotated version tag, clean worktree, reserved preview version, completed third-party inventory, recorded clean rebuild, resolved privacy review. These are reported as ``pending`` in development mode and block in release mode.
Modes:
``--mode development`` (default) Report everything and exit 0 unless an integrity check failed.
``--mode release`` The publication preflight. Every integrity and policy check must be ``verified``; otherwise exit 1 and list the blockers.
Exit codes: 0 passed, 1 checks failed, 2 usage error, 3 internal error."""
from __future__ import annotations
import argparseimport hashlibimport jsonimport sysimport tarfileimport tempfilefrom pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import package_release as packaging # noqa: E402
EXIT_OK = 0EXIT_FAILED = 1EXIT_USAGE = 2EXIT_INTERNAL = 3
INTEGRITY = "integrity"POLICY = "policy"
VERIFIED = "verified"PENDING = "pending"NOT_VERIFIED = "not_verified"NOT_CHECKED = "not_checked"
STATE_ORDER = (VERIFIED, PENDING, NOT_VERIFIED, NOT_CHECKED)TAR_SCAN_LIMIT = 1 << 30# A finite cap on the runtime archive members a re-read will enumerate, so an# archive that declares an unbounded number of unknown extras is refused rather# than walked indefinitely.RUNTIME_MEMBER_CAP = 200_000
class Report: def __init__(self) -> None: self.checks: list[dict] = []
def add(self, check: str, state: str, severity: str, detail: str) -> None: if state not in STATE_ORDER: raise ValueError(f"unknown check state {state!r}") self.checks.append({"check": check, "state": state, "severity": severity, "detail": detail})
def integrity_failures(self) -> list[dict]: return [ item for item in self.checks if item["severity"] == INTEGRITY and item["state"] in (NOT_VERIFIED, "failed") ]
def policy_pending(self) -> list[dict]: return [item for item in self.checks if item["severity"] == POLICY and item["state"] != VERIFIED]
def counted(self) -> dict: counts = dict.fromkeys(STATE_ORDER, 0) for item in self.checks: counts[item["state"]] += 1 return counts
def load_release_manifest(release_dir: Path, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS) -> dict | None: path = release_dir / "release-manifest.json" if not path.exists(): report.add("release_manifest", NOT_VERIFIED, INTEGRITY, f"{path} is missing") return None try: manifest = json.loads( packaging.read_regular_bytes( path, max_bytes=limits.max_manifest_bytes, description="release-manifest.json" ) ) except packaging.CheckFailure as error: report.add("release_manifest", NOT_VERIFIED, INTEGRITY, str(error)) return None except json.JSONDecodeError as error: report.add("release_manifest", NOT_VERIFIED, INTEGRITY, f"{path} is not valid JSON: {error}") return None required = ( "schema_version", "record_kind", "product", "version", "target", "baseline", "source", "build", "artifacts", "archive", "source_archive", "source_completeness", "inventory", "privacy", "verification", "release_ready", "publication", "release_policy", ) missing = [key for key in required if key not in manifest] if missing: report.add("release_manifest", NOT_VERIFIED, INTEGRITY, f"missing required fields: {', '.join(missing)}") return None if manifest.get("schema_version") != 1 or manifest.get("record_kind") != "release_manifest": report.add( "release_manifest", NOT_VERIFIED, INTEGRITY, f"expected schema_version 1 release_manifest, got {manifest.get('schema_version')!r} " f"{manifest.get('record_kind')!r}", ) return None report.add( "release_manifest", VERIFIED, INTEGRITY, f"schema 1 release manifest for {manifest['product']} {manifest['version']}", ) return manifest
def check_checksums(release_dir: Path, manifest: dict, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS) -> dict: path = release_dir / "SHA256SUMS" if not path.exists(): report.add("checksums", NOT_VERIFIED, INTEGRITY, "SHA256SUMS is missing") return {} try: text = packaging.read_regular_bytes(path, max_bytes=limits.max_manifest_bytes, description="SHA256SUMS").decode( "utf-8", "replace" ) except packaging.CheckFailure as error: report.add("checksums", NOT_VERIFIED, INTEGRITY, str(error)) return {} listed: dict[str, str] = {} problems = [] for line in text.splitlines(): line = line.strip() if not line: continue digest, _, name = line.partition(" ") name = name.strip() digest = digest.strip() if not name or packaging.unsafe_member_name(name, limits): problems.append(f"unsafe checksum entry name: {name!r}") continue if name in listed and listed[name] != digest: problems.append(f"conflicting duplicate SHA256SUMS entries for {name}") continue listed[name] = digest for name, digest in sorted(listed.items()): candidate = release_dir / name if not candidate.exists(): problems.append(f"{name} is listed but missing") continue try: actual = packaging.anchored_sha256( candidate, max_bytes=limits.max_archive_bytes, description=f"listed file {name}" ) except packaging.CheckFailure as error: problems.append(str(error)) continue if actual != digest: problems.append(f"{name}: SHA256SUMS says {digest[:16]}…, file is {actual[:16]}…") for name in (manifest["archive"]["name"], manifest["source_archive"]["name"], "release-manifest.json"): if name not in listed: problems.append(f"{name} is not covered by SHA256SUMS") recorded = manifest["archive"].get("sha256") if recorded and listed.get(manifest["archive"]["name"]) != recorded: problems.append("the binary archive hash disagrees with release-manifest.json") source_recorded = manifest["source_archive"].get("sha256") if source_recorded and listed.get(manifest["source_archive"]["name"]) != source_recorded: problems.append("the source archive hash disagrees with release-manifest.json") if problems: report.add("checksums", NOT_VERIFIED, INTEGRITY, "; ".join(problems)) else: report.add("checksums", VERIFIED, INTEGRITY, f"{len(listed)} entries recomputed from the files") return listed
def check_mandatory_artifacts(manifest: dict, report: Report) -> None: """The mandatory artifact names must be present and bound by hash.""" missing = packaging.mandatory_artifact_names(manifest) problems = [f"missing {item}" for item in missing] for entry in manifest.get("artifacts", []): if entry["kind"] in ("host-binary", "source-archive") and not entry.get("sha256"): problems.append(f"{entry['kind']} {entry['name']} carries no hash") if entry["kind"] == "host-binary" and not (entry.get("elf") or {}).get("machine"): problems.append(f"binary {entry['name']} carries no ELF target metadata") if problems: report.add("mandatory_artifacts", NOT_VERIFIED, INTEGRITY, "; ".join(problems)) else: report.add( "mandatory_artifacts", VERIFIED, INTEGRITY, "both host binaries, the source archive, the binary archive and both manifests are named and hashed", )
def open_archives( release_dir: Path, manifest: dict, report: Report, limits: packaging.Limits, passed: dict[str, dict] | None = None,) -> dict: """Audit both archives; only archives that pass are recorded as usable.
``passed`` receives the audits that succeeded, so every later reader of the same archive can be gated on it instead of re-opening a file that already failed its own safety check. """ archives = {} for key in ("archive", "source_archive"): info = manifest.get(key, {}) name = info.get("name") if not name: report.add("archive_safety", NOT_VERIFIED, INTEGRITY, f"{key} has no name in the manifest") continue path = release_dir / name if not path.exists(): report.add("archive_safety", NOT_VERIFIED, INTEGRITY, f"{name} is missing from the release directory") continue try: audit = packaging.audit_tar(path, limits) except packaging.CheckFailure as error: report.add("archive_safety", NOT_VERIFIED, INTEGRITY, str(error)) continue archives[key] = audit if passed is not None: passed[key] = audit if info.get("sha256") and audit["sha256"] != info["sha256"]: report.add( "archive_safety", NOT_VERIFIED, INTEGRITY, f"{name}: recorded hash does not match the archive bytes", ) continue report.add( "archive_safety", VERIFIED, INTEGRITY, f"{name}: {audit['members']} members, no traversal, no absolute path, no symlink, no non-regular member", ) return archives
def scan_archive(path: Path, prefix: str, limits: packaging.Limits = packaging.DEFAULT_LIMITS) -> tuple[list[dict], dict[str, str]]: """Re-derive privacy findings and per-member hashes for one archive.""" findings: list[dict] = [] hashes: dict[str, str] = {} with packaging.open_bounded_tar(path, limits) as tar: for member in tar: if not member.isreg(): continue handle = tar.extractfile(member) if handle is None: # pragma: no cover - defensive continue digest = hashlib.sha256() carry = b"" identifiers: list[str] = [] total = 0 while True: chunk = handle.read(1 << 20) if not chunk: break total += len(chunk) if total > TAR_SCAN_LIMIT: # pragma: no cover - defensive bound break digest.update(chunk) window = carry + chunk for identifier in packaging.scan_bytes(window): if identifier not in identifiers: identifiers.append(identifier) carry = window[-packaging.SENTINEL_OVERLAP :] relative = member.name if prefix and relative.startswith(prefix + "/"): relative = relative[len(prefix) + 1 :] hashes[relative] = digest.hexdigest() if identifiers: findings.append({"member": member.name, "path": relative, "sentinels": identifiers}) return findings, hashes
def check_archives_contents( release_dir: Path, manifest: dict, source_manifest: dict, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS, binary_audit_passed: bool = True,) -> None: archive_name = manifest["archive"]["name"] source_name = manifest["source_archive"]["name"] source_findings, _ = scan_archive(release_dir / source_name, manifest["source_archive"]["prefix"], limits) if binary_audit_passed: binary_findings, binary_hashes = scan_archive(release_dir / archive_name, manifest["archive"]["prefix"], limits) else: binary_findings, binary_hashes = [], {}
recorded = {(item["path"], item["sentinel"]) for item in manifest.get("privacy", {}).get("findings", [])} observed = {(item["path"], sentinel) for item in source_findings for sentinel in item["sentinels"]} severities = manifest.get("privacy", {}).get("policy", {}) exempt = { (item["path"], item["sentinel"]) for item in manifest.get("privacy", {}).get("exemptions", []) } blocking = sorted( { f"{item['path']} [{sentinel}]" for item in source_findings for sentinel in item["sentinels"] if severities.get(sentinel, "block") == "block" and (item["path"], sentinel) not in exempt } ) if blocking: report.add( "privacy_sentinels", NOT_VERIFIED, INTEGRITY, "blocking sentinel content is present in the shipped source archive: " + "; ".join(blocking[:10]), ) else: unreported = sorted(observed - recorded) if unreported: report.add( "privacy_sentinels", NOT_VERIFIED, INTEGRITY, "sentinel findings present in the source archive that the release manifest does not record: " + "; ".join(f"{path} [{sentinel}]" for path, sentinel in unreported[:10]), ) else: report.add( "privacy_sentinels", VERIFIED if not observed else PENDING, INTEGRITY if not observed else POLICY, f"{len(observed)} recorded review-required finding(s), 0 blocking findings", )
blocking_members = [ item["member"] for item in binary_findings for sentinel in item["sentinels"] if severities.get(sentinel) == "block" and (item["path"], sentinel) not in exempt ] if not binary_audit_passed: report.add( "binary_archive_privacy", NOT_VERIFIED, INTEGRITY, "skipped: the binary archive did not pass its audit", ) elif blocking_members: report.add( "binary_archive_privacy", NOT_VERIFIED, INTEGRITY, "blocking sentinel content is present in the binary archive: " + ", ".join(blocking_members[:10]), ) else: recorded_generated = manifest.get("privacy", {}).get("generated_documents_reviewed", True) report.add( "binary_archive_privacy", VERIFIED if recorded_generated else PENDING, INTEGRITY if recorded_generated else POLICY, f"{len(binary_hashes)} members scanned, no blocking sentinel content", )
files = {entry["path"]: entry for entry in source_manifest.get("files", [])} prefix = manifest["source_archive"]["prefix"] problems = [] if not files: problems.append("source-manifest.json lists no source files") checked = 0 with packaging.open_bounded_tar(release_dir / source_name, limits) as tar: members = {member.name: member for member in tar.getmembers() if member.isreg()}
def verify(relative: str, entry: dict, label: str) -> None: nonlocal checked member = members.get(f"{prefix}/{relative}") if member is None: problems.append(f"{label} {relative} is listed but absent from the source archive") return if member.size != entry.get("size"): problems.append(f"{label} {relative}: archive size {member.size} != recorded {entry.get('size')}") return handle = tar.extractfile(member) digest = hashlib.sha256() while True: chunk = handle.read(1 << 20) if not chunk: break digest.update(chunk) if digest.hexdigest() != entry.get("sha256"): problems.append(f"{label} {relative}: archived bytes do not match the recorded hash") checked += 1
for relative, entry in sorted(files.items()): verify(relative, entry, "source file") for entry in source_manifest.get("distribution_docs") or []: verify(entry["archive_path"], entry, "distribution document") vendor_entries = (source_manifest.get("vendor") or {}).get("files") or [] for entry in vendor_entries: verify(f"vendor/{entry['path']}", entry, "vendored file")
if files and packaging.aggregate_digest(list(files.values())) != source_manifest.get("aggregate_sha256"): problems.append("the source file aggregate does not match the recorded aggregate") vendor = source_manifest.get("vendor", {}) if vendor.get("files") and packaging.aggregate_digest(vendor["files"]) != vendor.get("aggregate_sha256"): problems.append("the vendor file aggregate does not match the recorded aggregate")
generated = { f"{prefix}/SOURCE-PROVENANCE.json", f"{prefix}/vendor-sources.toml", } # A reviewed sanitized distribution document ships at the archive path of the # withheld repository original it replaces; the prepared source manifest # records each alias with the copy's size and SHA-256, verified just above. generated |= { f"{prefix}/{entry['archive_path']}" for entry in source_manifest.get("distribution_docs") or [] } # The runtime corresponding-source bundle stages its own members into the source # archive; check_runtime_source_bundle re-verifies them independently, so they are # expected members here rather than unexplained extras. generated |= runtime_source_archive_members(source_manifest, prefix) extra = sorted(set(members) - {f"{prefix}/{relative}" for relative in files} - generated) unexpected = [ name for name in extra if not (name.startswith(f"{prefix}/vendor/") and name != f"{prefix}/vendor/") ] if vendor.get("files"): expected_vendor = {f"{prefix}/vendor/{entry['path']}" for entry in vendor["files"]} unrecorded_vendor = sorted(name for name in extra if name.startswith(f"{prefix}/vendor/") and name not in expected_vendor) if unrecorded_vendor: problems.append( "vendored members are present in the source archive that the manifest does not record: " + ", ".join(unrecorded_vendor[:5]) ) if unexpected: problems.append( "the source archive has members that are neither allowlisted source, recorded vendor output nor " "generated provenance: " + ", ".join(unexpected[:5]) ) if problems: report.add("source_archive_contents", NOT_VERIFIED, INTEGRITY, "; ".join(problems[:10])) else: report.add( "source_archive_contents", VERIFIED, INTEGRITY, f"{checked} source and vendored members verified member-by-member for size and SHA-256", )
vendor_names = sorted(name for name in members if name.startswith(f"{prefix}/vendor/")) if vendor.get("status") in ("generated_cargo_vendor", "provided") and vendor.get("file_count"): if len(vendor_names) != vendor["file_count"]: report.add( "vendor_contents", NOT_VERIFIED, INTEGRITY, f"vendor tree has {len(vendor_names)} archived files, manifest records {vendor['file_count']}", ) else: report.add( "vendor_contents", VERIFIED, INTEGRITY, f"{len(vendor_names)} vendored crate files ({vendor['status']}) present", ) else: report.add( "vendor_contents", PENDING, POLICY, "no vendored crate sources are included; corresponding source is incomplete", )
def check_binary_archive_members( release_dir: Path, manifest: dict, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS,) -> dict: prefix = manifest["archive"]["prefix"] name = manifest["archive"]["name"] modes: dict[str, int] = {} license_bytes = None problems = [] with packaging.open_bounded_tar(release_dir / name, limits) as tar: for member in tar.getmembers(): if member.name.startswith(f"{prefix}/bin/"): modes[member.name] = member.mode if member.name == f"{prefix}/{manifest['release_policy']['license_file']}": handle = tar.extractfile(member) license_bytes = handle.read() if handle else None for binary in ("ultimate-tool", "ultimate-tool-mcp"): member = f"{prefix}/bin/{binary}" if member not in modes: problems.append(f"{member} is missing from the binary archive") elif not modes[member] & 0o111: problems.append(f"{member} is not marked executable (mode {modes[member]:o})") if license_bytes is None: problems.append("the archive has no LICENSE file") elif b"GNU GENERAL PUBLIC LICENSE" not in license_bytes or b"Version 3" not in license_bytes: problems.append("LICENSE does not look like the full GNU General Public License version 3 text") if problems: report.add("binary_archive_members", NOT_VERIFIED, INTEGRITY, "; ".join(problems)) else: report.add( "binary_archive_members", VERIFIED, INTEGRITY, "both binaries present and executable, full GPL-3.0 license text present", ) return modes
def check_binaries_remeasured( release_dir: Path, manifest: dict, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS,) -> None: """Re-measure both artifacts from the bytes inside the archive.""" baseline = manifest["target"]["glibc_baseline"] expected = {entry["name"]: entry for entry in manifest["artifacts"] if entry["kind"] == "host-binary"} prefix = manifest["archive"]["prefix"] problems = [] details = [] with tempfile.TemporaryDirectory(prefix="ut-release-check-") as workdir: for name, recorded in sorted(expected.items()): member = f"{prefix}/{recorded['path']}" target = Path(workdir) / name try: with packaging.open_bounded_tar(release_dir / manifest["archive"]["name"], limits) as tar: handle = tar.extractfile(member) if handle is None: problems.append(f"{member} is missing from the archive") continue with open(target, "wb") as out: while True: chunk = handle.read(1 << 20) if not chunk: break out.write(chunk) except (KeyError, tarfile.TarError) as error: problems.append(f"{member} could not be read from the archive: {error}") continue try: measured = packaging.measure_elf(target, baseline, limits) except packaging.CheckFailure as error: problems.append(str(error)) continue if measured["sha256"] != recorded["sha256"]: problems.append(f"{name}: measured hash differs from the recorded hash") if measured["size"] != recorded["size"]: problems.append(f"{name}: measured size {measured['size']} differs from {recorded['size']}") for field in ("class", "data", "machine", "type", "linkage", "interpreter", "build_id"): if recorded.get("elf", {}).get(field) != measured[field]: problems.append(f"{name}: {field} differs (recorded {recorded.get('elf', {}).get(field)!r})") if recorded.get("elf", {}).get("glibc_max_required") != measured["glibc_max_required"]: problems.append( f"{name}: recorded glibc requirement {recorded.get('elf', {}).get('glibc_max_required')!r} " f"differs from the measured {measured['glibc_max_required']!r}" ) details.append( f"{name}: {measured['class']} {measured['machine']} {measured['type']}, " f"glibc {measured['glibc_max_required'] or 'unversioned'}" ) if not measured["within_baseline"]: problems.append( f"{name} requires glibc {measured['glibc_max_required']} above the baseline {baseline}" ) if problems: report.add("binary_remeasured", NOT_VERIFIED, INTEGRITY, "; ".join(problems[:10])) else: report.add("binary_remeasured", VERIFIED, INTEGRITY, "; ".join(details))
def check_source_manifest_identity(release_dir: Path, source_manifest: dict, source_root: Path | None, report: Report) -> None: manifest_path = (source_root / packaging.SOURCE_MANIFEST_RELATIVE) if source_root else None if not manifest_path or not manifest_path.exists(): report.add( "source_manifest_identity", NOT_CHECKED, POLICY, "the repository release source manifest is not available; cannot compare its digest with " "the recorded one", ) return recorded = source_manifest.get("manifest", {}).get("sha256") actual = packaging.sha256_file(manifest_path) if recorded != actual: report.add( "source_manifest_identity", NOT_VERIFIED, POLICY, "tools/release_source_manifest.json has changed since this package was produced; " "re-run --prepare-source and --pack", ) else: report.add( "source_manifest_identity", VERIFIED, POLICY, f"{packaging.SOURCE_MANIFEST_RELATIVE} digest matches the packaged record", )
def check_build_environment(manifest: dict, report: Report) -> None: expected = manifest["build"]["expected_environment"] environment = manifest["build"]["environment"] wrapper = {"build_environment": expected, "release": {"build_command": manifest["build"]["command"]}} try: result = packaging.validate_build_environment(environment, wrapper, development=False) except packaging.CheckFailure as error: report.add("build_environment", NOT_VERIFIED, POLICY, str(error).splitlines()[0]) return if manifest["build"]["environment_status"] != result["status"]: report.add( "build_environment", NOT_VERIFIED, POLICY, f"the manifest records environment status {manifest['build']['environment_status']!r}, " f"re-validation says {result['status']!r}", ) return report.add( "build_environment", VERIFIED, POLICY, f"{expected['os']} {expected['os_version_id']}, glibc {expected['glibc']}, rustc {expected['rust_version']}, " f"base {expected['base_image']} pinned", )
def check_elf_target_policy(manifest: dict, report: Report) -> None: expected_triple = manifest["build"]["expected_environment"]["target_triple"] recorded_triple = manifest["target"]["target_triple"] if expected_triple != recorded_triple or manifest["target"]["arch"] != "x86_64": report.add( "target_boundary", NOT_VERIFIED, INTEGRITY, f"target triple {recorded_triple!r} does not match the documented boundary {expected_triple!r}", ) return report.add( "target_boundary", VERIFIED, INTEGRITY, f"linux x86_64 gnu, glibc baseline {manifest['target']['glibc_baseline']}", )
def check_version_policy(manifest: dict, report: Report) -> None: reserved = manifest["release_policy"]["reserved_preview_version"] if manifest["version"] != reserved: report.add( "reserved_preview_version", PENDING, POLICY, f"this archive is version {manifest['version']}; the reserved preview version is {reserved}", ) else: report.add( "reserved_preview_version", VERIFIED, POLICY, f"version {manifest['version']} is the reserved preview version" )
def vendor_execution_problems(vendor: dict) -> list[str]: """Re-derive the declared ``cargo vendor`` execution from the vendor record.
Nothing here can replay the run, so there is no authentic command oracle. The contract is a declared execution bound to the bytes: the recorded status, the command line that produced the tree, the file list, and the aggregate this checker recomputes from that same list. A record that carries a file list but no matching aggregate, a command that is not a ``cargo vendor`` invocation, or a count that disagrees with the list is refused. """ problems: list[str] = [] status = vendor.get("status") if status != "generated_cargo_vendor": problems.append(f"the vendor record's status is {status!r}, not 'generated_cargo_vendor'") command = vendor.get("command") if not isinstance(command, (list, tuple)) or len(command) < 2: problems.append("the vendor record does not record the command that produced the tree") elif "vendor" not in [str(part) for part in command] or not str(command[0]).endswith("cargo"): problems.append(f"the recorded command {list(command)!r} is not a cargo vendor invocation") files = list(vendor.get("files") or []) if not files: problems.append("the vendor record lists no vendored files") else: if vendor.get("aggregate_sha256") != packaging.aggregate_digest(files): problems.append("the vendor record's aggregate does not match the file list it ships with") if vendor.get("file_count") != len(files): problems.append( f"the vendor record counts {vendor.get('file_count')} files but lists {len(files)}" ) return problems
def vendor_resolution_problems( static_inventory: list[dict], resolved_inventory: list[dict], vendor: dict) -> list[str]: """The vendor resolution the resolved inventory is allowed to claim.
An entry the static inventory marks ``conditional_on_vendor`` may appear as ``collected`` in the resolved copy only when the vendor record proves an observed ``cargo vendor`` execution and the entry names the whole vendored file index. An operator-supplied tree never resolves the entry, so a package that ships one cannot relabel the entry as collected. """ static_status = {entry.get("component"): entry.get("status") for entry in static_inventory} expected_paths = sorted(f"vendor/{entry['path']}" for entry in vendor.get("files") or []) status = vendor.get("status") problems: list[str] = [] for entry in resolved_inventory: if entry.get("status") != "collected": continue if static_status.get(entry.get("component")) != "conditional_on_vendor": continue if status != "generated_cargo_vendor": problems.append( f"the resolved inventory marks {entry.get('component')!r} collected from a vendor tree recorded " f"as {status!r}; only an observed cargo vendor run resolves it" ) continue problems.extend(vendor_execution_problems(vendor)) recorded_paths = sorted(entry.get("collected_paths") or []) if not recorded_paths: problems.append( f"the resolved inventory entry {entry.get('component')!r} claims status 'collected' without " "collected_paths" ) elif recorded_paths != expected_paths: problems.append( f"the resolved inventory entry {entry.get('component')!r} names {len(recorded_paths)} vendored " f"paths, not the whole vendored file index of {len(expected_paths)}" ) return problems
def unrecorded_vendored_paths(resolved_inventory: list[dict], source_manifest: dict) -> list[str]: """Every collected vendored path must be a member the source archive records.
The vendored record itself is verified member-by-member against the archive by ``check_archives_contents``, so a collected path outside that record is either a moved file or a rewritten inventory: both are refused here. """ included = {f"vendor/{entry['path']}" for entry in (source_manifest.get("vendor") or {}).get("files") or []} problems: list[str] = [] for entry in resolved_inventory: if entry.get("status") != "collected": continue for relative in entry.get("collected_paths") or []: if str(relative).startswith("vendor/") and relative not in included: problems.append( f"inventory entry {entry.get('component')!r} names {relative}, which the vendored record " "does not contain" ) return problems
# The statuses an inventory entry may carry. ``conditional_on_runtime_bundle`` and# ``not_distributed`` are declared by the shipped manifest; anything else is an# unrecognised label and is refused rather than passed through.SUPPORTED_INVENTORY_STATUSES = ( "collected", "pending_external", "conditional_on_vendor", "conditional_on_runtime_bundle", "not_distributed",)
# The static statuses a resolved entry may legitimately move away from, and how.VENDOR_RESOLVED_FROM = ("conditional_on_vendor",)RUNTIME_RESOLVED_FROM = ("pending_external", "conditional_on_runtime_bundle")
def runtime_resolution_recorded(entry: dict) -> bool: """Whether a resolved entry carries an explicit, reviewed runtime-bundle resolution.
``resolve_runtime_inventory`` flips a ``pending_external`` or ``conditional_on_runtime_bundle`` entry only when the bundled coverage is complete and reviewed, and it stamps that resolution onto the entry it writes. An entry that claims the same transition without those stamps did not come from a bundle. """ coverage = entry.get("runtime_source_coverage") or {} return ( bool(entry.get("bundled_by")) and coverage.get("complete") is True and coverage.get("reviewed") is True )
def inventory_membership_problems(static_inventory: list[dict], resolved_inventory: list[dict]) -> list[str]: """Reconcile component membership: every static entry, exactly once.
The resolved copy must carry every component the static inventory lists, and carry it once. A dropped component, a duplicate, or a component the static inventory never listed (unless the runtime bundle produced it) is refused, so deleting, renaming or repeating an entry cannot make an obligation disappear from the resolved inventory while the static one still declares it. """ problems: list[str] = [] static_components = [entry.get("component") for entry in static_inventory] known = set(static_components) counts: dict = {} for entry in resolved_inventory: component = entry.get("component") counts[component] = counts.get(component, 0) + 1 for component, count in counts.items(): if count > 1: problems.append(f"the resolved inventory lists {component!r} {count} times") for component in static_components: if counts.get(component, 0) == 0: problems.append(f"the resolved inventory drops the static entry {component!r}") for entry in resolved_inventory: component = entry.get("component") if component not in known and not runtime_resolution_recorded(entry): problems.append( f"the resolved inventory adds {component!r}, which the static inventory does not list " "and no runtime bundle produced" ) return problems
def inventory_transition_problems(static_inventory: list[dict], resolved_inventory: list[dict]) -> list[str]: """Refuse a resolved status that the static status cannot legitimately become.
Identity is always allowed. A ``conditional_on_vendor`` entry may become ``collected`` only through the vendor resolution (proved separately by ``vendor_resolution_problems``), and a ``pending_external`` or ``conditional_on_runtime_bundle`` entry only through an explicit runtime-bundle resolution. Any other change — an unsupported status, a resolution that weakens a collected entry, a flip with no recorded bundle behind it — is refused. """ static_status = {entry.get("component"): entry.get("status") for entry in static_inventory} problems: list[str] = [] for entry in resolved_inventory: component = entry.get("component") status = entry.get("status") if status not in SUPPORTED_INVENTORY_STATUSES: problems.append(f"inventory entry {component!r} records the unsupported status {status!r}") continue was = static_status.get(component) if was is None or was == status: continue if was in VENDOR_RESOLVED_FROM and status == "collected": continue if was in RUNTIME_RESOLVED_FROM and status == "collected": if not runtime_resolution_recorded(entry): problems.append( f"inventory entry {component!r} moves from {was!r} to 'collected' without a resolved " "runtime-bundle record" ) continue problems.append( f"inventory entry {component!r} moves from {was!r} to {status!r}, which the static status cannot become" ) return problems
def vendored_tree_accounting_problems( static_inventory: list[dict], resolved_inventory: list[dict], vendor: dict) -> list[str]: """A shipped vendored tree must be accounted for by some inventory entry.
Derived from the bytes rather than from the records. When the source archive carries vendored crate sources, some entry must either declare the vendor obligation (``conditional_on_vendor``, the pending state an operator-supplied tree keeps) or name at least one shipped member of that tree as ``collected``, which is how an operator's own hand-collected entry declares it. Deleting the entry from the inventory therefore fails even when a supplied tree otherwise looks complete.
How many vendored members a resolved entry must name is a separate, stricter rule: an entry the packager itself resolved out of ``conditional_on_vendor`` has to name the whole vendored file index (``vendor_resolution_problems``), and every named member must exist in the record (``unrecorded_vendored_paths``). """ included = {f"vendor/{entry['path']}" for entry in vendor.get("files") or []} if not included: return [] for inventory in (static_inventory, resolved_inventory): for entry in inventory: if entry.get("status") == "conditional_on_vendor": return [] if entry.get("status") == "collected" and included.intersection(entry.get("collected_paths") or []): return [] return ["the source archive ships a vendored crate tree that no inventory entry accounts for"]
def check_source_completeness(manifest: dict, source_manifest: dict, report: Report) -> None: """Independently derive the source-completeness state from the wired records.
The packager records a status; this check re-derives what the inventory and the vendor record allow and refuses a status that claims more. The static inventory in the release manifest and the resolved inventory in ``source-manifest.json`` are reconciled component by component — every static entry present exactly once, an allowed status transition, and a shipped vendored tree that some entry accounts for — so the one vendor resolution the packager performs, flipping ``conditional_on_vendor`` to ``collected``, is accepted only for an observed ``cargo vendor`` execution bound to the recorded file aggregate, and no other rewrite of the inventory passes. """ static_inventory = manifest["inventory"] resolved_inventory = source_manifest.get("inventory") or static_inventory vendor = source_manifest.get("vendor", {}) problems = vendor_resolution_problems(static_inventory, resolved_inventory, vendor) problems.extend(inventory_membership_problems(static_inventory, resolved_inventory)) problems.extend(inventory_transition_problems(static_inventory, resolved_inventory)) problems.extend(vendored_tree_accounting_problems(static_inventory, resolved_inventory, vendor)) problems.extend(unrecorded_vendored_paths(resolved_inventory, source_manifest)) pending = [ entry for entry in resolved_inventory if entry["status"] in ("pending_external", "conditional_on_vendor") ] if vendor.get("status") not in ("generated_cargo_vendor", "provided"): expected = "incomplete_missing_vendored_crate_sources" elif pending: expected = "incomplete_pending_third_party_notices" else: expected = manifest["source_completeness"].get("status") recorded = manifest["source_completeness"].get("status") if problems: report.add("source_completeness", NOT_VERIFIED, INTEGRITY, "; ".join(problems[:5])) return if recorded != expected and expected in ( "incomplete_missing_vendored_crate_sources", "incomplete_pending_third_party_notices", ): report.add( "source_completeness", NOT_VERIFIED, INTEGRITY, f"the manifest claims source completeness {recorded!r}, but the inventory and vendor state require " f"{expected!r}", ) elif recorded in ("incomplete_missing_vendored_crate_sources", "incomplete_pending_third_party_notices"): report.add( "source_completeness", PENDING, POLICY, f"{recorded}: {len(pending)} inventory entr(y|ies) still must be collected", ) else: report.add("source_completeness", VERIFIED, INTEGRITY, f"{recorded}")
def runtime_source_archive_members(source_manifest: dict, prefix: str) -> set[str]: """Full member names the runtime corresponding-source bundle stages into the source archive.""" entry = source_manifest.get("runtime_bundle") or {} if not entry.get("present"): return set() return {f"{prefix}/{item['path']}" for item in entry.get("source_members", [])}
def _rehash_archive_member(path: Path, member_name: str, limits: packaging.Limits) -> str: """Re-read one member and return its SHA-256, through the shared bounded reader.""" with packaging.open_bounded_tar(path, limits) as tar: try: member = tar.getmember(member_name) except KeyError as error: raise packaging.CheckFailure(f"{path.name} has no member {member_name}") from error handle = tar.extractfile(member) if handle is None: raise packaging.CheckFailure(f"{path.name} member {member_name} is not readable") digest = hashlib.sha256() while True: chunk = handle.read(1 << 20) if not chunk: break digest.update(chunk) return digest.hexdigest()
def runtime_inventory_requires_bundle(manifest: dict) -> list[str]: """Inventory components the manifest marks as produced by the runtime bundle.
Absence of the bundle is only a pending release requirement when the manifest actually declares runtime components that need it; a manifest with no such component (or a synthetic one) is not asked for a bundle it never referenced. """ required: list[str] = [] for entry in manifest.get("inventory", []): if entry.get("runtime_source_key") or entry.get("runtime_source_coverage"): required.append(entry.get("component") or entry.get("runtime_source_key")) elif entry.get("status") == "conditional_on_runtime_bundle": required.append(entry.get("component")) return [item for item in required if item]
def _runtime_member_table(archive_path: Path, limits: packaging.Limits) -> dict[str, dict]: """Re-read every regular member of a runtime archive: name -> {sha256, size}.
The raw hash and size come straight from the bytes, so a downstream check can compare an outer notice copy to the nested member it corresponds to without trusting any recorded value. The member count is finitely capped so an archive that declares an unbounded number of unknown extras is refused, not walked. """ table: dict[str, dict] = {} with packaging.open_bounded_tar(archive_path, limits) as tar: for member in tar: if not member.isreg(): continue if len(table) >= RUNTIME_MEMBER_CAP: raise packaging.CheckFailure( f"the runtime archive declares more than {RUNTIME_MEMBER_CAP} members" ) handle = tar.extractfile(member) if handle is None: raise packaging.CheckFailure(f"the runtime archive member {member.name} is not readable") digest = hashlib.sha256() while True: chunk = handle.read(1 << 20) if not chunk: break digest.update(chunk) table[member.name] = {"sha256": digest.hexdigest(), "size": member.size} return table
def _lockfile_crate_versions(text: str) -> dict: """Map crate name -> locked version list, parsed from a Cargo.lock's package blocks.""" versions: dict[str, list] = {} current = None for raw in text.splitlines(): line = raw.strip() if line.startswith("[[package]]"): current = None elif line.startswith("name = "): current = line.split("=", 1)[1].strip().strip('"') elif line.startswith("version = ") and current is not None: versions.setdefault(current, []).append(line.split("=", 1)[1].strip().strip('"')) current = None return versions
def _runtime_member_bytes(archive_path: Path, member_name: str, limits: packaging.Limits) -> bytes: """Read one bounded regular member's bytes from a runtime archive already audited.""" with packaging.open_bounded_tar(archive_path, limits) as tar: try: member = tar.getmember(member_name) except KeyError as error: raise packaging.CheckFailure(f"{archive_path.name} has no member {member_name}") from error handle = tar.extractfile(member) if handle is None: raise packaging.CheckFailure(f"{archive_path.name} member {member_name} is not readable") data = handle.read(limits.max_member_bytes + 1) if len(data) > limits.max_member_bytes: raise packaging.CheckFailure(f"{member_name} is larger than the member bound") return data
def _derive_runtime_coverage( record: dict, members: dict[str, dict], factory: list, bundle_module, lockfiles: dict | None = None) -> list[str]: """Derive runtime coverage from the nested member hashes, never from the flags.
The recorded ``reviewed``/``complete`` booleans are not evidence. For every component of the trusted factory mapping the checker requires the actual nested files: each source prefix, each required path and each notice member must be present in the re-read archive, the selection under a component's archive prefix must reproduce the recorded pinned tree digest, the recorded required mapping must not be weaker than the factory's, every source-bearing dependency must carry an explicit disposition, and every pinned commit must be preserved in the bundle provenance. Any shortfall is returned as a problem. """ problems: list[str] = [] coverage = record.get("coverage") or {} recorded = { spec.get("key"): spec for spec in (coverage.get("requirements") or []) if isinstance(spec, dict) } evaluated = { component.get("key"): component for component in (coverage.get("components") or []) if isinstance(component, dict) } provenance = record.get("provenance") or {} pins = {pin.get("commit") for pin in provenance.get("pins", []) if isinstance(pin, dict)} digests = {name: meta["sha256"] for name, meta in members.items()} for spec in factory: key = spec["key"] claimed = recorded.get(key) if claimed is None: problems.append(f"{key}: the bundle record states no requirement for this factory component") continue for field in ("required_source_prefixes", "required_paths", "required_notices", "required_collected_artifacts"): if set(claimed.get(field) or []) != set(spec.get(field) or []): problems.append(f"{key}: the recorded {field} does not match the trusted requirement") if (claimed.get("commit") or None) != (spec.get("commit") or None): problems.append(f"{key}: the recorded commit does not match the trusted pin") claimed_deps = { dep.get("name"): dep for dep in (claimed.get("dependencies") or []) if isinstance(dep, dict) } for dep in spec.get("dependencies", []): name = dep.get("name") recorded_dep = claimed_deps.get(name) if recorded_dep is None: problems.append(f"{key}: the recorded requirements drop dependency {name!r}") continue if (recorded_dep.get("locked_version") or None) != (dep.get("locked_version") or None): problems.append( f"{key}: the recorded dependency {name!r} locked_version does not match the trusted requirement" ) if (recorded_dep.get("disposition") or None) != (dep.get("disposition") or None): problems.append( f"{key}: the recorded dependency {name!r} disposition does not match the trusted requirement" ) prefix = spec.get("archive_prefix") or "" for required_prefix in spec.get("required_source_prefixes", []): if not any(name.startswith(required_prefix) for name in members): problems.append(f"{key}: the nested archive has no source member under {required_prefix}") for path in spec.get("required_paths", []): if bundle_module.member_for_required_path(prefix, path, digests) is None: problems.append(f"{key}: the nested archive is missing required path {path!r}") for notice in spec.get("required_notices", []): if notice not in members: problems.append(f"{key}: the nested archive is missing notice member {notice}") for dependency in spec.get("dependencies", []): name = dependency.get("name") if dependency.get("requires_source") and dependency.get("disposition") in (None, "unresolved"): problems.append(f"{key}: dependency {name!r} is unresolved") locked = dependency.get("locked_version") if locked: # The collector's declared locked dependency is acceptable evidence, but it # must exist exactly once and never contradict the archive's own lockfile. declared = [ dep for dep in ((evaluated.get(key) or {}).get("evidence") or {}).get("locked_dependencies", []) if isinstance(dep, dict) and dep.get("crate") == name ] if len(declared) != 1: problems.append( f"{key}: expected exactly one recorded locked dependency for {name!r}, found {len(declared)}" ) else: recorded_version = declared[0].get("locked_version") if recorded_version is None: recorded_version = declared[0].get("version") if str(recorded_version) != str(locked): problems.append( f"{key}: recorded locked dependency {name!r} version {recorded_version!r} " f"is not the required {locked!r}" ) text = (lockfiles or {}).get(key) if text is None: problems.append( f"{key}: no nested lockfile bytes were derivable to confirm {name!r} {locked}" ) else: found = _lockfile_crate_versions(text).get(name, []) if len(found) != 1: problems.append( f"{key}: the nested lockfile records {len(found)} {name!r} entries, expected exactly one" ) elif found[0] != str(locked): problems.append( f"{key}: the nested lockfile records {name!r} {found[0]!r}, not the required {locked!r}" ) if spec.get("commit") and spec["commit"] not in pins: problems.append( f"{key}: pinned commit {str(spec['commit'])[:12]}… is not preserved in the bundle provenance" ) if spec.get("source_input"): selection = bundle_module.component_selection({"member_hashes": digests}, prefix) evaluated_evidence = (evaluated.get(key) or {}).get("evidence") or {} pinned = evaluated_evidence.get("tree_digest") if not isinstance(pinned, str) or len(pinned) != 64: problems.append(f"{key}: the record carries no pinned tree digest for {prefix!r}") elif bundle_module.tree_digest(selection) != pinned: problems.append( f"{key}: the nested selection under {prefix!r} does not match the recorded pinned tree digest" ) floor = spec.get("min_members") if isinstance(floor, int) and floor: recorded_floor = evaluated_evidence.get("min_members") if not isinstance(recorded_floor, int) or recorded_floor < floor: problems.append( f"{key}: the recorded min_members {recorded_floor!r} is below the trusted floor {floor}" ) if len(selection) < floor: problems.append( f"{key}: the nested selection under {prefix!r} has {len(selection)} members, " f"below the trusted floor {floor}" ) for component in coverage.get("components") or []: if not isinstance(component, dict): continue if component.get("missing_source") or component.get("missing_notices"): problems.append(f"{component.get('key')}: the bundle record itself reports a coverage gap") if component.get("missing_artifacts"): problems.append( f"{component.get('key')}: collected artifact evidence is absent {component['missing_artifacts']}" ) return problems
def _required_notice_names(components_model: list, members: dict, bundle_module) -> list[str]: """The notice members the checker must bind: the factory set plus every audited notice.
Driving the required set from the trusted factory (and from the re-read nested archive) means a record that omits an entry cannot remove a required check. """ required: set[str] = set() for spec in components_model: required.update(spec.get("required_notices") or []) for name in members: if bundle_module.is_notice_member(name): required.add(name) return sorted(required)
def _verify_notice_binding( record: dict, members: dict[str, dict], entry: dict, source_archive_path: Path, binary_archive_path: Path, prefix: str, binary_prefix: str, required: list[str], bundle_module, limits: packaging.Limits,) -> tuple[list[str], list[str]]: """Bind every required notice to its nested bytes, its fetched artifact and both copies.
The record's own ``notice_mapping``/``notices`` lists are never trusted to define the required set: for each required notice the checker demands exactly one fetched binding whose hash and positive size match the nested member, the outer declared ``source_members``/``binary_notices`` must cover the canonical default staged paths, and the source and binary copies at those default paths must match the nested member byte for byte. Returns ``(byte_problems, binding_problems)``. """ byte_problems: list[str] = [] binding_problems: list[str] = [] mapping: dict[str, list] = {} for item in record.get("notice_mapping") or []: if isinstance(item, dict) and item.get("member"): mapping.setdefault(item["member"], []).append(item) declared_source = { item.get("path") for item in (entry.get("source_members") or []) if isinstance(item, dict) } binary_by_path = { item.get("path"): item for item in (entry.get("binary_notices") or []) if isinstance(item, dict) } if not required: binding_problems.append("the trusted mapping requires no runtime notice to bind") for name in required: nested = members.get(name) if nested is None: binding_problems.append(f"the nested archive is missing required notice {name}") continue entries = mapping.get(name, []) if len(entries) != 1: binding_problems.append( f"notice {name} needs exactly one fetched-artifact binding, found {len(entries)}" ) continue bound = entries[0] if bound.get("sha256") != nested["sha256"]: byte_problems.append(f"the binding for notice {name} does not match the nested notice hash") size = bound.get("size_bytes") if not isinstance(size, int) or size <= 0 or size != nested["size"]: byte_problems.append( f"the binding for notice {name} has no positive exact size matching the nested notice" ) default_source = bundle_module.staged_source_notice_path(name) default_binary = bundle_module.staged_binary_notice_path(name) if default_source not in declared_source: binding_problems.append( f"the source manifest does not declare the default staged source copy {default_source}" ) declared_binary = binary_by_path.get(default_binary) if declared_binary is None: binding_problems.append( f"the source manifest does not declare the default staged binary copy {default_binary}" ) elif declared_binary.get("source_member") != default_source: binding_problems.append( f"the binary notice {default_binary} does not map to the default source copy {default_source}" ) for label, outer_relative, archive_path, outer_prefix in ( ("source", default_source, source_archive_path, prefix), ("binary", default_binary, binary_archive_path, binary_prefix), ): try: outer = packaging.read_archive_bytes(archive_path, f"{outer_prefix}/{outer_relative}", limits) except (packaging.CheckFailure, KeyError) as error: byte_problems.append(f"the {label} notice copy {outer_relative} is missing: {error}") continue if hashlib.sha256(outer).hexdigest() != nested["sha256"] or len(outer) != nested["size"]: byte_problems.append( f"the {label} notice copy {outer_relative} does not match the nested notice " f"{name} (hash or size)" ) return byte_problems, binding_problems
def check_runtime_source_bundle( release_dir: Path, manifest: dict, source_manifest: dict, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS, factory: list | None = None,) -> None: """Independently re-verify the digest-bound third-party runtime source bundle.
Nothing the packager or the bundle record asserts is accepted on its word. The nested runtime archive is re-read out of the shipped source archive, re-hashed and re-audited, and its members are re-read as raw hash and size. Coverage is *derived* from those member hashes against the trusted factory mapping -- the recorded ``reviewed``/``complete`` flags are never evidence -- and every outer staged notice (in the source archive) and binary notice copy must exactly match the nested member it corresponds to, byte for byte. A record that claims complete coverage while a required member is gone fails integrity, and the bundle stays pending until the third-party legal review is certified. """ entry = source_manifest.get("runtime_bundle") or {} required = runtime_inventory_requires_bundle(manifest) if not entry.get("present"): if required: report.add( "runtime_source_bundle", PENDING, POLICY, "the inventory declares runtime components this bundle must produce, but no bundle is attached: " + ", ".join(required[:5]), ) else: report.add( "runtime_source_bundle", VERIFIED, POLICY, "the inventory declares no runtime component that requires a corresponding-source bundle", ) return
prefix = (source_manifest.get("archive") or {}).get("prefix") or manifest["source_archive"]["prefix"] source_name = (source_manifest.get("archive") or {}).get("name") or manifest["source_archive"]["name"] binary_prefix = manifest["archive"]["prefix"] binary_name = manifest["archive"]["name"] archive = entry.get("archive") or {} source_root = entry.get("source_root", "runtime-sources") source_archive_path = release_dir / source_name binary_archive_path = release_dir / binary_name
byte_problems: list[str] = [] coverage_problems: list[str] = []
archive_member = f"{prefix}/{source_root}/{archive.get('name')}" try: payload = packaging.read_archive_bytes(source_archive_path, archive_member, limits) except packaging.CheckFailure as error: report.add("runtime_source_bundle", NOT_VERIFIED, INTEGRITY, str(error)) return if hashlib.sha256(payload).hexdigest() != archive.get("sha256"): byte_problems.append("the runtime archive member does not match the recorded SHA-256") if len(payload) != archive.get("size_bytes"): byte_problems.append( f"the runtime archive member is {len(payload)} bytes, recorded {archive.get('size_bytes')}" )
bundle_module = packaging.runtime_bundle_module() model = bundle_module.REQUIRED_COVERAGE # ``factory`` is a caller (test) seam: production always uses the trusted model. components_model = factory if factory is not None else model.get("components", []) record: object = None members: dict[str, dict] = {} lockfiles: dict[str, str] = {} with tempfile.TemporaryDirectory() as tmp: extracted = Path(tmp) / (archive.get("name") or "runtime-corresponding-sources.tar.gz") extracted.write_bytes(payload) try: audit = bundle_module.audit_archive(extracted, limits=limits) except bundle_module.BundleError as error: report.add( "runtime_source_bundle", NOT_VERIFIED, INTEGRITY, f"the nested runtime archive does not re-audit: {error}", ) return for field in ("sha256", "size_bytes", "members", "source_members", "tree_digest", "internal_manifest_sha256"): if audit[field] != archive.get(field): byte_problems.append(f"the nested runtime archive {field} does not match the recorded value") try: members = _runtime_member_table(extracted, limits) except packaging.CheckFailure as error: byte_problems.append(str(error)) # Derive each component's lockfile bytes from the archive itself, so a locked # dependency version is confirmed against the actual nested lockfile. digests = {name: meta["sha256"] for name, meta in members.items()} for spec in components_model: for path in spec.get("required_paths", []): if not path.endswith("Cargo.lock"): continue resolved = bundle_module.member_for_required_path(spec.get("archive_prefix") or "", path, digests) if resolved: try: lockfiles[spec["key"]] = _runtime_member_bytes(extracted, resolved, limits).decode( "utf-8", "replace" ) except packaging.CheckFailure as error: byte_problems.append(str(error))
required_notices = _required_notice_names(components_model, members, bundle_module) record_member = f"{prefix}/{entry.get('bundle_record', f'{source_root}/{bundle_module.BUNDLE_RECORD_NAME}')}" try: record = json.loads(packaging.read_archive_bytes(source_archive_path, record_member, limits)) except (packaging.CheckFailure, json.JSONDecodeError) as error: record = None byte_problems.append(f"the bundle record is unreadable: {error}")
if isinstance(record, dict): if (record.get("archive") or {}).get("sha256") != archive.get("sha256"): byte_problems.append("the bundle record does not bind the recorded archive digest") if (record.get("source_completeness") or {}).get("release_ready"): byte_problems.append("the bundle record claims release_ready, which it must never do") record_coverage = record.get("coverage") or {} derived = _derive_runtime_coverage(record, members, components_model, bundle_module, lockfiles) notice_byte, notice_binding = _verify_notice_binding( record, members, entry, source_archive_path, binary_archive_path, prefix, binary_prefix, required_notices, bundle_module, limits, ) byte_problems.extend(notice_byte) gaps = derived + notice_binding if record_coverage.get("complete") and gaps: # A record that says complete while a required nested member or notice is gone # is lying about the bytes: a coherent re-attestation is not accepted. byte_problems.append( "the bundle record claims complete coverage, but the coverage derived from the nested " "archive and its notice bindings is incomplete: " + "; ".join(gaps[:4]) ) elif gaps: coverage_problems.extend(gaps) if not record_coverage.get("reviewed"): coverage_problems.append("the bundle record's coverage mapping is not marked reviewed") elif not gaps: legal_review = ((record.get("provenance") or {}).get("coverage_mapping") or {}).get("legal_review") if legal_review in (None, "pending"): coverage_problems.append( "the coverage derived from the nested archive is mechanically satisfied, but the " "third-party legal review is still pending (no legal certificate)" ) else: coverage_problems.append("the bundle record could not be read, so coverage could not be derived")
# The recorded outer members are still re-hashed against the archives. try: for item in entry.get("source_members", []): digest = _rehash_archive_member(source_archive_path, f"{prefix}/{item['path']}", limits) if digest != item["sha256"]: byte_problems.append(f"runtime member {item['path']} does not match its recorded hash") for item in entry.get("binary_notices", []): digest = _rehash_archive_member(binary_archive_path, f"{binary_prefix}/{item['path']}", limits) if digest != item["sha256"]: byte_problems.append(f"runtime notice {item['path']} does not match its recorded hash") except packaging.CheckFailure as error: byte_problems.append(str(error))
# The source manifest's own requirement list must not be weaker than the factory's. recorded_requirements = entry.get("requirements") if recorded_requirements is not None: factory_by_key = {spec["key"]: spec for spec in components_model} for spec in recorded_requirements: if not isinstance(spec, dict): continue trusted = factory_by_key.get(spec.get("key")) if trusted is None: coverage_problems.append( f"{spec.get('key')}: the source manifest declares an unknown runtime component" ) continue for field in ("required_source_prefixes", "required_notices", "required_collected_artifacts"): if set(spec.get(field) or []) != set(trusted.get(field) or []): coverage_problems.append( f"{spec.get('key')}: the source manifest {field} does not match the trusted requirement" )
if entry.get("release_ready"): coverage_problems.append("the runtime bundle records release_ready true") if entry.get("human_review"): coverage_problems.append("the runtime bundle records a human review it cannot have performed")
if byte_problems: report.add("runtime_source_bundle", NOT_VERIFIED, INTEGRITY, "; ".join(byte_problems[:8])) else: report.add( "runtime_source_bundle", VERIFIED, INTEGRITY, f"{archive.get('name')} re-read, re-hashed, re-audited and re-derived from {len(members)} nested members; " f"{len(entry.get('source_members', []))} staged members and " f"{len(entry.get('binary_notices', []))} binary notices matched to their nested hashes and sizes", ) if coverage_problems: report.add("runtime_source_bundle_coverage", PENDING, POLICY, "; ".join(coverage_problems[:8])) else: report.add( "runtime_source_bundle_coverage", VERIFIED, POLICY, "the coverage derived from the nested archive is satisfied and reviewed; " "human license approval is still outstanding", )
def check_git_state(manifest: dict, source_root: Path | None, release_dir: Path, report: Report) -> None: recorded = manifest["source"] if not source_root or not (source_root / ".git").exists(): report.add( "git_state", NOT_CHECKED, POLICY, "no Git worktree was supplied; the recorded revision and tag were not re-derived", ) return try: state = packaging.resolve_source_state( source_root, recorded.get("tag") or None, True, exclude_prefixes=[release_dir] ) except packaging.CheckFailure as error: report.add("git_state", NOT_VERIFIED, POLICY, str(error).splitlines()[0]) return problems = [] if state["revision"] != recorded["revision"]: problems.append(f"HEAD is {state['revision']}, the manifest records {recorded['revision']}") if state["dirty"] != recorded["dirty"]: problems.append(f"the worktree dirty state is {state['dirty']}, the manifest records {recorded['dirty']}") if recorded.get("tag"): if state["tag_object"] != recorded.get("tag_object"): problems.append("the tag object differs from the recorded one") if state["tag_commit"] != recorded.get("tag_commit"): problems.append("the tag commit differs from the recorded one") if problems: report.add("git_state", NOT_VERIFIED, POLICY, "; ".join(problems)) else: report.add( "git_state", VERIFIED, POLICY, f"revision {state['revision']} ({state['tree_state']}), dirty={state['dirty']}", )
def check_tag_policy(manifest: dict, source_root: Path | None, release_dir: Path, report: Report) -> None: recorded = manifest["source"] if manifest["development"]: report.add( "annotated_version_tag", PENDING, POLICY, f"development package labeled {manifest['development_label']!r} carries no release tag", ) return if not recorded.get("tag") or recorded.get("tag_type") != "annotated": report.add("annotated_version_tag", NOT_VERIFIED, POLICY, "no annotated version tag is recorded") return if recorded.get("tag_object") in (None, "") or recorded.get("tag_commit") != recorded.get("revision"): report.add( "annotated_version_tag", NOT_VERIFIED, POLICY, "the recorded tag object or tag commit does not match the recorded revision", ) return if not source_root or not (source_root / ".git").exists(): report.add( "annotated_version_tag", PENDING, POLICY, f"tag {recorded['tag']} is recorded but no Git worktree is available to re-verify it", ) return try: state = packaging.resolve_source_state( source_root, recorded["tag"], False, exclude_prefixes=[release_dir] ) except packaging.CheckFailure as error: report.add("annotated_version_tag", NOT_VERIFIED, POLICY, str(error).splitlines()[0]) return if state["tree_state"] != "clean_tagged": report.add("annotated_version_tag", NOT_VERIFIED, POLICY, f"tag state is {state['tree_state']}") return report.add( "annotated_version_tag", VERIFIED, POLICY, f"annotated tag {recorded['tag']} on commit {recorded['tag_commit']} (object {recorded['tag_object']})", )
def check_in_archive_manifest( release_dir: Path, manifest: dict, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS,) -> None: prefix = manifest["archive"]["prefix"] member = f"{prefix}/release-manifest.json" try: with packaging.open_bounded_tar(release_dir / manifest["archive"]["name"], limits) as tar: handle = tar.extractfile(member) if handle is None: raise KeyError(member) inner = json.loads(handle.read().decode("utf-8")) except (KeyError, tarfile.TarError, json.JSONDecodeError) as error: report.add( "in_archive_manifest", NOT_VERIFIED, INTEGRITY, f"the binary archive has no readable release-manifest.json: {error}", ) return outer = json.loads(json.dumps(manifest)) inner.pop("archive", None) outer.pop("archive", None)
def normalise_self_hash(document: dict) -> None: for artifact in document.get("artifacts", []): if artifact.get("kind") == "binary-archive": artifact.pop("sha256", None) artifact.pop("self_reference", None)
normalise_self_hash(inner) normalise_self_hash(outer) if inner != outer: report.add( "in_archive_manifest", NOT_VERIFIED, INTEGRITY, "the release manifest inside the archive differs from the release directory copy", ) return report.add( "in_archive_manifest", VERIFIED, INTEGRITY, "the archive-local release manifest matches the release directory copy", )
def check_clean_rebuild( release_dir: Path, manifest: dict, source_manifest: dict, report: Report, limits: packaging.Limits = packaging.DEFAULT_LIMITS,) -> None: """Independently verify the registered clean-rebuild record.
The recorded status alone is not evidence: the checker re-reads the portable evidence file registered inside the release directory, re-hashes it, and re-checks every binding against the shipped archive, the recomputed source aggregate, the shipped binaries and the expected environment. """ completeness = manifest["source_completeness"] if completeness.get("clean_rebuild") != "verified": report.add( "clean_extracted_source_rebuild", NOT_VERIFIED, POLICY, "no clean extracted-source rebuild is recorded; a release candidate needs one in the prepared toolchain image", ) return name = completeness.get("evidence_file") if not name or Path(name).is_absolute() or ".." in Path(name).parts: report.add( "clean_extracted_source_rebuild", NOT_VERIFIED, POLICY, f"the recorded evidence file name is missing or unsafe: {name!r}", ) return try: data = packaging.read_regular_bytes( release_dir / name, max_bytes=limits.max_evidence_bytes, description="registered rebuild evidence" ) except packaging.CheckFailure as error: report.add("clean_extracted_source_rebuild", NOT_VERIFIED, INTEGRITY, str(error)) return problems: list[str] = [] digest = hashlib.sha256(data).hexdigest() if digest != completeness.get("evidence_file_sha256"): problems.append("the registered evidence does not match the hash recorded for it") try: evidence = json.loads(data) except json.JSONDecodeError as error: report.add( "clean_extracted_source_rebuild", NOT_VERIFIED, INTEGRITY, f"the registered evidence is not JSON: {error}" ) return
shipped_archive = (manifest.get("source_archive") or {}).get("sha256") if not shipped_archive: problems.append("the release manifest records no shipped source archive hash") recomputed_aggregate = packaging.aggregate_digest(list(source_manifest.get("files") or [])) if not recomputed_aggregate: problems.append("the source manifest lists no files to recompute an aggregate from")
outputs = {item.get("name"): item for item in (evidence.get("outputs") or [])} prefix = manifest["archive"]["prefix"] measured: dict[str, dict] = {} try: for entry in manifest["artifacts"]: if entry["kind"] != "host-binary": continue payload = packaging.read_archive_bytes( release_dir / manifest["archive"]["name"], f"{prefix}/{entry['path']}" ) measured[entry["name"]] = {"sha256": hashlib.sha256(payload).hexdigest(), "size": len(payload)} except packaging.CheckFailure as error: report.add("clean_extracted_source_rebuild", NOT_VERIFIED, INTEGRITY, str(error)) return
problems += packaging.validate_rebuild_document( evidence, expected_archive={ "name": manifest["source_archive"]["name"], "sha256": manifest["source_archive"].get("sha256"), }, expected_aggregate=packaging.aggregate_digest(list(source_manifest.get("files") or [])), expected_binaries=measured, expected_environment=manifest["build"]["expected_environment"], )
if problems: report.add( "clean_extracted_source_rebuild", NOT_VERIFIED, POLICY, "the registered rebuild evidence does not verify: " + "; ".join(problems[:6]), ) return report.add( "clean_extracted_source_rebuild", VERIFIED, POLICY, f"{name} re-read, re-hashed ({digest[:16]}…) and re-bound to the source archive, aggregate, " "binaries and environment", )
def check_publication_claims(manifest: dict, report: Report) -> None: if manifest["development"]: report.add( "development_label", PENDING, POLICY, f"development package labeled {manifest['development_label']!r} can never be published", ) else: report.add("development_label", VERIFIED, POLICY, "not a development package") verification = manifest.get("verification", []) unproven = [item for item in verification if item.get("state") not in ("verified", "not_applicable")] if unproven: report.add( "packager_claims", PENDING, POLICY, "the packager recorded unproven checks: " + ", ".join(f"{item['check']}={item['state']}" for item in unproven[:8]), ) else: report.add("packager_claims", VERIFIED, POLICY, f"{len(verification)} packager checks all verified")
def summarise(report: Report, *, mode: str, release_dir: Path, manifest: dict | None, require_ready: bool) -> dict: """Turn the collected checks into the preflight result document.""" integrity_failures = report.integrity_failures() policy_pending = report.policy_pending() blockers = [f"{item['check']}: {item['detail']}" for item in integrity_failures] if mode == "release": blockers += [f"{item['check']}: {item['detail']}" for item in policy_pending] recorded_ready = bool(manifest and manifest.get("release_ready")) if require_ready and not recorded_ready: blockers.append("release_ready: the release manifest does not claim release readiness") verdict = "passed" if not blockers else "failed" return { "mode": mode, "release_dir": str(release_dir), "verdict": verdict, "ready": verdict == "passed" and recorded_ready, "require_ready": require_ready, "release_ready_as_recorded": recorded_ready, "counts": report.counted(), "checks": report.checks, "blockers": blockers, }
def validate_release( release_dir: Path | str, *, source_root: Path | str | None = None, require_ready: bool = True, mode: str = "release", report_path: Path | str | None = None,) -> dict: """Validate a release directory and return a structured verdict.
This is the function the publication path calls **before** it obtains any secret. It imports cleanly, performs no network access, opens every local file through a bounded descriptor that does not follow a symlink, and returns a document rather than raising, so a caller can gate on ``result["ready"]`` and inspect ``result["checks"]`` and ``result["blockers"]`` without re-encoding a failure.
``require_ready=True`` (the default) additionally requires the release manifest's recorded ``release_ready`` to be true, so a caller cannot ignore the packager's own blocking list. ``mode="release"`` treats every policy requirement as blocking; ``mode="development"`` reports them but only fails on integrity problems. """ if mode not in ("development", "release"): raise ValueError(f"mode must be development or release, got {mode!r}") release_dir = Path(release_dir) report = Report() if not release_dir.is_dir(): report.add("release_directory", NOT_VERIFIED, INTEGRITY, f"{release_dir} is not a directory") return summarise(report, mode=mode, release_dir=release_dir, manifest=None, require_ready=require_ready)
manifest = load_release_manifest(release_dir, report, packaging.DEFAULT_LIMITS) if manifest is None: return summarise(report, mode=mode, release_dir=release_dir, manifest=None, require_ready=require_ready)
resolved_source_root = Path(source_root) if source_root else None if resolved_source_root is not None and not (resolved_source_root / "Cargo.toml").exists(): resolved_source_root = None
def guarded(name: str, severity: str, function, *arguments) -> None: """Run one check; a corrupt input is a reported failure, never a crash.""" try: function(*arguments) except packaging.CheckFailure as error: report.add(name, NOT_VERIFIED, severity, str(error).splitlines()[0]) except Exception as error: # noqa: BLE001 - a damaged release directory must still produce a report report.add(name, NOT_VERIFIED, severity, f"{type(error).__name__}: {error}")
limits = packaging.Limits(**manifest.get("limits", {})) guarded("checksums", INTEGRITY, check_checksums, release_dir, manifest, report, limits) guarded("mandatory_artifacts", INTEGRITY, check_mandatory_artifacts, manifest, report) audits: dict[str, dict] = {} guarded("archive_safety", INTEGRITY, open_archives, release_dir, manifest, report, limits, audits) # Every dependent read is gated on the archive having passed its own audit, # so a failed audit is not followed by re-opening the same archive. binary_ok = "archive" in audits source_ok = "source_archive" in audits
def skipped(name: str, which: str) -> None: report.add(name, NOT_VERIFIED, INTEGRITY, f"skipped: the {which} did not pass its audit") source_manifest_path = release_dir / "source-manifest.json" source_manifest = {} if source_manifest_path.exists(): try: source_manifest = json.loads( packaging.read_regular_bytes( source_manifest_path, max_bytes=limits.max_manifest_bytes, description="source-manifest.json", ) ) report.add("source_manifest", VERIFIED, INTEGRITY, "source-manifest.json is present and readable") except (json.JSONDecodeError, packaging.CheckFailure) as error: report.add("source_manifest", NOT_VERIFIED, INTEGRITY, f"source-manifest.json is unusable: {error}") else: report.add( "source_manifest", NOT_VERIFIED, INTEGRITY, "source-manifest.json is missing; binaries must ship with their corresponding source", ) if source_manifest and source_ok: guarded( "source_archive_contents", INTEGRITY, check_archives_contents, release_dir, manifest, source_manifest, report, limits, binary_ok, ) elif source_manifest: skipped("source_archive_contents", "source archive") if source_manifest: guarded( "source_manifest_identity", POLICY, check_source_manifest_identity, release_dir, source_manifest, resolved_source_root, report, ) guarded("source_completeness", INTEGRITY, check_source_completeness, manifest, source_manifest, report) guarded( "runtime_source_bundle", INTEGRITY, check_runtime_source_bundle, release_dir, manifest, source_manifest, report, limits, ) if binary_ok: guarded("binary_archive_members", INTEGRITY, check_binary_archive_members, release_dir, manifest, report, limits) guarded("binary_remeasured", INTEGRITY, check_binaries_remeasured, release_dir, manifest, report, limits) guarded("in_archive_manifest", INTEGRITY, check_in_archive_manifest, release_dir, manifest, report, limits) else: skipped("binary_archive_members", "binary archive") skipped("binary_remeasured", "binary archive") skipped("in_archive_manifest", "binary archive") guarded("target_boundary", INTEGRITY, check_elf_target_policy, manifest, report) guarded("build_environment", POLICY, check_build_environment, manifest, report) guarded("reserved_preview_version", POLICY, check_version_policy, manifest, report) guarded("git_state", POLICY, check_git_state, manifest, resolved_source_root, release_dir, report) guarded("annotated_version_tag", POLICY, check_tag_policy, manifest, resolved_source_root, release_dir, report) if binary_ok and source_ok: guarded( "clean_extracted_source_rebuild", POLICY, check_clean_rebuild, release_dir, manifest, source_manifest, report, limits, ) else: skipped("clean_extracted_source_rebuild", "binary or source archive") guarded("publication_claims", POLICY, check_publication_claims, manifest, report)
result = summarise(report, mode=mode, release_dir=release_dir, manifest=manifest, require_ready=require_ready) if report_path: write_report(Path(report_path), result) return result
def write_report(path: Path, result: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def print_result(result: dict) -> None: print(f"release directory: {result['release_dir']}") print(f"mode: {result['mode']}") print(f"verdict: {result['verdict']}") for item in result["checks"]: marker = {"verified": "ok ", "pending": "todo", "not_verified": "FAIL", "not_checked": "skip"}[item["state"]] print(f" [{marker}] {item['check']} ({item['severity']}): {item['detail']}") integrity_failures = [item for item in result["checks"] if item["severity"] == INTEGRITY and item["state"] in (NOT_VERIFIED, "failed")] if integrity_failures: print("\nintegrity failures:") for item in integrity_failures: print(f" - {item['check']}: {item['detail']}") if result["blockers"]: print("\nunresolved release requirements:") for item in result["blockers"]: print(f" - {item}")
def run(argv: list[str]) -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--release-dir", type=Path, required=True) parser.add_argument("--source-root", type=Path, default=None, help="repository root for Git and manifest re-checks") parser.add_argument("--mode", choices=("development", "release"), default="development") parser.add_argument( "--require-ready", action="store_true", help="also require the release manifest's recorded release_ready flag (publication preflight uses this)", ) parser.add_argument("--report", type=Path, default=None, help="write the JSON report to this path") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) if not args.release_dir.is_dir(): print(f"error: {args.release_dir} is not a directory", file=sys.stderr) return EXIT_USAGE source_root = args.source_root or Path(__file__).resolve().parent.parent result = validate_release( args.release_dir, source_root=source_root, require_ready=args.require_ready or args.mode == "release", mode=args.mode, report_path=args.report, ) if args.json: print(json.dumps(result, indent=2, sort_keys=True)) else: print_result(result) return EXIT_OK if result["verdict"] == "passed" else EXIT_FAILED
def main(argv: list[str] | None = None) -> int: try: return run(list(sys.argv[1:] if argv is None else argv)) except packaging.UsageError as error: print(f"error: {error}", file=sys.stderr) return EXIT_USAGE except packaging.CheckFailure as error: print(f"check failed: {error}", file=sys.stderr) return EXIT_FAILED except Exception as error: # pragma: no cover - internal error path import traceback
traceback.print_exc() print(f"internal error: {error}", file=sys.stderr) return EXIT_INTERNAL
if __name__ == "__main__": raise SystemExit(main())