Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
40 kB · 804 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805"""Tests for the runtime corresponding-source hook in the release packager.
The packager gains two narrow, optional options:``--runtime-sources-report PATH`` and ``--runtime-sources-dir PATH``. When theyare absent nothing changes: the pending third-party inventory is retained.When they are present the verified bundle is read through the shared bounded,no-follow reader, its members ship inside the host source archive and itsnotices beside the binaries, and the pending inventory entries are resolved fromthe verified coverage -- without ever approving the coverage mapping on a human'sbehalf, and without ever flipping the static manifest recorded in the repository.
These tests build small synthetic runtime archives with the collector's own``deterministic_tar_gz`` so the format matches the real producer, and they reuse``tools/package_release.py``'s bounded reader and ``tools/runtime_source_bundle.py``so the hook is exercised through the same code paths as production."""
from __future__ import annotations
import contextlibimport ioimport jsonimport typesimport unittestfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"import sys # noqa: E402
sys.path.insert(0, str(TOOLS))
import check_release as cr # noqa: E402import collect_runtime_sources as crs # noqa: E402import package_release as pr # noqa: E402import runtime_source_bundle as rsb # noqa: E402
import tempfile # noqa: E402
def runtime_archive_bytes(members: dict) -> bytes: """A byte-reproducible runtime archive carrying its internal member manifest.""" payload = dict(members) manifest_text = "".join(f"{crs.sha256_hex(payload[name])} {name}\n" for name in sorted(payload)) payload[rsb.INTERNAL_MANIFEST] = manifest_text.encode("utf-8") return crs.deterministic_tar_gz(payload)
def factory_requirements() -> list: """The trusted per-component mapping the strict evaluation derives against.""" return json.loads(json.dumps(rsb.REQUIRED_COVERAGE["components"]))
LOCKFILE_VERSION = "0.1.160"
def lockfile_text(version: str = LOCKFILE_VERSION) -> bytes: return f'[[package]]\nname = "compiler_builtins"\nversion = "{version}"\n'.encode("utf-8")
def small_factory() -> list: """The trusted mapping with a tiny member floor, so a synthetic archive can satisfy it.
This is a caller test seam: production always uses ``rsb.REQUIRED_COVERAGE``. """ requirements = factory_requirements() for spec in requirements: if spec.get("min_members"): spec["min_members"] = 1 return requirements
def synth_lockfiles(version: str = LOCKFILE_VERSION) -> dict: return { spec["key"]: lockfile_text(version).decode("utf-8") for spec in factory_requirements() if any(path.endswith("Cargo.lock") for path in spec.get("required_paths", [])) }
def synth_bundle( root: Path, *, legal_review: str = "certified", drop_notice: bool = False, declared_dependency_version: str = LOCKFILE_VERSION, lockfile_version: str = LOCKFILE_VERSION, drop_binding: bool = False, duplicate_binding: bool = False,) -> dict: """A synthetic bundle shaped like load_runtime_bundle's return value.
The nested archive actually contains a file under every required source prefix and every required notice member, and the record preserves every pinned commit, so the strict derivation can pass. ``drop_notice`` removes one required notice member from the archive while the record still claims complete coverage, which is the "coherently re-attested partial" tamper the checker must reject. """ root.mkdir(parents=True, exist_ok=True) archive = root / rsb.ARCHIVE_NAME members: dict = {} notice_payloads: dict = {} for spec in factory_requirements(): for prefix in spec.get("required_source_prefixes", []): members.setdefault(prefix + "module.rs", b"// corresponding source\n") for path in spec.get("required_paths", []): member = f"{spec.get('archive_prefix') or ''}{path}" members.setdefault(member, lockfile_text(lockfile_version) if path.endswith("Cargo.lock") else b"// required path\n") for notice in spec.get("required_notices", []): members.setdefault(notice, b"license and notice text\n") notice_payloads.setdefault(notice, members[notice]) if drop_notice: dropped = sorted(notice_payloads)[0] members.pop(dropped) notice_payloads.pop(dropped) archive.write_bytes(runtime_archive_bytes(members)) audit = rsb.audit_archive(archive) requirements = factory_requirements() components = [] for spec in requirements: prefix = spec.get("archive_prefix") or "" evidence = {} if spec.get("source_input"): selection = { name: crs.sha256_hex(members[name]) for name in members if prefix and name.startswith(prefix) } evidence = { "source_input": spec.get("source_input"), "archive_prefix": prefix, "members": len(selection), "min_members": 1, "min_members_floor": spec.get("min_members", 0), "tree_digest": rsb.tree_digest(selection), "required_paths": spec.get("required_paths", []), "locked_dependencies": [ {"crate": dep.get("name"), "locked_version": declared_dependency_version} for dep in spec.get("dependencies", []) if dep.get("locked_version") ], } components.append( { "key": spec["key"], "scope": spec.get("scope"), "status": "satisfied", "missing_source": [], "missing_notices": [], "missing_artifacts": [], "artifact_evidence": {}, "dependencies": spec.get("dependencies", []), "evidence": evidence, } ) notice_mapping = [ { "member": notice, "output_name": notice[len("notices/"):] if notice.startswith("notices/") else notice, "sha256": crs.sha256_hex(notice_payloads[notice]), "size_bytes": len(notice_payloads[notice]), "url": "https://example.invalid/notice", "status": "collected", } for notice in sorted(notice_payloads) ] if drop_binding and notice_mapping: notice_mapping.pop(0) if duplicate_binding and notice_mapping: notice_mapping.append(dict(notice_mapping[0])) record = { "schema": 1, "record_kind": rsb.RECORD_KIND, "archive": {"sha256": audit["sha256"]}, "source_completeness": {"release_ready": False}, "coverage": { "reviewed": True, "complete": True, "problems": [], "requirements": requirements, "components": components, }, "provenance": { "collector_report": {"record_kind": rsb.REPORT_RECORD_KIND}, "coverage_mapping": {"legal_review": legal_review}, "pins": [ {"component": spec["key"], "repository": spec.get("repository"), "commit": spec.get("commit")} for spec in requirements if spec.get("commit") ], }, "notices": [ { "member": notice, "sha256": crs.sha256_hex(notice_payloads[notice]), "staged_source_path": rsb.staged_source_notice_path(notice), "staged_binary_path": rsb.staged_binary_notice_path(notice), } for notice in sorted(notice_payloads) ], "notice_mapping": notice_mapping, } record_bytes = json.dumps(record, indent=2, sort_keys=True).encode("utf-8") + b"\n" return { "rsb": rsb, "record": record, "record_bytes": record_bytes, "fragment": b"## runtime fragment\n", "audit": audit, "archive_path": archive, "notice_payloads": notice_payloads, "state": {"reviewed": True, "complete": True, "problems": [], "components": []}, "findings": [], "patch": {"replacement_entries": {}, "added_entries": [], "operations": []}, "coverage": {"maintainer": "test", "legal_review": legal_review}, "notice_mapping": notice_mapping, "report_name": "runtime-sources-report.json", "report_sha256": "c" * 64, }
def write_release_dir( root: Path, bundle: dict, entry: dict, *, source_prefix="SRC", binary_prefix="BIN", drop_binary_notice=False) -> tuple[Path, dict, dict]: """A minimal release directory carrying the staged runtime trees.""" root.mkdir(parents=True, exist_ok=True) source_members = list(pr.runtime_bundle_source_members(bundle, {"source_prefix": source_prefix})) pr.write_archive(root / "SRC.tar.gz", source_members) binary_members = [] dropped = sorted(bundle["notice_payloads"])[0] if drop_binary_notice and bundle["notice_payloads"] else None for member, payload in bundle["notice_payloads"].items(): if member == dropped: continue binary_members.append((f"{binary_prefix}/{rsb.staged_binary_notice_path(member)}", payload, pr.ARCHIVE_MODE_FILE)) # A binary archive needs at least one member even when there are no notices kept. if not binary_members: binary_members.append((f"{binary_prefix}/placeholder", b"", pr.ARCHIVE_MODE_FILE)) pr.write_archive(root / "BIN.tar.gz", binary_members) source_manifest = {"archive": {"name": "SRC.tar.gz", "prefix": source_prefix}, "runtime_bundle": entry} manifest = {"archive": {"name": "BIN.tar.gz", "prefix": binary_prefix}, "source_archive": {"name": "SRC.tar.gz", "prefix": source_prefix}} return root, source_manifest, manifest
class NoOptionsTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name)
def test_no_options_returns_none(self) -> None: args = types.SimpleNamespace(runtime_sources_report=None, runtime_sources_dir=None) context = {"manifest": {}, "limits": pr.DEFAULT_LIMITS, "created_at": "2026-01-01T00:00:00Z"} self.assertIsNone(pr.load_runtime_bundle(args, context))
def test_only_one_option_is_a_usage_error(self) -> None: args = types.SimpleNamespace(runtime_sources_report=Path("report.json"), runtime_sources_dir=None) context = {"manifest": {}, "limits": pr.DEFAULT_LIMITS, "created_at": "2026-01-01T00:00:00Z"} with self.assertRaises(pr.UsageError): pr.load_runtime_bundle(args, context)
def test_missing_directory_is_refused(self) -> None: args = types.SimpleNamespace( runtime_sources_report=self.root / "nope.json", runtime_sources_dir=self.root / "nope" ) context = {"manifest": {}, "limits": pr.DEFAULT_LIMITS, "created_at": "2026-01-01T00:00:00Z"} with self.assertRaises(pr.CheckFailure): pr.load_runtime_bundle(args, context)
class BindingFailureTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) self.collected = self.root / "collected" self.collected.mkdir() archive = runtime_archive_bytes({"pkg/library/lib.rs": b"x\n"}) (self.collected / rsb.ARCHIVE_NAME).write_bytes(archive)
def report(self, **overrides) -> Path: blob = (self.collected / rsb.ARCHIVE_NAME).read_bytes() report = { "schema": 1, "record_kind": rsb.REPORT_RECORD_KIND, "source_archive": { "name": rsb.ARCHIVE_NAME, "sha256": crs.sha256_hex(blob), "size_bytes": len(blob), "members": 2, "source_members": 1, "tree_digest": crs.tree_digest({"pkg/library/lib.rs": b"x\n"}), "notices_included": [], "written": True, }, } report["source_archive"].update(overrides) path = self.root / "runtime-sources-report.json" path.write_text(json.dumps(report)) return path
def test_tampered_report_digest_fails_closed(self) -> None: args = types.SimpleNamespace( runtime_sources_report=self.report(sha256="0" * 64), runtime_sources_dir=self.collected ) context = {"manifest": {}, "limits": pr.DEFAULT_LIMITS, "created_at": "2026-01-01T00:00:00Z"} with self.assertRaises(pr.CheckFailure): pr.load_runtime_bundle(args, context)
def test_wrong_record_kind_fails_closed(self) -> None: path = self.report() data = json.loads(path.read_text()) data["record_kind"] = "something_else" path.write_text(json.dumps(data)) args = types.SimpleNamespace(runtime_sources_report=path, runtime_sources_dir=self.collected) context = {"manifest": {}, "limits": pr.DEFAULT_LIMITS, "created_at": "2026-01-01T00:00:00Z"} with self.assertRaises(pr.CheckFailure): pr.load_runtime_bundle(args, context)
class ManifestEntryTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) self.bundle = synth_bundle(self.root / "source") self.entry = pr.runtime_bundle_manifest_entry(self.bundle)
def test_entry_is_portable_and_sentinel_free(self) -> None: blob = json.dumps(self.entry, sort_keys=True).encode("utf-8") self.assertEqual(list(pr.scan_bytes(blob)), []) self.assertNotIn(b"/home/", blob)
def test_entry_records_the_staged_members_and_notices(self) -> None: self.assertTrue(self.entry["present"]) self.assertFalse(self.entry["human_review"]) self.assertFalse(self.entry["release_ready"]) self.assertEqual(self.entry["archive"]["sha256"], self.bundle["audit"]["sha256"]) notice = sorted(self.bundle["notice_payloads"])[0] paths = {item["path"] for item in self.entry["source_members"]} self.assertIn(f"{rsb.STAGED_SOURCE_DIR}/{self.bundle['audit']['name']}", paths) self.assertIn(f"{rsb.STAGED_SOURCE_DIR}/{rsb.BUNDLE_RECORD_NAME}", paths) self.assertIn(f"{rsb.STAGED_SOURCE_DIR}/{rsb.FRAGMENT_NAME}", paths) self.assertIn(rsb.staged_source_notice_path(notice), paths) binary = {item["path"] for item in self.entry["binary_notices"]} self.assertIn(rsb.staged_binary_notice_path(notice), binary)
def test_entry_records_the_factory_mapping_and_nested_notice_hashes(self) -> None: keys = {spec["key"] for spec in self.entry["requirements"]} self.assertEqual(keys, {spec["key"] for spec in factory_requirements()}) nested = {item["member"]: item for item in self.entry["archive"]["notice_members"]} for notice, payload in self.bundle["notice_payloads"].items(): self.assertEqual(nested[notice]["sha256"], crs.sha256_hex(payload)) self.assertEqual(nested[notice]["size"], len(payload))
def test_runtime_inputs_record_a_digest_per_file(self) -> None: inputs = {item["name"]: item for item in self.entry["runtime_inputs"]} self.assertEqual(inputs[self.bundle["report_name"]]["sha256"], self.bundle["report_sha256"]) self.assertEqual(inputs[self.bundle["audit"]["name"]]["sha256"], self.bundle["audit"]["sha256"])
def test_source_members_require_the_source_prefix(self) -> None: members = pr.runtime_bundle_source_members(self.bundle, {"source_prefix": "P"}) names = [name for name, _, _ in members] self.assertTrue(all(name.startswith("P/runtime-sources/") for name in names))
class ResolveInventoryTests(unittest.TestCase): def bundle_with_patch(self, ready: bool) -> dict: entry = {"component": "pkg runtime", "status": "pending_external", "collected_paths": []} replacement = dict(entry) replacement["status"] = "collected" if ready else "pending_external" return { "patch": { "replacement_entries": {"pkg runtime": replacement}, "added_entries": ["llvm-mos compiler"], "operations": [], } }
def test_complete_coverage_flips_to_collected(self) -> None: manifest = {"inventory": [{"component": "pkg runtime", "status": "pending_external", "collected_paths": []}]} resolved = pr.resolve_runtime_inventory(manifest["inventory"], self.bundle_with_patch(ready=True)) self.assertEqual(resolved[0]["status"], "collected") # The static input is never mutated. self.assertEqual(manifest["inventory"][0]["status"], "pending_external")
def test_incomplete_coverage_never_flips(self) -> None: manifest = {"inventory": [{"component": "pkg runtime", "status": "pending_external", "collected_paths": []}]} resolved = pr.resolve_runtime_inventory(manifest["inventory"], self.bundle_with_patch(ready=False)) self.assertEqual(resolved[0]["status"], "pending_external")
def test_unmatched_component_is_added(self) -> None: manifest = {"inventory": [{"component": "pkg runtime", "status": "pending_external", "collected_paths": []}]} bundle = self.bundle_with_patch(ready=True) bundle["patch"]["replacement_entries"]["llvm-mos compiler"] = { "component": "llvm-mos compiler", "status": "collected", "collected_paths": [] } resolved = pr.resolve_runtime_inventory(manifest["inventory"], bundle) self.assertIn("llvm-mos compiler", {entry["component"] for entry in resolved})
class SourceSectionTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) self.entry = pr.runtime_bundle_manifest_entry(synth_bundle(self.root / "source"))
def test_absent_bundle_adds_no_section(self) -> None: self.assertEqual(pr.render_runtime_sources_section({"source_manifest": {"runtime_bundle": {"present": False}}}), [])
def test_present_bundle_section_is_digest_bound_and_honest(self) -> None: lines = "\n".join(pr.render_runtime_sources_section({"source_manifest": {"runtime_bundle": self.entry}})) self.assertIn(self.entry["archive"]["sha256"], lines) self.assertIn(self.entry["archive"]["tree_digest"], lines) self.assertIn("not a human license approval", lines)
class CheckerTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) # Build the bundle once so the entry and the staged bytes agree. self.bundle = synth_bundle(self.root / "source") self.entry = pr.runtime_bundle_manifest_entry(self.bundle) self.release_dir, self.source_manifest, self.manifest = write_release_dir( self.root / "release", self.bundle, self.entry ) self.report = cr.Report()
def run_check(self) -> dict: self.report = cr.Report() cr.check_runtime_source_bundle( self.release_dir, self.manifest, self.source_manifest, self.report, pr.DEFAULT_LIMITS, factory=small_factory() ) return {item["check"]: item for item in self.report.checks}
def test_bundle_reverifies_independently(self) -> None: checks = self.run_check() self.assertEqual(checks["runtime_source_bundle"]["state"], "verified") self.assertEqual(checks["runtime_source_bundle"]["severity"], "integrity") self.assertEqual(checks["runtime_source_bundle_coverage"]["state"], "verified")
def test_absent_bundle_without_a_runtime_component_is_not_a_failure(self) -> None: self.report = cr.Report() cr.check_runtime_source_bundle( self.release_dir, self.manifest, {"archive": self.source_manifest["archive"], "runtime_bundle": {"present": False}}, self.report, ) checks = {item["check"]: item for item in self.report.checks} self.assertEqual(checks["runtime_source_bundle"]["state"], "verified") self.assertEqual(checks["runtime_source_bundle"]["severity"], "policy")
def test_absent_bundle_with_a_declared_runtime_component_is_pending(self) -> None: manifest = dict(self.manifest) manifest["inventory"] = [{"component": "Rust std", "status": "pending_external", "runtime_source_key": "rust-std-1.93.0"}] self.report = cr.Report() cr.check_runtime_source_bundle( self.release_dir, manifest, {"archive": self.source_manifest["archive"], "runtime_bundle": {"present": False}}, self.report, ) checks = {item["check"]: item for item in self.report.checks} self.assertEqual(checks["runtime_source_bundle"]["state"], "pending") self.assertEqual(checks["runtime_source_bundle"]["severity"], "policy")
def test_requires_bundle_reads_the_declared_components(self) -> None: self.assertEqual(cr.runtime_inventory_requires_bundle({"inventory": [{"status": "collected"}]}), []) declared = cr.runtime_inventory_requires_bundle( {"inventory": [{"component": "Rust std", "status": "pending_external", "runtime_source_key": "rust-std-1.93.0"}]} ) self.assertEqual(declared, ["Rust std"])
def test_tampered_staged_member_is_refused(self) -> None: self.source_manifest["runtime_bundle"]["source_members"][0]["sha256"] = "0" * 64 checks = self.run_check() self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified")
def _scan(self, bundle, entry, name): release_dir, source_manifest, manifest = write_release_dir(self.root / name, bundle, entry) report = cr.Report() cr.check_runtime_source_bundle( release_dir, manifest, source_manifest, report, pr.DEFAULT_LIMITS, factory=small_factory() ) return {item["check"]: item for item in report.checks}
def test_a_coherently_changed_dependency_record_fails(self) -> None: # The archive and its staged members keep the same bytes; only the record's # declared dependency metadata changed. The checker must still reject it. bundle = synth_bundle(self.root / "changed", declared_dependency_version="0.1.161") entry = pr.runtime_bundle_manifest_entry(bundle) checks = self._scan(bundle, entry, "changed-release") self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified")
def test_required_notice_set_is_derived_from_the_factory(self) -> None: names = cr._required_notice_names(rsb.REQUIRED_COVERAGE["components"], {}, rsb) expected = { name for spec in rsb.REQUIRED_COVERAGE["components"] for name in spec.get("required_notices", []) } self.assertEqual(set(names), expected) self.assertEqual(len(expected), 8)
def test_omitting_the_fetch_binding_fails(self) -> None: # Copies and the nested notice are intact; only the fetched-artifact binding is gone. bundle = synth_bundle(self.root / "binding", drop_binding=True) entry = pr.runtime_bundle_manifest_entry(bundle) checks = self._scan(bundle, entry, "binding-release") self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified")
def test_duplicate_fetch_binding_fails(self) -> None: bundle = synth_bundle(self.root / "dup", duplicate_binding=True) entry = pr.runtime_bundle_manifest_entry(bundle) checks = self._scan(bundle, entry, "dup-release") self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified")
def test_missing_binary_notice_copy_fails(self) -> None: bundle = synth_bundle(self.root / "nocopy") entry = pr.runtime_bundle_manifest_entry(bundle) release_dir, source_manifest, manifest = write_release_dir( self.root / "nocopy-release", bundle, entry, drop_binary_notice=True ) report = cr.Report() cr.check_runtime_source_bundle( release_dir, manifest, source_manifest, report, pr.DEFAULT_LIMITS, factory=small_factory() ) checks = {item["check"]: item for item in report.checks} self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified")
def test_removing_metadata_entry_and_binary_copy_fails(self) -> None: # The inner notice stays valid; the mapping entry and the binary copy are both # gone. The factory still requires the notice, so the checker must fail. bundle = synth_bundle(self.root / "gone", drop_binding=True) entry = pr.runtime_bundle_manifest_entry(bundle) release_dir, source_manifest, manifest = write_release_dir( self.root / "gone-release", bundle, entry, drop_binary_notice=True ) report = cr.Report() cr.check_runtime_source_bundle( release_dir, manifest, source_manifest, report, pr.DEFAULT_LIMITS, factory=small_factory() ) checks = {item["check"]: item for item in report.checks} self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified")
def test_pending_legal_review_is_policy_pending(self) -> None: bundle = synth_bundle(self.root / "legal", legal_review="pending") entry = pr.runtime_bundle_manifest_entry(bundle) checks = self._scan(bundle, entry, "legal-release") self.assertEqual(checks["runtime_source_bundle"]["state"], "verified") self.assertEqual(checks["runtime_source_bundle_coverage"]["state"], "pending") self.assertIn("legal review", checks["runtime_source_bundle_coverage"]["detail"])
def test_removed_nested_notice_with_a_complete_flag_fails_integrity(self) -> None: # The archive, its internal manifest and the record are all coherent, but a # required notice member is gone while the record still claims complete # coverage: a coherent re-attestation must be rejected, not believed. bundle = synth_bundle(self.root / "partial", drop_notice=True) entry = pr.runtime_bundle_manifest_entry(bundle) checks = self._scan(bundle, entry, "partial-release") self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified") self.assertIn("derived", checks["runtime_source_bundle"]["detail"])
def test_binary_notice_copy_mismatch_fails_integrity(self) -> None: bundle = synth_bundle(self.root / "swap") entry = pr.runtime_bundle_manifest_entry(bundle) release_dir, source_manifest, manifest = write_release_dir(self.root / "swap-release", bundle, entry) (release_dir / "BIN.tar.gz").unlink() members = [] for index, (member, payload) in enumerate(sorted(bundle["notice_payloads"].items())): probe = b"tampered notice\n" if index == 0 else payload members.append((f"BIN/{rsb.staged_binary_notice_path(member)}", probe, pr.ARCHIVE_MODE_FILE)) pr.write_archive(release_dir / "BIN.tar.gz", members) report = cr.Report() cr.check_runtime_source_bundle(release_dir, manifest, source_manifest, report, pr.DEFAULT_LIMITS) checks = {item["check"]: item for item in report.checks} self.assertEqual(checks["runtime_source_bundle"]["state"], "not_verified")
def test_a_claimed_human_review_is_refused(self) -> None: self.source_manifest["runtime_bundle"]["human_review"] = True checks = self.run_check() self.assertEqual(checks["runtime_source_bundle_coverage"]["state"], "pending")
def test_runtime_archive_member_names_are_expectable(self) -> None: members = cr.runtime_source_archive_members(self.source_manifest, "SRC") self.assertIn("SRC/runtime-sources/" + self.entry["archive"]["name"], members) self.assertEqual(cr.runtime_source_archive_members({"runtime_bundle": {"present": False}}, "SRC"), set())
class StrictDerivationTests(unittest.TestCase): """The checker must derive coverage from the nested bytes, not the flags."""
def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) self.bundle = synth_bundle(self.root / "source") self.record = self.bundle["record"] self.members = cr._runtime_member_table(self.bundle["archive_path"], pr.DEFAULT_LIMITS)
def derive(self, record=None, members=None, lockfiles=None, factory=None): return cr._derive_runtime_coverage( record if record is not None else self.record, members if members is not None else self.members, factory if factory is not None else small_factory(), rsb, lockfiles if lockfiles is not None else synth_lockfiles(), )
def test_factory_mapping_is_fully_satisfied_by_the_nested_files(self) -> None: self.assertEqual(self.derive(), [])
def test_missing_required_notice_is_a_gap(self) -> None: members = dict(self.members) members.pop("notices/llvm-mos-sdk/LICENSE") self.assertTrue(any("notice member" in p for p in self.derive(members=members)))
def test_missing_required_source_prefix_is_a_gap(self) -> None: members = {n: m for n, m in self.members.items() if not n.startswith("llvm-mos-sdk/")} self.assertTrue(any("no source member" in p for p in self.derive(members=members)))
def test_missing_required_path_is_a_gap(self) -> None: members = {n: m for n, m in self.members.items() if n != "rust-mos/library/Cargo.lock"} self.assertTrue(any("missing required path" in p for p in self.derive(members=members)))
def test_a_changed_selection_is_a_gap(self) -> None: members = dict(self.members) members["rust-mos/library/extra.rs"] = {"sha256": "0" * 64, "size": 3} self.assertTrue(any("pinned tree digest" in p for p in self.derive(members=members)))
def test_a_weaker_recorded_requirement_is_a_gap(self) -> None: record = json.loads(json.dumps(self.record)) record["coverage"]["requirements"][0]["required_notices"] = [] self.assertTrue(any("does not match the trusted requirement" in p for p in self.derive(record=record)))
def test_a_missing_pin_is_a_gap(self) -> None: record = json.loads(json.dumps(self.record)) record["provenance"]["pins"] = [] self.assertTrue(any("pinned commit" in p for p in self.derive(record=record)))
def test_a_self_reported_artifact_gap_is_a_gap(self) -> None: record = json.loads(json.dumps(self.record)) record["coverage"]["components"][2]["missing_artifacts"] = ["mos-sysroot/c64/link.ld"] self.assertTrue(any("artifact evidence" in p for p in self.derive(record=record)))
def test_wrong_nested_lockfile_version_is_a_gap(self) -> None: self.assertTrue( any("nested lockfile records" in p for p in self.derive(lockfiles=synth_lockfiles("0.1.161"))) )
def test_conflicting_declared_dependency_version_is_a_gap(self) -> None: bundle = synth_bundle(self.root / "declared", declared_dependency_version="0.1.161") self.assertTrue(any("recorded locked dependency" in p for p in self.derive(record=bundle["record"])))
def test_absent_declared_locked_dependency_is_a_gap(self) -> None: record = json.loads(json.dumps(self.record)) for component in record["coverage"]["components"]: component["evidence"]["locked_dependencies"] = [] self.assertTrue(any("exactly one recorded locked dependency" in p for p in self.derive(record=record)))
def test_duplicate_declared_locked_dependency_is_a_gap(self) -> None: record = json.loads(json.dumps(self.record)) for component in record["coverage"]["components"]: deps = component.get("evidence", {}).get("locked_dependencies") if deps: component["evidence"]["locked_dependencies"] = deps + [dict(deps[0])] self.assertTrue(any("exactly one recorded locked dependency" in p for p in self.derive(record=record)))
def test_a_weaker_min_members_is_a_gap(self) -> None: factory = small_factory() for spec in factory: if spec.get("min_members"): spec["min_members"] = 5000 self.assertTrue(any("below the trusted floor" in p for p in self.derive(factory=factory)))
def test_changed_dependency_metadata_is_a_gap(self) -> None: record = json.loads(json.dumps(self.record)) record["coverage"]["requirements"][0]["dependencies"][0]["locked_version"] = "0.1.161" self.assertTrue( any("locked_version does not match the trusted requirement" in p for p in self.derive(record=record)) )
class TrustedCoverageTests(unittest.TestCase): def test_explicit_factory_mapping_is_accepted(self) -> None: # The built-in mapping names per-component requirements, so it passes. pr._require_explicit_runtime_coverage(rsb.REQUIRED_COVERAGE)
def test_generic_claim_is_refused(self) -> None: coverage = { "components": [ { "key": "x", "required_source_prefixes": [], "required_notices": [], "required_collected_artifacts": [], } ] } with self.assertRaises(pr.CheckFailure): pr._require_explicit_runtime_coverage(coverage)
class ShippedManifestTests(unittest.TestCase): def setUp(self) -> None: self.path = ROOT / "tools/release_source_manifest.json" if not self.path.exists(): self.skipTest("the release source manifest is not present") self.manifest = json.loads(self.path.read_text())
def test_runtime_sources_policy_block_is_present(self) -> None: block = self.manifest["runtime_sources"] self.assertEqual(block["bundle_record_kind"], rsb.RECORD_KIND) self.assertEqual(block["collector_report_record_kind"], rsb.REPORT_RECORD_KIND) self.assertIn("conditional_on_runtime_bundle", block["inventory_status"]["supported_helper_statuses"]) self.assertEqual(block["narrow_options"]["--runtime-sources-report"].split()[0], "path")
def test_bundler_is_allowlisted(self) -> None: include = {entry["path"] for entry in self.manifest["source_paths"]["include"]} self.assertIn("tools/runtime_source_bundle.py", include)
def test_static_inventory_still_gates_on_the_runtime_entries(self) -> None: # The hook resolves these only in the resolved manifest; the repository copy # must keep them pending so source completeness is never claimed here. pending = [entry for entry in self.manifest["inventory"] if entry["status"] == "pending_external"] self.assertTrue(pending)
class RealArtifactTests(unittest.TestCase): """Exercise the hook against the real collected bundle when it is present."""
def setUp(self) -> None: self.report_path = ROOT / "target/release-runtime-sources/runtime-sources-report.json" self.collected = ROOT / "target/release-runtime-sources/collected" if not self.report_path.exists() or not self.collected.is_dir(): self.skipTest("the collected runtime bundle is not present") self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name)
def load(self) -> dict: args = types.SimpleNamespace(runtime_sources_report=self.report_path, runtime_sources_dir=self.collected) manifest = json.loads((ROOT / "tools/release_source_manifest.json").read_text()) context = {"manifest": manifest, "limits": pr.DEFAULT_LIMITS, "created_at": "2026-01-01T00:00:00Z"} return manifest, pr.load_runtime_bundle(args, context)
def test_real_bundle_is_source_complete_but_not_release_ready(self) -> None: _, bundle = self.load() entry = pr.runtime_bundle_manifest_entry(bundle) self.assertTrue(entry["source_complete"]) self.assertFalse(entry["human_review"]) self.assertFalse(entry["release_ready"]) # 11 staged source members plus 8 notices beside the binaries. self.assertEqual(len(entry["source_members"]) + len(entry["binary_notices"]), 19) self.assertTrue(entry["archive"]["sha256"].startswith("d37aa9ce"))
def test_real_bundle_resolves_the_pending_inventory(self) -> None: manifest, bundle = self.load() resolved = pr.resolve_runtime_inventory(manifest["inventory"], bundle) runtime = [entry for entry in resolved if entry.get("runtime_source_coverage")] self.assertTrue(runtime) self.assertTrue(all(entry["status"] == "collected" for entry in runtime))
def test_real_bundle_satisfies_the_strict_derivation(self) -> None: # The production factory (member floors included) must derive cleanly from the # real archive's own lockfiles, not from any recorded flag. _, bundle = self.load() archive = bundle["archive_path"] members = cr._runtime_member_table(archive, pr.DEFAULT_LIMITS) digests = {name: meta["sha256"] for name, meta in members.items()} lockfiles = {} for spec in rsb.REQUIRED_COVERAGE["components"]: for path in spec.get("required_paths", []): if path.endswith("Cargo.lock"): resolved = rsb.member_for_required_path(spec.get("archive_prefix") or "", path, digests) if resolved: lockfiles[spec["key"]] = cr._runtime_member_bytes( archive, resolved, pr.DEFAULT_LIMITS ).decode("utf-8", "replace") problems = cr._derive_runtime_coverage( bundle["record"], members, rsb.REQUIRED_COVERAGE["components"], rsb, lockfiles ) self.assertEqual(problems, [])
def test_real_bundle_passes_the_full_checker(self) -> None: _, bundle = self.load() entry = pr.runtime_bundle_manifest_entry(bundle) release_dir, source_manifest, manifest = write_release_dir(self.root / "real-release", bundle, entry) report = cr.Report() cr.check_runtime_source_bundle(release_dir, manifest, source_manifest, report, pr.DEFAULT_LIMITS) checks = {item["check"]: item for item in report.checks} self.assertEqual( checks["runtime_source_bundle"]["state"], "verified", checks["runtime_source_bundle"]["detail"] ) self.assertEqual(checks["runtime_source_bundle"]["severity"], "integrity") # The derivation is satisfied, but the third-party legal review is pending. self.assertEqual(checks["runtime_source_bundle_coverage"]["state"], "pending")
if __name__ == "__main__": unittest.main()