Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
108 kB · 2289 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290#!/usr/bin/env python3"""Tests for the bounded local release verification harness.
These tests are pure standard-library ``unittest``. They exercise the parts of``tools/verify_release_local.py`` that can be decided without Docker:
- archive extraction and its fail-closed safety rules,- the isolation guard that keeps a verification container away from the checkout,- the release-directory checksum chain,- the ``release_rebuild_evidence`` record, proven consumable by calling the packager's own validator rather than by asserting the shape from memory,- the fail-on-invocation shim, run for real under ``/bin/sh``, because an empty attempt journal only means something if the shim actually records,- owned-process cleanup, with a real child process,- bounded readiness structure, so a blocking read cannot creep back in.
What a passing run does not prove: that the pinned Ubuntu image rebuilds thesebytes, that the runtime container works, or that any C64 program executed.Those need the two container gates, which are gated behind the environmentvariable ``ULTIMATE_TOOL_VERIFY_LOCAL_DOCKER=1`` and skipped by default."""
from __future__ import annotations
import atexitimport contextlibimport hashlibimport http.serverimport ioimport jsonimport osimport shutilimport statimport subprocessimport sysimport tarfileimport tempfileimport threadingimport unittestfrom pathlib import Pathfrom unittest import mock
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"if str(TOOLS) not in sys.path: sys.path.insert(0, str(TOOLS))
import package_release as packaging # noqa: E402import verify_release_local as verify # noqa: E402
PACKAGER_FAILURE = getattr(packaging, "CheckFailure", Exception)DOCKER_GATE = os.environ.get("ULTIMATE_TOOL_VERIFY_LOCAL_DOCKER") == "1"SOURCE_PREFIX = "ultimate-tool-0.2.0-preview.1-linux-x86_64-gnu-source"BINARY_PREFIX = "ultimate-tool-0.2.0-preview.1-linux-x86_64-gnu"
def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest()
class DockerInvocationForbidden(AssertionError): """A mocked test asked the real Docker CLI for something and was refused."""
DOCKER_GUARD_NOTE = ( "this test stubs the Docker seams, so no docker subprocess may start; the guard " "refused the invocation instead of asking a daemon")
def looks_like_docker(argv) -> bool: """Whether an argument vector names the Docker CLI.""" if isinstance(argv, bytes): argv = argv.decode("utf-8", "replace") if isinstance(argv, str): argv = argv.split() try: executable = Path(str(argv[0])) except (IndexError, KeyError, TypeError): return False return executable.name.startswith("docker")
@contextlib.contextmanagerdef forbid_docker_subprocesses(): """Refuse every Docker entry point for the life of one test.
The harness resolves the CLI through ``verify.docker_executable`` and then spawns it, so both seams refuse: a mocked test either stubs the seam it needs or fails loudly. Nothing is installed on ``PATH`` and no variable is exported, so the refusal covers one test only and a test elsewhere in the suite can still reach the real daemon seams. """ real_run, real_popen = subprocess.run, subprocess.Popen
def refuse(what: str) -> None: raise DockerInvocationForbidden(f"{DOCKER_GUARD_NOTE}: {what}")
def guarded_run(*args, **kwargs): argv = args[0] if args else kwargs.get("args") if looks_like_docker(argv): refuse(f"subprocess.run({argv!r})") return real_run(*args, **kwargs)
def guarded_popen(*args, **kwargs): argv = args[0] if args else kwargs.get("args") if looks_like_docker(argv): refuse(f"subprocess.Popen({argv!r})") return real_popen(*args, **kwargs)
def guarded_executable() -> str: refuse("docker_executable() was asked for the CLI") return "docker" # pragma: no cover - refuse() always raises
with mock.patch.object(verify, "docker_executable", side_effect=guarded_executable), \ mock.patch.object(verify.subprocess, "run", side_effect=guarded_run), \ mock.patch.object(verify.subprocess, "Popen", side_effect=guarded_popen): yield
def make_archive(path: Path, members: list[tuple[str, bytes, int]], links: list[tuple[str, bytes, int, str]] | None = None) -> Path: """Write a deterministic tar.gz from (name, content, type) entries.""" with tarfile.open(path, "w:gz") as handle: for name, content, kind in members: info = tarfile.TarInfo(name) info.type = kind if kind == tarfile.DIRTYPE: info.size = 0 handle.addfile(info, None) continue info.size = len(content) info.mode = 0o644 handle.addfile(info, io.BytesIO(content)) for name, linkname, kind, target in links or []: info = tarfile.TarInfo(name) info.type = kind info.linkname = target info.size = 0 handle.addfile(info, None) return path
class TemporaryDirectoryCase(unittest.TestCase): def setUp(self) -> None: self._root = Path(tempfile.mkdtemp(prefix="ut-test-release-local-")) self.addCleanup(shutil.rmtree, self._root, ignore_errors=True)
@property def root(self) -> Path: return self._root
def write(self, relative: str, content: bytes | str) -> Path: path = self._root / relative path.parent.mkdir(parents=True, exist_ok=True) if isinstance(content, str): content = content.encode("utf-8") path.write_bytes(content) return path
# ---------------------------------------------------------------------------# Archive extraction# ---------------------------------------------------------------------------
class ArchiveExtractionTests(TemporaryDirectoryCase): def build(self, members: list[tuple[str, bytes, int]], links=None) -> tuple[Path, dict]: archive = make_archive(self.root / "archive.tar.gz", members, links) expected = { name: {"sha256": sha256(content), "size": len(content)} for name, content, kind in members if kind == tarfile.REGTYPE } return archive, expected
def test_extracts_a_verified_archive_and_reports_its_prefix(self) -> None: archive, expected = self.build( [ (f"{SOURCE_PREFIX}/Cargo.toml", b"[workspace]\n", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/crates/a/src/lib.rs", b"fn a() {}\n", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/bin/tool", b"#!/bin/sh\n", tarfile.REGTYPE), ] ) (self.root / "out").mkdir() result = verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertEqual(result["prefix"], SOURCE_PREFIX) self.assertEqual(result["members"], 3) self.assertEqual(result["archive_sha256"], sha256(archive.read_bytes())) self.assertTrue((self.root / "out" / SOURCE_PREFIX / "Cargo.toml").is_file())
def test_marks_an_executable_member_executable(self) -> None: archive, expected = self.build([(f"{SOURCE_PREFIX}/bin/tool", b"#!/bin/sh\n", tarfile.REGTYPE)]) (self.root / "out").mkdir() verify.extract_verified_archive(archive, self.root / "out", expected=expected) mode = stat.S_IMODE((self.root / "out" / SOURCE_PREFIX / "bin" / "tool").stat().st_mode) self.assertEqual(mode, 0o644) # A member carrying the executable bit keeps it; the packaged binaries # rely on this, so assert it with an explicit archive. with tarfile.open(self.root / "exec.tar.gz", "w:gz") as handle: info = tarfile.TarInfo(f"{SOURCE_PREFIX}/bin/ultimate-tool") info.size = 4 info.mode = 0o755 handle.addfile(info, io.BytesIO(b"tool")) (self.root / "out2").mkdir() verify.extract_verified_archive(self.root / "exec.tar.gz", self.root / "out2") mode = stat.S_IMODE((self.root / "out2" / SOURCE_PREFIX / "bin" / "ultimate-tool").stat().st_mode) self.assertEqual(mode, 0o755)
def test_refuses_absolute_path_member(self) -> None: archive, expected = self.build([("/etc/passwd", b"nope", tarfile.REGTYPE)]) (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertIn("unsafe member", str(caught.exception))
def test_refuses_traversal_member(self) -> None: archive, expected = self.build([(f"{SOURCE_PREFIX}/../../escape.txt", b"nope", tarfile.REGTYPE)]) (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertIn("unsafe member", str(caught.exception)) self.assertFalse((self.root / "escape.txt").exists())
def test_refuses_a_symlink_member(self) -> None: archive = make_archive( self.root / "archive.tar.gz", [], links=[(f"{SOURCE_PREFIX}/link", b"", tarfile.SYMTYPE, "/etc/passwd")], ) (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out") self.assertIn("non-regular member", str(caught.exception))
def test_refuses_a_hardlink_member(self) -> None: archive = make_archive( self.root / "archive.tar.gz", [], links=[(f"{SOURCE_PREFIX}/link", b"", tarfile.LNKTYPE, f"{SOURCE_PREFIX}/Cargo.toml")], ) (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out") self.assertIn("non-regular member", str(caught.exception))
def test_refuses_a_device_member(self) -> None: archive = make_archive( self.root / "archive.tar.gz", [(f"{SOURCE_PREFIX}/dev", b"", tarfile.CHRTYPE)], ) (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out") self.assertIn("non-regular member", str(caught.exception))
def test_refuses_a_member_whose_content_changed(self) -> None: # Same declared size, different bytes: only the digest can catch this. archive, expected = self.build([(f"{SOURCE_PREFIX}/Cargo.toml", b"real\n", tarfile.REGTYPE)]) expected[f"{SOURCE_PREFIX}/Cargo.toml"] = {"sha256": sha256(b"fake\n"), "size": 5} (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertIn("does not match the expected SHA-256", str(caught.exception))
def test_refuses_a_member_whose_declared_size_changed(self) -> None: archive, expected = self.build([(f"{SOURCE_PREFIX}/Cargo.toml", b"real\n", tarfile.REGTYPE)]) expected[f"{SOURCE_PREFIX}/Cargo.toml"] = {"sha256": sha256(b"real\n"), "size": 99} (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertIn("expected 99", str(caught.exception))
def test_refuses_a_missing_member(self) -> None: archive, expected = self.build([(f"{SOURCE_PREFIX}/Cargo.toml", b"x\n", tarfile.REGTYPE)]) expected[f"{SOURCE_PREFIX}/absent.rs"] = {"sha256": sha256(b"x\n"), "size": 2} (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertIn("is missing", str(caught.exception))
def test_records_an_extra_member_rather_than_refusing_it(self) -> None: # A prepared source archive carries generated members the manifest does # not list, so an extra member is recorded, not treated as a failure. archive, expected = self.build( [ (f"{SOURCE_PREFIX}/Cargo.toml", b"x\n", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", b"{}\n", tarfile.REGTYPE), ] ) del expected[f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json"] (self.root / "out").mkdir() result = verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertEqual(result["extras"], [f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json"])
def test_requires_a_generated_member(self) -> None: archive, expected = self.build([(f"{SOURCE_PREFIX}/Cargo.toml", b"x\n", tarfile.REGTYPE)]) (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive( archive, self.root / "out", expected=expected, required=(f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json",), ) self.assertIn("missing generated member", str(caught.exception))
def test_accepts_the_generated_members_of_a_prepared_archive(self) -> None: archive, expected = self.build( [ (f"{SOURCE_PREFIX}/Cargo.toml", b"[workspace]\n", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", b"{}\n", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/vendor-sources.toml", b"[source.crates-io]\n", tarfile.REGTYPE), ] ) del expected[f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json"] del expected[f"{SOURCE_PREFIX}/vendor-sources.toml"] (self.root / "out").mkdir() result = verify.extract_verified_archive( archive, self.root / "out", expected=expected, required=(f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", f"{SOURCE_PREFIX}/vendor-sources.toml"), ) self.assertEqual(result["extras"], []) self.assertEqual(result["members"], 3)
def test_the_archive_is_audited_with_the_packagers_own_rules(self) -> None: archive, expected = self.build([(f"{SOURCE_PREFIX}/Cargo.toml", b"x\n", tarfile.REGTYPE)]) (self.root / "out").mkdir() result = verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertIn("package_release", result["audit"]["audit"]) self.assertEqual(result["audit"]["tar_safety"], "verified")
def test_the_member_name_policy_matches_the_packagers(self) -> None: for name in ("/etc/passwd", "../escape", "a/../b", "a\\b", "", "a//b", "a/./b", "ok.rs"): self.assertEqual(verify.unsafe_member_reason(name), packaging.unsafe_member_name(name), name)
def test_refuses_a_non_empty_destination(self) -> None: archive, expected = self.build([(f"{SOURCE_PREFIX}/Cargo.toml", b"x\n", tarfile.REGTYPE)]) destination = self.root / "out" destination.mkdir() (destination / "already-here").write_text("keep\n") with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, destination, expected=expected) self.assertIn("is not empty", str(caught.exception))
def test_refuses_an_archive_above_the_member_bound(self) -> None: archive, _ = self.build( [ (f"{SOURCE_PREFIX}/a", b"a", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/b", b"b", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/c", b"c", tarfile.REGTYPE), ] ) (self.root / "out").mkdir() with mock.patch.object(verify, "MAX_MEMBERS", 2): with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out") self.assertIn("more than 2 members", str(caught.exception))
def test_refuses_a_duplicate_member(self) -> None: archive, _ = self.build( [ (f"{SOURCE_PREFIX}/Cargo.toml", b"one\n", tarfile.REGTYPE), (f"{SOURCE_PREFIX}/Cargo.toml", b"two\n", tarfile.REGTYPE), ] ) (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out") self.assertIn("duplicate member", str(caught.exception))
def test_refuses_an_archive_whose_decompressed_bytes_exceed_the_bound(self) -> None: # A highly compressible member keeps the archive file itself small, so # this exercises the declared/actual byte bound rather than the file # size bound. A header that lies about its size cannot get past the # in-stream accounting, which counts bytes actually decompressed. archive, _ = self.build([(f"{SOURCE_PREFIX}/big.bin", b"\0" * 4096, tarfile.REGTYPE)]) self.assertLess(archive.stat().st_size, 2000) (self.root / "out").mkdir() with mock.patch.object(verify, "MAX_ARCHIVE_BYTES", 200): with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out") self.assertIn("byte bound", str(caught.exception))
def test_reports_the_bytes_actually_decompressed(self) -> None: archive, expected = self.build( [ (f"{SOURCE_PREFIX}/a", b"a" * 100, tarfile.REGTYPE), (f"{SOURCE_PREFIX}/b", b"b" * 50, tarfile.REGTYPE), ] ) (self.root / "out").mkdir() result = verify.extract_verified_archive(archive, self.root / "out", expected=expected) self.assertEqual(result["bytes"], 150) self.assertEqual(result["actual_bytes"], 150)
# ---------------------------------------------------------------------------# Isolation guard# ---------------------------------------------------------------------------
class IsolationGuardTests(TemporaryDirectoryCase): def test_accepts_one_mount_of_the_scratch_directory(self) -> None: scratch = self.root / "scratch" / SOURCE_PREFIX scratch.mkdir(parents=True) problems = verify.isolation_problems( checkout_root=self.root / "checkout", mounts=[{"source": str(scratch), "target": "/work"}], forbidden_paths=[self.root / "checkout"], expected_source=scratch, ) self.assertEqual(problems, [])
def test_refuses_a_second_mount(self) -> None: scratch = self.root / "scratch" / SOURCE_PREFIX scratch.mkdir(parents=True) problems = verify.isolation_problems( checkout_root=self.root / "checkout", mounts=[ {"source": str(scratch), "target": "/work"}, {"source": str(self.root), "target": "/extra"}, ], forbidden_paths=[], expected_source=scratch, ) self.assertIn("expected exactly one bind mount, found 2", problems)
def test_refuses_mounting_the_checkout(self) -> None: checkout = self.root / "checkout" checkout.mkdir() problems = verify.isolation_problems( checkout_root=checkout, mounts=[{"source": str(checkout), "target": "/work"}], forbidden_paths=[checkout], expected_source=checkout, ) self.assertTrue(any("exposes the forbidden path" in item for item in problems), problems)
def test_refuses_a_scratch_directory_inside_the_checkout(self) -> None: checkout = self.root / "checkout" scratch = checkout / "scratch" / SOURCE_PREFIX scratch.mkdir(parents=True) problems = verify.isolation_problems( checkout_root=checkout, mounts=[{"source": str(scratch), "target": "/work"}], forbidden_paths=[checkout], expected_source=scratch, ) self.assertTrue(any("lives under the forbidden path" in item for item in problems), problems)
def test_refuses_the_docker_socket(self) -> None: socket = self.root / "docker.sock" socket.write_bytes(b"") problems = verify.isolation_problems( checkout_root=self.root / "checkout", mounts=[{"source": str(socket), "target": "/var/run/docker.sock"}], forbidden_paths=[], expected_source=socket, ) self.assertTrue(any("Docker socket" in item for item in problems), problems)
def test_default_forbidden_covers_the_checkout_parent_and_socket(self) -> None: # Assembled at run time so the distributed source carries no literal # absolute-home sentinel; the composed path is the exact hostile value. checkout = Path("/home", "example", "projects", "ultimate-tool") forbidden = [str(path) for path in verify.default_forbidden(checkout)] self.assertIn(str(checkout), forbidden) self.assertIn(str(checkout.parent), forbidden) self.assertIn("/var/run/docker.sock", forbidden)
def test_the_parser_defaults_its_checkout_root_to_this_repository(self) -> None: """The production default is unchanged; the mocked tests name a root.""" args = verify.build_parser().parse_args(["--mode", "runtime", "--release-dir", str(self.root)]) self.assertEqual(args.checkout_root.resolve(), ROOT.resolve())
def test_a_checkout_directly_under_a_filesystem_root_still_forbids_every_scratch(self) -> None: """The CI layout, pinned so the policy is never relaxed to fit it.
A checkout at ``/work`` makes its parent ``/`` forbidden. The guard refuses any scratch directory on that volume, and that refusal is the production policy rather than something the tests may switch off. """ simulated = Path("/work") forbidden = verify.default_forbidden(simulated) self.assertIn(Path("/").resolve(), [path.resolve() for path in forbidden]) scratch = (self.root / "scratch").resolve() problems = verify.isolation_problems( checkout_root=simulated, mounts=[{"source": str(scratch), "target": "/work"}], forbidden_paths=forbidden, expected_source=scratch, ) self.assertTrue(any("forbidden path" in item for item in problems), problems)
# ---------------------------------------------------------------------------# Release directory checksum chain# ---------------------------------------------------------------------------
class ReleaseDirectoryTests(TemporaryDirectoryCase): def build_release(self) -> Path: release = self.root / "dist" release.mkdir() source_members = { "Cargo.toml": b"[workspace]\n", "Cargo.lock": b"# lock\n", } source_archive = make_archive( release / f"{SOURCE_PREFIX}.tar.gz", [(f"{SOURCE_PREFIX}/{name}", content, tarfile.REGTYPE) for name, content in source_members.items()], ) files = [ {"path": name, "sha256": sha256(content), "size": len(content)} for name, content in sorted(source_members.items()) ] (release / "source-manifest.json").write_text( json.dumps( { "schema": 1, "record_kind": "release_source_manifest_resolved", "archive": {"name": source_archive.name, "sha256": sha256(source_archive.read_bytes())}, "files": files, "aggregate_sha256": verify.aggregate_digest(files), "vendor": {"files": []}, } ) ) binaries = {} binary_members = [] for name in verify.BINARY_NAMES: content = f"binary {name}\n".encode("utf-8") binaries[name] = {"sha256": sha256(content), "size": len(content)} binary_members.append((f"{BINARY_PREFIX}/bin/{name}", content, tarfile.REGTYPE)) binary_archive = make_archive(release / f"{BINARY_PREFIX}.tar.gz", binary_members) (release / "release-manifest.json").write_text( json.dumps( { "schema": 1, "version": "0.2.0-preview.1", "development": "local-test", "release_ready": False, "source_completeness": {"status": "incomplete_pending_third_party_notices"}, "artifacts": [ {"name": name, "kind": "host-binary", "path": f"bin/{name}", **record} for name, record in sorted(binaries.items()) ], "source_archive": { "name": source_archive.name, "sha256": sha256(source_archive.read_bytes()), }, "archive": {"name": binary_archive.name, "sha256": sha256(binary_archive.read_bytes())}, "build_environment": { "base_image_digest": verify.PINNED_BASE_DIGEST, "rust_version": verify.PINNED_RUST_VERSION, "target_triple": verify.PINNED_TARGET_TRIPLE, }, } ) ) (release / "SHA256SUMS").write_text( f"{sha256(binary_archive.read_bytes())} {binary_archive.name}\n" f"{sha256(source_archive.read_bytes())} {source_archive.name}\n" ) return release
def test_accepts_a_consistent_release_directory(self) -> None: summary = verify.verify_release_directory(self.build_release()) self.assertEqual(summary["problems"], []) self.assertEqual(summary["version"], "0.2.0-preview.1") self.assertEqual(summary["development_label"], "local-test") self.assertFalse(summary["release_ready_as_recorded"]) self.assertEqual(summary["source_archive"]["name"], f"{SOURCE_PREFIX}.tar.gz") self.assertEqual(set(summary["packaged_binaries"]), set(verify.BINARY_NAMES))
def test_detects_a_tampered_archive(self) -> None: release = self.build_release() with open(release / f"{SOURCE_PREFIX}.tar.gz", "ab") as handle: handle.write(b"tamper") summary = verify.verify_release_directory(release) self.assertTrue(summary["problems"]) self.assertTrue(any("SHA256SUMS" in item for item in summary["problems"]), summary["problems"])
def test_reports_a_missing_sums_file(self) -> None: release = self.build_release() (release / "SHA256SUMS").unlink() summary = verify.verify_release_directory(release) self.assertTrue(any("SHA256SUMS is missing" in item for item in summary["problems"]))
def test_refuses_a_directory_without_manifests(self) -> None: empty = self.root / "empty" empty.mkdir() with self.assertRaises(verify.HarnessError): verify.verify_release_directory(empty)
# ---------------------------------------------------------------------------# Rebuild evidence, proven consumable by the packager itself# ---------------------------------------------------------------------------
class RebuildEvidenceTests(ReleaseDirectoryTests): def build_evidence(self, release: Path, *, environment=None, commands=None, outputs=None) -> tuple[Path, dict, dict]: source_manifest = json.loads((release / "source-manifest.json").read_text()) release_manifest = json.loads((release / "release-manifest.json").read_text()) binaries = { entry["name"]: {"sha256": entry["sha256"], "size": entry["size"]} for entry in release_manifest["artifacts"] } source_archive = { "name": source_manifest["archive"]["name"], "sha256": source_manifest["archive"]["sha256"], "aggregate_sha256": source_manifest["aggregate_sha256"], } default_outputs = [ { "name": name, "sha256": binaries[name]["sha256"], "size": binaries[name]["size"], "path": f"target/release/{name}", } for name in verify.BINARY_NAMES ] journal = verify.Journal() journal.record( verify.CommandRecord( argv=verify.BUILD_COMMAND, exit_code=0, duration_ms=12, cwd="/work", stdout="Finished\n", ) ) document = verify.rebuild_evidence_document( source_archive=source_archive, environment=environment or { "base_image_digest": verify.PINNED_BASE_DIGEST, "rustc": "rustc 1.93.0 (254b59607 2025-01-06)", "target_triple": verify.PINNED_TARGET_TRIPLE, }, commands=commands or [{"argv": verify.BUILD_COMMAND, "exit_code": 0, "cwd": "/work", "network": "none"}], outputs=outputs if outputs is not None else default_outputs, journal=journal, ) path = self.root / "rebuild-evidence.json" path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n") return path, source_archive, release_manifest
def validate(self, path: Path, source_archive: dict, release_manifest: dict) -> dict: binaries = { entry["name"]: {"sha256": entry["sha256"], "size": entry["size"]} for entry in release_manifest["artifacts"] } return packaging.validate_rebuild_evidence( path, source_archive=source_archive, binaries=binaries, manifest=release_manifest, )
def test_the_packager_accepts_the_record_the_harness_writes(self) -> None: release = self.build_release() path, source_archive, release_manifest = self.build_evidence(release) result = self.validate(path, source_archive, release_manifest) self.assertEqual(result["status"], "verified") self.assertEqual(result["evidence_sha256"], sha256(path.read_bytes()))
def test_the_packager_refuses_a_record_without_the_documented_command(self) -> None: release = self.build_release() path, source_archive, release_manifest = self.build_evidence( release, commands=[{"argv": ["cargo", "build", "--release", "-p", "ultimate-tool"], "exit_code": 0}], ) with self.assertRaises(PACKAGER_FAILURE) as caught: self.validate(path, source_archive, release_manifest) self.assertIn("documented command", str(caught.exception))
def test_the_packager_refuses_a_record_from_another_base_image(self) -> None: release = self.build_release() path, source_archive, release_manifest = self.build_evidence( release, environment={ "base_image_digest": "sha256:" + "0" * 64, "rustc": "rustc 1.93.0 (254b59607 2025-01-06)", "target_triple": verify.PINNED_TARGET_TRIPLE, }, ) with self.assertRaises(PACKAGER_FAILURE) as caught: self.validate(path, source_archive, release_manifest) self.assertIn("pinned base image", str(caught.exception))
def test_the_packager_refuses_a_record_with_a_nonzero_exit_code(self) -> None: release = self.build_release() path, source_archive, release_manifest = self.build_evidence( release, commands=[{"argv": verify.BUILD_COMMAND, "exit_code": 101}], ) with self.assertRaises(PACKAGER_FAILURE) as caught: self.validate(path, source_archive, release_manifest) self.assertIn("exit code 101", str(caught.exception))
def test_the_packager_refuses_a_record_that_omits_a_binary(self) -> None: release = self.build_release() path, source_archive, release_manifest = self.build_evidence( release, outputs=[{"name": "ultimate-tool", "sha256": "0" * 64, "size": 1}], ) with self.assertRaises(PACKAGER_FAILURE) as caught: self.validate(path, source_archive, release_manifest) self.assertIn("ultimate-tool-mcp", str(caught.exception))
def test_the_record_is_bound_to_the_prepared_source_archive(self) -> None: release = self.build_release() path, source_archive, release_manifest = self.build_evidence(release) document = json.loads(path.read_text()) self.assertEqual(document["source_archive"]["sha256"], source_archive["sha256"]) self.assertEqual(document["source_aggregate_sha256"], source_archive["aggregate_sha256"]) self.assertEqual(document["record_kind"], "release_rebuild_evidence") self.assertEqual(document["schema"], verify.SCHEMA)
def test_a_record_over_the_evidence_bound_is_reduced_not_rejected(self) -> None: document = { "schema": 1, "record_kind": "release_rebuild_evidence", "harness": {"commands": [{"argv": ["cargo", "build"], "stdout_tail": "x" * 4000}], "notes": []}, "commands": [{"argv": verify.BUILD_COMMAND, "exit_code": 0}], } with mock.patch.object(verify, "MAX_EVIDENCE_BYTES", 512): verify._fit_evidence_bound(document) self.assertEqual(document["harness"]["commands"][0]["stdout_tail"], "") self.assertTrue(any("evidence bound" in note for note in document["harness"]["notes"]))
# ---------------------------------------------------------------------------# Runtime assets: shim behaviour, driver structure, generated image# ---------------------------------------------------------------------------
class RuntimeShimTests(TemporaryDirectoryCase): def shim(self) -> Path: path = self.root / "ultimate-tool-runtime-shim" path.write_text(verify.RUNTIME_SHIM_SOURCE) path.chmod(0o755) return path
def test_the_shim_refuses_and_records_the_attempt(self) -> None: journal = self.root / "attempts.jsonl" environment = dict(os.environ, SHIM_JOURNAL=str(journal)) result = subprocess.run( [str(self.shim()), "--probe", "value"], capture_output=True, env=environment, timeout=30, check=False, ) self.assertEqual(result.returncode, 127) self.assertIn(b"isolated release runtime", result.stderr) lines = journal.read_text().splitlines() self.assertEqual(len(lines), 1) self.assertIn('"tool":"ultimate-tool-runtime-shim"', lines[0]) self.assertIn("--probe value", lines[0])
def test_the_shim_records_every_attempt_under_its_shipped_name(self) -> None: journal = self.root / "attempts.jsonl" shim = self.shim() environment = dict(os.environ, SHIM_JOURNAL=str(journal)) for tool in ("docker", "cargo", "x64sc"): link = self.root / tool link.symlink_to(shim) result = subprocess.run([str(link)], capture_output=True, env=environment, timeout=30, check=False) self.assertEqual(result.returncode, 127) lines = journal.read_text().splitlines() self.assertEqual(len(lines), 3) self.assertIn('"tool":"docker"', lines[0]) self.assertIn('"tool":"cargo"', lines[1]) self.assertIn('"tool":"x64sc"', lines[2])
def test_an_unset_journal_does_not_break_the_refusal(self) -> None: environment = {key: value for key, value in os.environ.items() if key != "SHIM_JOURNAL"} environment["SHIM_JOURNAL"] = str(self.root / "absent" / "attempts.jsonl") result = subprocess.run([str(self.shim())], capture_output=True, env=environment, timeout=30, check=False) self.assertEqual(result.returncode, 127)
class RuntimeDriverStructureTests(TemporaryDirectoryCase): def test_the_driver_compiles(self) -> None: compile(verify.RUNTIME_DRIVER_SOURCE, "<runtime_driver>", "exec")
def test_readiness_is_deadline_driven_not_a_blocking_read(self) -> None: source = verify.RUNTIME_DRIVER_SOURCE self.assertIn("select.select", source) self.assertNotIn("readline()", source) self.assertIn("options.ready_timeout", source)
def test_the_driver_uses_an_ephemeral_port_and_strict_origin(self) -> None: source = verify.RUNTIME_DRIVER_SOURCE self.assertIn('"--port", "0"', source) self.assertIn("invalid_mutation_token", source) self.assertIn("invalid_origin", source) self.assertIn('"--target", "fake"', source) self.assertIn("--runs", source)
def test_the_driver_terminates_the_service_it_started(self) -> None: source = verify.RUNTIME_DRIVER_SOURCE self.assertIn("process.terminate()", source) self.assertIn("process.kill()", source) self.assertIn('service["terminated"]', source)
def test_the_driver_proves_absence_by_inspection(self) -> None: source = verify.RUNTIME_DRIVER_SOURCE self.assertIn("shim_attempts", source) # The absence checks must not invoke the shimmed tools, or the empty # journal would be impossible by construction. self.assertIn("shutil.which", source) self.assertIn("/usr/local/cargo", source)
def test_the_runtime_image_is_a_fresh_pinned_base_with_only_python(self) -> None: context = verify.runtime_image_context(self.root, verify.PINNED_BASE_REFERENCE) dockerfile = (context / "Dockerfile").read_text() self.assertIn(f"FROM {verify.PINNED_BASE_REFERENCE}", dockerfile) self.assertIn("apt-get install --no-install-recommends --yes python3", dockerfile) self.assertIn("FROM ubuntu:24.04@sha256:", dockerfile) for tool in ("docker", "vice", "x64sc", "cargo", "rustc", "cc", "gcc"): self.assertIn(tool, dockerfile) self.assertNotIn("rust:1.93", dockerfile) self.assertNotIn("vice-emu", dockerfile) self.assertTrue((context / "shim").is_file()) self.assertEqual(stat.S_IMODE((context / "shim").stat().st_mode), 0o755)
# ---------------------------------------------------------------------------# Readiness parsing, cleanup and CLI behaviour# ---------------------------------------------------------------------------
class ReadinessTests(unittest.TestCase): def test_parses_the_documented_dashboard_line(self) -> None: self.assertEqual( verify.parse_dashboard_url("Ultimate Tool dashboard: http://127.0.0.1:41337\n"), "http://127.0.0.1:41337", )
def test_ignores_unrelated_output(self) -> None: self.assertIsNone(verify.parse_dashboard_url("building\nerror: nothing here\n")) self.assertIsNone(verify.parse_dashboard_url("Ultimate Tool dashboard: http://0.0.0.0:80\n"))
class OwnedProcessCleanupTests(unittest.TestCase): def test_a_started_child_is_terminated_and_gone(self) -> None: process = subprocess.Popen(["/bin/sh", "-c", "exec sleep 300"]) try: result = verify.terminate_process(process, grace=5.0) finally: if process.poll() is None: process.kill() process.wait(timeout=10) self.assertTrue(result["terminated"], result) self.assertIsNotNone(result["returncode"]) self.assertIsNotNone(process.poll()) with self.assertRaises(ProcessLookupError): os.kill(result["pid"], 0)
def test_cleanup_runs_actions_once_in_reverse_order(self) -> None: calls: list[str] = [] cleanup = verify.Cleanup() cleanup.add("first", lambda: calls.append("first")) cleanup.add("second", lambda: calls.append("second")) cleanup.run() cleanup.run() self.assertEqual(calls, ["second", "first"])
def test_a_failing_cleanup_action_does_not_hide_the_rest(self) -> None: calls: list[str] = []
def boom() -> None: raise RuntimeError("cleanup failure")
cleanup = verify.Cleanup() cleanup.add("boom", boom) cleanup.add("after", lambda: calls.append("after")) cleanup.run() self.assertEqual(calls, ["after"])
def test_the_harness_removes_containers_carrying_its_label(self) -> None: with mock.patch.object(verify, "docker_executable", return_value="docker"): with mock.patch.object(subprocess, "run") as runner: runner.return_value = mock.Mock(returncode=0, stdout=b"abc\n") removed = verify.remove_containers_with_label(verify.CONTAINER_LABEL) self.assertEqual(removed, ["abc"]) first_call = runner.call_args_list[0][0][0] self.assertIn("label=" + verify.CONTAINER_LABEL, first_call)
class CommandLineTests(TemporaryDirectoryCase): def test_the_harness_only_agrees_with_the_packager_on_its_constants(self) -> None: policy = verify.cross_check_policy() self.assertTrue(policy["checked"]) self.assertEqual(list(packaging.BUILD_COMMAND), list(verify.BUILD_COMMAND)) self.assertEqual(tuple(packaging.BINARY_NAMES), tuple(verify.BINARY_NAMES))
def test_check_evidence_requires_an_evidence_path(self) -> None: with self.assertRaises(SystemExit) as caught: verify.main(["--mode", "check-evidence", "--release-dir", str(self.root)]) self.assertEqual(caught.exception.code, verify.EXIT_USAGE)
def test_an_out_of_range_run_count_is_a_usage_error(self) -> None: with self.assertRaises(SystemExit) as caught: verify.main(["--mode", "runtime", "--release-dir", str(self.root), "--runs", "0"]) self.assertEqual(caught.exception.code, verify.EXIT_USAGE)
def test_missing_docker_is_reported_as_environment_unavailable(self) -> None: environment = dict(os.environ, ULTIMATE_TOOL_DOCKER=str(self.root / "no-such-docker")) with mock.patch.dict(os.environ, environment, clear=False): code = verify.main(["--mode", "runtime", "--release-dir", str(self.root)]) self.assertEqual(code, verify.EXIT_ENVIRONMENT)
def test_a_missing_release_directory_fails_verification_not_environment(self) -> None: release = self.root / "no-such-dist" with mock.patch.object(verify, "docker_available", return_value=(True, "test")): code = verify.main( ["--mode", "source-rebuild", "--release-dir", str(release), "--image", "unused"] ) self.assertEqual(code, verify.EXIT_VERIFICATION_FAILED)
def test_an_existing_record_is_never_overwritten_without_replace(self) -> None: target = self.root / "record.json" first = verify.write_new_file(target, b'{"a":1}\n') self.assertFalse(first["replaced"]) with self.assertRaises(verify.HarnessError): verify.write_new_file(target, b'{"a":2}\n') self.assertEqual(target.read_bytes(), b'{"a":1}\n') second = verify.write_new_file(target, b'{"a":3}\n', replace=True) self.assertTrue(second["replaced"]) self.assertEqual(target.read_bytes(), b'{"a":3}\n')
def test_a_record_never_overwrites_through_a_symlinked_directory(self) -> None: real = self.root / "real" real.mkdir() link = self.root / "link" link.symlink_to(real) with self.assertRaises(verify.HarnessError): verify.write_new_file(link / "record.json", b"{}\n") self.assertFalse((real / "record.json").exists())
class CommandRecordTests(unittest.TestCase): def test_a_command_record_keeps_digests_not_just_text(self) -> None: record = verify.CommandRecord( argv=["cargo", "build"], exit_code=0, duration_ms=5, stdout="Finished\n", stderr="", note="documented build", ) payload = record.as_json() self.assertEqual(payload["exit_code"], 0) self.assertEqual(payload["stdout_sha256"], sha256(b"Finished\n")) self.assertEqual(payload["note"], "documented build") self.assertEqual(payload["stdout_tail"], "Finished\n")
def test_the_journal_keeps_command_and_mount_order(self) -> None: journal = verify.Journal() journal.mount("/scratch", "/work", "rw", "purpose") journal.record(verify.CommandRecord(argv=["docker", "image", "inspect"], exit_code=0, duration_ms=1)) journal.note("a note") payload = journal.as_json() self.assertEqual(len(payload["commands"]), 1) self.assertEqual(payload["mounts"][0]["target"], "/work") self.assertEqual(payload["notes"], ["a note"])
def test_aggregate_digest_matches_the_packager_recipe(self) -> None: entries = [ {"path": "b.rs", "sha256": "1" * 64}, {"path": "a.rs", "sha256": "2" * 64}, ] self.assertEqual(verify.aggregate_digest(entries), packaging.aggregate_digest(entries))
class BoundedReadTests(TemporaryDirectoryCase): def test_reading_a_symlink_leaf_is_refused(self) -> None: target = self.write("real.json", "{}\n") link = self.root / "link.json" link.symlink_to(target) with self.assertRaises(verify.HarnessError): verify.read_regular_file(link)
def test_reading_a_directory_is_refused(self) -> None: directory = self.root / "dir" directory.mkdir() with self.assertRaises(verify.HarnessError): verify.read_regular_file(directory)
def test_a_file_above_the_bound_is_refused(self) -> None: path = self.write("big.json", b"x" * 4096) with self.assertRaises(verify.HarnessError): verify.read_regular_file(path, max_bytes=128)
def test_invalid_json_is_reported_with_the_path(self) -> None: path = self.write("bad.json", "{not json") with self.assertRaises(verify.HarnessError) as caught: verify.load_json_file(path, "release manifest") self.assertIn("not valid JSON", str(caught.exception))
class PreparedSourceInventoryTests(unittest.TestCase): def manifest(self) -> dict: return { "archive": {"prefix": SOURCE_PREFIX}, "files": [{"path": "Cargo.toml", "sha256": "a" * 64, "size": 1}], "vendor": {"files": [{"path": "serde/src/lib.rs", "sha256": "b" * 64, "size": 2}]}, }
def test_the_inventory_is_the_manifest_files_plus_the_generated_members(self) -> None: inventory = verify.prepared_source_inventory(self.manifest()) self.assertEqual(inventory["prefix"], SOURCE_PREFIX) self.assertEqual( set(inventory["expected"]), {f"{SOURCE_PREFIX}/Cargo.toml", f"{SOURCE_PREFIX}/vendor/serde/src/lib.rs"}, ) self.assertEqual( inventory["expected"][f"{SOURCE_PREFIX}/vendor/serde/src/lib.rs"]["sha256"], "b" * 64 ) self.assertIn(f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", inventory["required"]) self.assertIn(f"{SOURCE_PREFIX}/vendor-sources.toml", inventory["required"])
def test_every_inventory_key_carries_the_archive_prefix(self) -> None: # Members are read out of the tar by their full path, so an unprefixed # inventory could never match a prepared archive. inventory = verify.prepared_source_inventory(self.manifest()) for key in list(inventory["expected"]) + list(inventory["required"]): self.assertTrue(key.startswith(f"{SOURCE_PREFIX}/"), key)
def test_an_unvendored_archive_requires_no_vendor_configuration(self) -> None: manifest = self.manifest() manifest["vendor"] = {"files": []} inventory = verify.prepared_source_inventory(manifest) self.assertEqual(set(inventory["expected"]), {f"{SOURCE_PREFIX}/Cargo.toml"}) self.assertEqual(list(inventory["required"]), [f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json"])
def test_a_manifest_without_a_recorded_prefix_is_refused(self) -> None: with self.assertRaises(verify.HarnessError) as caught: verify.prepared_source_inventory({"files": [], "archive": {}}) self.assertIn("no archive prefix", str(caught.exception))
def test_the_packager_module_is_reused_when_it_is_importable(self) -> None: self.assertIsNotNone(verify._packager_module())
class WorkDirectorySafetyTests(TemporaryDirectoryCase): """Cleanup must never be able to remove anything the caller owns."""
def test_an_owned_child_is_created_inside_the_parent(self) -> None: parent = self.root / "parent" parent.mkdir() verify.INVOCATION = "test-invocation" owned, removable = verify.owned_scratch_directory(parent, "runtime", []) self.assertEqual(owned.parent, parent) self.assertTrue(owned.is_dir()) self.assertEqual(removable, owned) self.assertNotEqual(owned, parent) self.assertIn("test-invocation", owned.name)
def test_cleanup_removes_only_the_owned_child_and_spares_the_parent(self) -> None: parent = self.root / "parent" parent.mkdir() sentinel = parent / "keep-me.txt" sentinel.write_text("must survive\n", encoding="utf-8") verify.INVOCATION = "test-invocation" owned, removable = verify.owned_scratch_directory(parent, "runtime", []) (owned / "scratch.bin").write_bytes(b"x" * 32) cleanup = verify.Cleanup() cleanup.add(f"remove {removable}", lambda: shutil.rmtree(removable, ignore_errors=True)) cleanup.run() self.assertFalse(owned.exists()) self.assertEqual(sentinel.read_text(encoding="utf-8"), "must survive\n") self.assertTrue(parent.is_dir())
def test_a_parent_inside_a_forbidden_path_is_refused(self) -> None: checkout = self.root / "checkout" checkout.mkdir() verify.INVOCATION = "test-invocation" with self.assertRaises(verify.HarnessError) as caught: verify.owned_scratch_directory(checkout, "runtime", [checkout]) self.assertIn("forbidden path", str(caught.exception)) self.assertFalse((checkout / "ut-verify-runtime-test-invocation").exists())
def test_a_parent_that_is_not_a_directory_is_refused(self) -> None: verify.INVOCATION = "test-invocation" with self.assertRaises(verify.HarnessError): verify.owned_scratch_directory(self.root / "nope", "runtime", [])
def test_a_fresh_scratch_directory_is_created_when_no_parent_is_given(self) -> None: verify.INVOCATION = "test-invocation" owned, removable = verify.owned_scratch_directory(None, "runtime", []) self.addCleanup(shutil.rmtree, owned, ignore_errors=True) self.assertTrue(owned.is_dir()) self.assertEqual(removable, owned)
class InvocationScopeTests(unittest.TestCase): def test_each_invocation_gets_its_own_container_label(self) -> None: original = verify.INVOCATION self.addCleanup(setattr, verify, "INVOCATION", original) verify.INVOCATION = "one" first = verify.container_label() verify.INVOCATION = "two" second = verify.container_label() self.assertNotEqual(first, second) self.assertTrue(first.startswith(verify.CONTAINER_LABEL)) self.assertEqual(len({verify.new_invocation_id() for _ in range(5)}), 5)
class BoundedCaptureTests(unittest.TestCase): def test_command_output_is_bounded_while_the_child_is_drained(self) -> None: journal = verify.Journal() with mock.patch.object(verify, "MAX_COMMAND_OUTPUT", 512): record = verify.run_command( ["/bin/sh", "-c", "head -c 200000 /dev/zero | tr '\\0' 'x'; echo done"], journal=journal, timeout=120, ) self.assertEqual(record.exit_code, 0) self.assertLessEqual(len(record.stdout), 512) self.assertTrue(record.truncated) self.assertEqual(len(journal.commands), 1)
def test_a_timed_out_command_is_killed_and_still_recorded(self) -> None: journal = verify.Journal() with self.assertRaises(verify.HarnessError) as caught: verify.run_command(["/bin/sh", "-c", "exec sleep 120"], journal=journal, timeout=1) self.assertIn("timed out", str(caught.exception)) self.assertEqual(len(journal.commands), 1)
def add_member(handle: tarfile.TarFile, name: str, content: bytes, mode: int = 0o644) -> None: info = tarfile.TarInfo(name) info.size = len(content) info.mode = mode handle.addfile(info, io.BytesIO(content))
def write_prepared_archive(directory: Path, *, vendored: bool = True) -> tuple[Path, dict]: """A prepared-style source archive and the manifest describing it.
It has the three parts a real one has: the first-party files and vendored crate files the manifest lists, and the members the packager *generates* while writing the archive, which the file lists never mention. """ files = { "Cargo.toml": b'[workspace]\nmembers = ["crates/x"]\n', "crates/x/Cargo.toml": b'[package]\nname = "x"\nversion = "0.2.0-preview.1"\n', "Cargo.lock": b"# lock\n", } vendor_files = ( {"serde/Cargo.toml": b'[package]\nname = "serde"\n', "serde/src/lib.rs": b"// serde\n"} if vendored else {} ) generated = {"SOURCE-PROVENANCE.json": b'{"producer": "tools/package_release.py"}\n'} if vendored: generated["vendor-sources.toml"] = ( b'[source.crates-io]\nreplace-with = "vendored-sources"\n\n' b'[source.vendored-sources]\ndirectory = "vendor"\n' ) archive = directory / f"{SOURCE_PREFIX}.tar.gz" with tarfile.open(archive, "w:gz") as handle: for name, content in sorted(files.items()): add_member(handle, f"{SOURCE_PREFIX}/{name}", content) for name, content in sorted(vendor_files.items()): add_member(handle, f"{SOURCE_PREFIX}/vendor/{name}", content) for name, content in sorted(generated.items()): add_member(handle, f"{SOURCE_PREFIX}/{name}", content) file_list = [{"path": n, "sha256": sha256(c), "size": len(c)} for n, c in sorted(files.items())] vendor_list = [{"path": n, "sha256": sha256(c), "size": len(c)} for n, c in sorted(vendor_files.items())] manifest = { "schema": 1, "record_kind": "release_source_manifest_resolved", "archive": { "name": archive.name, "prefix": SOURCE_PREFIX, "sha256": sha256(archive.read_bytes()), }, "files": file_list, "aggregate_sha256": verify.aggregate_digest(file_list), "vendor": {"files": vendor_list, "aggregate_sha256": verify.aggregate_digest(vendor_list)}, "source_completeness": {"status": "complete_pending_clean_rebuild_verification"}, } return archive, manifest
class PreparedSourceContractTests(TemporaryDirectoryCase): """The inventory, the extraction and the packager must agree on one archive."""
def test_the_inventory_extracts_the_prepared_archive_the_packager_wrote(self) -> None: archive, manifest = write_prepared_archive(self.root) inventory = verify.prepared_source_inventory(manifest) (self.root / "out").mkdir() result = verify.extract_verified_archive( archive, self.root / "out", expected=inventory["expected"], required=inventory["required"], ) self.assertEqual(result["prefix"], SOURCE_PREFIX) self.assertEqual(result["extras"], []) self.assertEqual(result["members"], 7) self.assertEqual( result["actual_bytes"], sum(entry["size"] for entry in result["files"].values()) )
def test_unprefixed_expectations_are_reported_missing_not_silently_accepted(self) -> None: # The defect this replaced: an unprefixed inventory could never match the # archive, so it must fail loudly instead of appearing to match. archive, manifest = write_prepared_archive(self.root) unprefixed = {entry["path"]: {"sha256": entry["sha256"]} for entry in manifest["files"]} (self.root / "out").mkdir() with self.assertRaises(verify.HarnessError) as caught: verify.extract_verified_archive(archive, self.root / "out", expected=unprefixed) self.assertIn("is missing", str(caught.exception))
def test_the_packagers_own_verifier_accepts_the_same_archive(self) -> None: archive, manifest = write_prepared_archive(self.root) check = verify.verify_prepared_source_with_packager(archive, manifest) self.assertTrue(check["checked"], check) self.assertEqual(check["verified_members"], 5) self.assertEqual(check["prefix"], SOURCE_PREFIX)
def test_an_unvendored_prepared_archive_needs_no_vendor_configuration(self) -> None: archive, manifest = write_prepared_archive(self.root, vendored=False) inventory = verify.prepared_source_inventory(manifest) (self.root / "out").mkdir() result = verify.extract_verified_archive( archive, self.root / "out", expected=inventory["expected"], required=inventory["required"], ) self.assertEqual(result["extras"], []) self.assertEqual(result["members"], 4)
def test_extraction_fails_closed_without_the_packager_module(self) -> None: archive, _ = write_prepared_archive(self.root) destination = self.root / "out" destination.mkdir() with mock.patch.object(verify, "_packager_module", return_value=None): with self.assertRaises(verify.EnvironmentUnavailable) as caught: verify.extract_verified_archive(archive, destination) self.assertIn("package_release", str(caught.exception)) self.assertEqual(list(destination.iterdir()), [])
class MockedRebuildSequenceTests(TemporaryDirectoryCase): """The rebuild gate's call sequence, driven **without Docker**.
This proves the extraction, the vendor-configuration installation, the observed environment descriptor and the documented build command are wired in the right order, and that the resulting record binds the packaged binaries and is accepted by the packager's own validator.
No container ran and nothing was compiled. This is a mocked path: it is evidence about the harness's wiring, and it must never be reported as a verified rebuild. """
DESCRIPTOR = { "schema": 1, "record_kind": "build_environment", "observed_at": "2026-01-01T00:00:00Z", "os": {"id": "ubuntu", "version_id": "24.04"}, "glibc": {"version": "2.39"}, "target_triple": verify.PINNED_TARGET_TRIPLE, "toolchain": { "rustc": "rustc 1.93.0 (254b59607 2025-01-06)", "cargo": "cargo 1.93.0 (083ac5135 2025-01-06)", "python": "Python 3.12.3", "node": "v22.23.3", }, "base_image": { "reference": verify.PINNED_BASE_REFERENCE, "digest": verify.PINNED_BASE_DIGEST, "pinned": True, }, }
def setUp(self) -> None: super().setUp() # Nothing in this class may reach the daemon, so every Docker seam the # gate can call is stubbed here and a real invocation is refused outright. self.container_commands: list[list[str]] = [] self.container_environments: list[dict] = [] self.container_removals: list[str] = [] self.image_identities: list[str] = [] self.container_handler = None guard = forbid_docker_subprocesses() guard.__enter__() self.addCleanup(guard.__exit__, None, None, None) for patcher in ( mock.patch.object(verify, "docker_available", return_value=(True, "stubbed: no daemon is asked")), mock.patch.object(verify, "docker_image_present", return_value=True), mock.patch.object(verify, "docker_image_identity", side_effect=self.stub_image_identity), mock.patch.object(verify, "remove_containers_with_label", side_effect=self.stub_container_removal), mock.patch.object(verify, "docker_run", side_effect=self.run_container_command), ): patcher.start() self.addCleanup(patcher.stop)
def stub_image_identity(self, tag: str) -> dict: """The local image identity the gate records, without a daemon.""" self.image_identities.append(tag) return { "tag": tag, "id": "sha256:stubbed-image-identity", "provenance": "stubbed by the mocked rebuild tests; no daemon was asked", }
def stub_container_removal(self, label: str) -> list[str]: """Container cleanup by label, without ``docker ps`` or ``docker rm``.""" self.container_removals.append(label) return []
def run_container_command(self, **kwargs): """Route one stubbed container command to the test's own stand-in.""" argv = [str(item) for item in kwargs.get("argv") or []] self.container_commands.append(argv) self.container_environments.append(dict(kwargs.get("environment") or {})) if self.container_handler is None: raise AssertionError(f"a container command ran in a mocked test: {argv}") return self.container_handler(**kwargs)
def synthetic_checkout(self) -> Path: """A checkout root that cannot contain this test's scratch directory.
The production policy forbids the checkout, its parent and the Docker socket, and it is checked as an ancestor. A checkout that sits directly under a filesystem root — ``/work`` in CI — therefore forbids ``/`` itself, so inheriting the module's real root would refuse any scratch directory on the volume. The mocked tests name a checkout that is never created and whose parent holds nothing else. """ return self.root / "unmounted" / "checkout"
def test_the_mocked_sequence_uses_a_synthetic_checkout(self) -> None: """The mocked gate never inherits the module's real checkout root.
Inheriting it would forbid the whole volume whenever the repository sits directly under a filesystem root, and the guard would refuse a scratch directory the test never asked to be connected to the checkout. """ checkout = self.synthetic_checkout() forbidden = verify.default_forbidden(checkout) scratch = (self.root / "work").resolve() self.assertNotEqual(checkout.resolve(), ROOT.resolve()) self.assertNotIn(ROOT.resolve(), checkout.resolve().parents) self.assertEqual( verify.isolation_problems( checkout_root=checkout, mounts=[{"source": str(scratch), "target": "/work"}], forbidden_paths=forbidden, expected_source=scratch, ), [], )
def test_a_real_docker_subprocess_is_refused_not_run(self) -> None: """The guard, not a daemon, is what keeps these tests offline.
A mocked test that forgets a seam fails here instead of starting a container, and the refusal covers one test only: an unrelated command still runs, so nothing global — ``PATH`` above all — was rewritten. """ with self.assertRaises(DockerInvocationForbidden): verify.docker_executable() with self.assertRaises(DockerInvocationForbidden): subprocess.run(["docker", "image", "inspect", verify.RELEASE_ENV_IMAGE], check=False) with self.assertRaises(DockerInvocationForbidden): subprocess.Popen(["docker", "ps", "--all"]) self.assertEqual(subprocess.run(["/bin/sh", "-c", "exit 0"], check=False).returncode, 0)
def test_the_image_identity_and_the_label_cleanup_are_stubbed(self) -> None: """Both seams the gate reaches outside a container are stubbed.""" identity = verify.docker_image_identity(verify.RELEASE_ENV_IMAGE) self.assertEqual(identity["tag"], verify.RELEASE_ENV_IMAGE) self.assertIn("stubbed", identity["id"]) label = verify.container_label() self.assertEqual(verify.remove_containers_with_label(label), []) self.assertEqual(self.image_identities, [verify.RELEASE_ENV_IMAGE]) self.assertEqual(self.container_removals, [label])
def test_a_container_command_without_a_stand_in_fails_the_test(self) -> None: """An unexpected container step is a failure, never a silent success.""" with self.assertRaises(AssertionError): verify.docker_run(image=verify.RELEASE_ENV_IMAGE, argv=["true"], journal=verify.Journal())
def build_release(self) -> tuple[Path, Path, dict, dict]: release = self.root / "dist" release.mkdir() source_archive, manifest = write_prepared_archive(release) (release / "source-manifest.json").write_text(json.dumps(manifest), encoding="utf-8") binaries = {name: f"packaged {name}\n".encode("utf-8") for name in verify.BINARY_NAMES} binary_archive = release / f"{BINARY_PREFIX}.tar.gz" with tarfile.open(binary_archive, "w:gz") as handle: for name, content in sorted(binaries.items()): add_member(handle, f"{BINARY_PREFIX}/bin/{name}", content, mode=0o755) (release / "release-manifest.json").write_text( json.dumps( { "schema": 1, "version": "0.2.0-preview.1", "development": "local-mocked-test", "release_ready": False, "artifacts": [ { "name": name, "kind": "host-binary", "path": f"bin/{name}", "sha256": sha256(content), "size": len(content), } for name, content in sorted(binaries.items()) ], "source_archive": { "name": source_archive.name, "sha256": sha256(source_archive.read_bytes()), }, "archive": { "name": binary_archive.name, "sha256": sha256(binary_archive.read_bytes()), }, "build_environment": { "base_image_digest": verify.PINNED_BASE_DIGEST, "rust_version": verify.PINNED_RUST_VERSION, "target_triple": verify.PINNED_TARGET_TRIPLE, }, } ), encoding="utf-8", ) (release / "SHA256SUMS").write_text( f"{sha256(binary_archive.read_bytes())} {binary_archive.name}\n" f"{sha256(source_archive.read_bytes())} {source_archive.name}\n", encoding="utf-8", ) return release, source_archive, manifest, binaries
def test_the_rebuild_gate_is_driven_in_order_without_docker(self) -> None: release, source_archive, manifest, binaries = self.build_release() calls = self.container_commands environments = self.container_environments
def fake_docker_run( *, image, argv, journal, mount=None, mount_purpose=None, workdir=None, user=None, environment=None, network="none", timeout=None, note=None, ): # run_container_command records every argv/environment pair the gate # passes, so the stand-in only produces the command's effect. argv = [str(item) for item in argv] root = mount[0] if argv[0] == "sh": self.assertIn("vendor-sources.toml", " ".join(argv)) elif argv[0].endswith("ultimate-tool-build-env"): (root / ".verify-build-env.json").write_text( json.dumps(self.DESCRIPTOR), encoding="utf-8" ) elif argv == list(verify.BUILD_COMMAND): target = root / "target" / "release" target.mkdir(parents=True, exist_ok=True) for name, content in binaries.items(): (target / name).write_bytes(content) return verify.CommandRecord(argv=argv, exit_code=0, duration_ms=1)
self.container_handler = fake_docker_run (self.root / "work").mkdir() args = verify.build_parser().parse_args( [ "--mode", "source-rebuild", "--release-dir", str(release), "--out", str(self.root / "evidence.json"), "--work-dir", str(self.root / "work"), "--checkout-root", str(self.synthetic_checkout()), ] ) # Every Docker seam is stubbed in setUp, and a real invocation is refused # outright, so this test cannot consult a daemon even by accident. cleanup = verify.Cleanup() document, code, failures = verify.run_source_rebuild(args, verify.Journal(), cleanup) # The registry also runs at exit, which would be after the stubs are gone # and would ask the real daemon about this label. Dispose of it here, # while the stubbed seams are still in place. cleanup.run()
self.assertEqual(failures, []) self.assertEqual(code, verify.EXIT_OK) self.assertIn(".cargo/config.toml", " ".join(calls[0])) self.assertEqual( calls[1], ["/usr/local/bin/ultimate-tool-build-env", "/work/.verify-build-env.json", "--replace"], ) self.assertEqual(calls[2], list(verify.BUILD_COMMAND)) # The job count travels in the environment, so the recorded command stays # exactly the documented command with no -j flag. self.assertEqual( [env.get("CARGO_BUILD_JOBS") for env in environments if "CARGO_BUILD_JOBS" in env], ["2"] ) self.assertNotIn("-j", calls[2]) self.assertEqual(document["environment"]["build_jobs"], 2) self.assertEqual(document["verified_outputs"], list(verify.BINARY_NAMES)) self.assertIsNone(document["failure"]) self.assertEqual(document["record_kind"], "release_rebuild_evidence") self.assertEqual(document["commands"][0]["argv"], list(verify.BUILD_COMMAND)) self.assertTrue(document["commands"][0]["offline"]) self.assertTrue(document["comparison"]["matches_packaged"]) self.assertEqual(document["extraction"]["extra_members"], []) self.assertIn(f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", document["extraction"]["generated_members"]) self.assertEqual(document["source_archive"]["sha256"], sha256(source_archive.read_bytes())) self.assertEqual(document["source_aggregate_sha256"], manifest["aggregate_sha256"]) self.assertIn("not an independently re-derived", document["environment"]["base_image_digest_provenance"])
record = self.root / "evidence.json" record.write_text(json.dumps(document), encoding="utf-8") accepted = packaging.validate_rebuild_evidence( record, source_archive={ "name": source_archive.name, "sha256": sha256(source_archive.read_bytes()), "aggregate_sha256": manifest["aggregate_sha256"], }, binaries={ name: {"sha256": sha256(content), "size": len(content)} for name, content in binaries.items() }, manifest=json.loads((release / "release-manifest.json").read_text(encoding="utf-8")), ) self.assertEqual(accepted["status"], "verified")
def test_a_failed_build_writes_an_investigable_failure_record(self) -> None: release, source_archive, manifest, binaries = self.build_release()
def fake_docker_run( *, image, argv, journal, mount=None, mount_purpose=None, workdir=None, user=None, environment=None, network="none", timeout=None, note=None, ): argv = [str(item) for item in argv] root = mount[0] exit_code, stdout, stderr = 0, "", "" if argv[0] == "sh": pass elif argv[0].endswith("ultimate-tool-build-env"): (root / ".verify-build-env.json").write_text( json.dumps(self.DESCRIPTOR), encoding="utf-8" ) else: # The documented build command fails and leaves no binaries. exit_code = 55 stdout = " Compiling ultimate-tool-core v0.2.0-preview.1\n" stderr = "error[E0432]: unresolved import `ultimate_tool_c64`\n" return journal.record( verify.CommandRecord( argv=argv, exit_code=exit_code, duration_ms=7, stdout=stdout, stderr=stderr, note=note, ) )
self.container_handler = fake_docker_run (self.root / "work").mkdir() args = verify.build_parser().parse_args( [ "--mode", "source-rebuild", "--release-dir", str(release), "--out", str(self.root / "failure-evidence.json"), "--work-dir", str(self.root / "work"), "--checkout-root", str(self.synthetic_checkout()), ] ) cleanup = verify.Cleanup() # The gate and the cleanup it registers both go through stubbed seams, so # the label-carrying containers are never enumerated on a real daemon. document, code, failures = verify.run_source_rebuild(args, verify.Journal(), cleanup)
self.assertEqual(code, verify.EXIT_VERIFICATION_FAILED) self.assertEqual(document["verdict"], "not_verified") self.assertEqual(document["outputs"], []) self.assertEqual(document["verified_outputs"], "none")
failure = document["failure"] self.assertEqual(failure["rebuilt_exit_code"], 55) self.assertEqual(failure["command"], list(verify.BUILD_COMMAND)) self.assertIn("unresolved import", failure["stderr_tail"]) self.assertIn("Compiling", failure["stdout_tail"]) self.assertEqual(sorted(failure["missing_outputs"]), sorted(verify.BINARY_NAMES)) self.assertEqual(failure["build_environment"]["CARGO_BUILD_JOBS"], "2") self.assertEqual(failure["source_archive"]["sha256"], sha256(source_archive.read_bytes()))
# The failure names the exit code and the missing binary rather than being # an opaque generic error. joined = " ".join(failures) self.assertIn("exited 55", joined) self.assertIn("produced no ultimate-tool", joined) self.assertEqual(document["commands"][0]["argv"], list(verify.BUILD_COMMAND))
# Cleanup honours the veto, so the failed outputs survive for inspection. cleanup.run() preserved = Path(failure["preserved_scratch"]) self.assertTrue(preserved.is_dir(), preserved) self.assertTrue((preserved / "extracted").is_dir())
# The record is written, is checksummed, and the packager refuses it, so a # failure can never be mistaken for a pass. record = self.root / "failure-evidence.json" written = verify.write_new_file( record, (json.dumps(document, indent=2, sort_keys=True) + "\n").encode("utf-8") ) self.assertTrue(written["sha256"]) with self.assertRaises(PACKAGER_FAILURE): packaging.validate_rebuild_evidence( record, source_archive={ "name": source_archive.name, "sha256": sha256(source_archive.read_bytes()), "aggregate_sha256": manifest["aggregate_sha256"], }, binaries={ name: {"sha256": sha256(content), "size": len(content)} for name, content in binaries.items() }, manifest=json.loads((release / "release-manifest.json").read_text(encoding="utf-8")), )
def test_the_cli_persists_the_failure_record_and_exits_one(self) -> None: release, _, _, _ = self.build_release()
def fake_docker_run( *, image, argv, journal, mount=None, mount_purpose=None, workdir=None, user=None, environment=None, network="none", timeout=None, note=None, ): argv = [str(item) for item in argv] exit_code, stderr = 0, "" if argv[0] == "sh": pass elif argv[0].endswith("ultimate-tool-build-env"): (mount[0] / ".verify-build-env.json").write_text( json.dumps(self.DESCRIPTOR), encoding="utf-8" ) else: exit_code = 55 stderr = "error: linking with `cc` failed\nexit status: 55\n" return journal.record( verify.CommandRecord( argv=argv, exit_code=exit_code, duration_ms=3, stderr=stderr, note=note, ) )
self.container_handler = fake_docker_run (self.root / "work2").mkdir() out = self.root / "main-failure.json" # The CLI path reaches the daemon only through the stubbed seams, including # the container cleanup it registers before the gate runs. code = verify.main( [ "--mode", "source-rebuild", "--release-dir", str(release), "--work-dir", str(self.root / "work2"), "--out", str(out), "--checkout-root", str(self.synthetic_checkout()), ] )
self.assertEqual(code, verify.EXIT_VERIFICATION_FAILED) self.assertTrue(out.is_file(), "the CLI wrote no failure record") document = json.loads(out.read_text(encoding="utf-8")) self.assertEqual(document["verdict"], "not_verified") self.assertEqual(document["failure"]["rebuilt_exit_code"], 55) self.assertIn("linking with `cc` failed", document["failure"]["stderr_tail"]) self.assertEqual(document["outputs"], [])
def load_driver(): """Exec the embedded runtime driver so its helpers can be tested directly.""" namespace = {"__name__": "runtime_driver_under_test"} exec(compile(verify.RUNTIME_DRIVER_SOURCE, "<runtime_driver>", "exec"), namespace) return namespace
class DriverHttpBoundsTests(unittest.TestCase): """The driver's HTTP reader under injected tiny, huge, error and flood bodies."""
@classmethod def setUpClass(cls) -> None: cls.driver = load_driver()
class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *args) -> None: # keep the test output quiet pass
def _send(self, status: int, payload: bytes) -> None: self.send_response(status) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", str(len(payload))) self.end_headers() try: self.wfile.write(payload) except (BrokenPipeError, ConnectionResetError, ValueError, OSError): # The client stopped reading at its cap, which is the # behaviour under test. pass
def do_GET(self) -> None: if self.path == "/tiny": self._send(200, b"ok\n") elif self.path == "/json": self._send(200, json.dumps({"schema_version": 13, "runs": []}).encode("utf-8")) elif self.path == "/huge": self._send(200, b"x" * (4 << 20)) elif self.path == "/flood": self._send(200, b"".join(b"line\n" for _ in range(600_000))) elif self.path == "/bightml": self._send(200, b"<html>\nUltimate Tool dashboard\n" + b"<!-- padding -->\n" * 200_000) elif self.path == "/padjson": self._send(200, json.dumps({"schema_version": 13, "runs": []}).encode("utf-8") + b" " * 200) elif self.path == "/bigpadjson": self._send(200, json.dumps({"schema_version": 13}).encode("utf-8") + b" " * 3_000_000) elif self.path == "/endless": self.send_response(200) self.send_header("Content-Type", "text/plain") self.end_headers() chunk = b"line\n" * 8192 try: for _ in range(4096): self.wfile.write(chunk) self.wfile.flush() except (BrokenPipeError, ConnectionResetError, ValueError, OSError): pass return else: self._send(500, b'{"error": {"code": "boom"}}')
cls.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) cls.url = f"http://127.0.0.1:{cls.server.server_address[1]}" cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) cls.thread.start()
@classmethod def tearDownClass(cls) -> None: cls.server.shutdown() cls.server.server_close() cls.thread.join(timeout=10)
def test_a_tiny_response_is_returned_intact(self) -> None: result = self.driver["http"]("GET", self.url + "/tiny") self.assertEqual(result["status"], 200) self.assertEqual(result["body"], "ok\n") self.assertFalse(result["body_overflow"]) self.assertEqual(result["body_excerpt_state"], "complete")
def test_a_json_response_parses(self) -> None: result = self.driver["http"]("GET", self.url + "/json", parse=True) self.assertEqual(result["json"]["schema_version"], 13)
def test_a_json_response_with_trailing_whitespace_parses(self) -> None: result = self.driver["http"]("GET", self.url + "/padjson", parse=True) self.assertFalse(result["body_overflow"]) self.assertEqual(result["json"]["schema_version"], 13)
def test_a_huge_response_stops_at_the_cap_and_is_flagged(self) -> None: result = self.driver["http"]("GET", self.url + "/huge") self.assertEqual(result["status"], 200) self.assertLessEqual(result["body_bytes"], self.driver["MAX_HTTP_BODY"]) self.assertTrue(result["body_overflow"]) self.assertEqual(result["body_excerpt_state"], "partial") self.assertIn("partial", result["body"]) # Nothing may be parsed from a body that overflowed. self.assertIsNone(self.driver["http"]("GET", self.url + "/huge", parse=True)["json"])
def test_a_flooding_response_is_capped(self) -> None: result = self.driver["http"]("GET", self.url + "/flood") self.assertLessEqual(result["body_bytes"], self.driver["MAX_HTTP_BODY"]) self.assertTrue(result["body_overflow"])
def test_an_endless_response_is_abandoned_at_the_cap(self) -> None: result = self.driver["http"]("GET", self.url + "/endless", limit=4096, timeout=10) self.assertTrue(result["body_overflow"]) self.assertLessEqual(result["body_bytes"], 4096) self.assertEqual(result["body_excerpt_state"], "partial")
def test_a_large_html_document_keeps_its_prefix_marker(self) -> None: # The excerpt is a prefix, so a marker at the start is still observable # while the record says the body was only partially read. result = self.driver["http"]("GET", self.url + "/bightml", limit=4096) self.assertTrue(result["body_overflow"]) self.assertIn("<html", result["body"].lower()) self.assertEqual(result["body_excerpt_state"], "partial")
def test_a_large_padded_json_document_is_not_parsed(self) -> None: result = self.driver["http"]("GET", self.url + "/bigpadjson", parse=True, limit=4096) self.assertTrue(result["body_overflow"]) self.assertIsNone(result["json"])
def test_an_error_response_is_captured_with_its_status(self) -> None: result = self.driver["http"]("GET", self.url + "/missing") self.assertEqual(result["status"], 500) self.assertIn("boom", result["body"])
def test_an_over_cap_response_becomes_a_gate_failure(self) -> None: failures = self.driver["overflow_failures"]( [ {"name": "get/api/v1/runs", "body_overflow": True}, {"name": "get/api/v1/health", "body_overflow": False}, ] ) self.assertEqual(len(failures), 1) self.assertIn("get/api/v1/runs", failures[0]) self.assertEqual(self.driver["overflow_failures"]([{"name": "x", "body_overflow": False}]), [])
def test_the_readiness_window_is_rolling_not_unbounded(self) -> None: source = verify.RUNTIME_DRIVER_SOURCE self.assertEqual(source.count("MAX_READINESS_WINDOW"), 3) self.assertNotIn("buffer += chunk", source) self.assertIn("[-MAX_READINESS_WINDOW:]", source)
class ShimJournalBoundsTests(TemporaryDirectoryCase): def test_a_many_line_journal_is_line_capped(self) -> None: driver = load_driver() path = self.root / "attempts.jsonl" path.write_text( "".join('{"tool":"git","args":"","at":"now"}\n' for _ in range(5000)), encoding="utf-8" ) result = driver["read_shim_attempts"](path, max_bytes=1 << 20, max_lines=100) self.assertEqual(result["lines"], 100) self.assertTrue(result["truncated"]) self.assertEqual(set(result["tools"]), {"git"})
def test_a_huge_journal_is_byte_capped(self) -> None: driver = load_driver() path = self.root / "attempts.jsonl" path.write_text('{"tool":"git","args":"' + "x" * 5000 + '"}\n', encoding="utf-8") result = driver["read_shim_attempts"](path, max_bytes=100, max_lines=10) self.assertTrue(result["truncated"]) self.assertLessEqual(result["bytes_read"], 100)
def test_only_optional_probe_attempts_produce_no_failure(self) -> None: driver = load_driver() path = self.root / "attempts.jsonl" path.write_text( "".join('{"tool":"git","args":"","at":"now"}\n' for _ in range(400)), encoding="utf-8" ) attempts = driver["read_shim_attempts"](path) required, optional, failures = driver["shim_journal_failures"](attempts) self.assertEqual(required, []) self.assertEqual(optional, ["git"]) self.assertEqual(failures, [])
def test_a_required_tool_beyond_the_read_bound_cannot_be_ruled_out(self) -> None: # Optional probes fill the journal, so a required tool sits past the cap. # The read is truncated and "no required tool was touched" must not be # claimed from a partial read. driver = load_driver() path = self.root / "attempts.jsonl" lines = ['{"tool":"git","args":"","at":"now"}\n'] * 300 lines.append('{"tool":"cargo","args":"--version","at":"now"}\n') path.write_text("".join(lines), encoding="utf-8") attempts = driver["read_shim_attempts"](path, max_bytes=1 << 20, max_lines=100) self.assertTrue(attempts["truncated"]) required, optional, failures = driver["shim_journal_failures"](attempts) self.assertEqual(required, []) self.assertEqual(optional, ["git"]) self.assertEqual(len(failures), 1) self.assertIn("cannot be shown that no required tool was touched", failures[0])
def test_a_required_attempt_inside_the_bound_is_a_failure(self) -> None: driver = load_driver() path = self.root / "attempts.jsonl" path.write_text('{"tool":"cargo","args":"","at":"now"}\n', encoding="utf-8") attempts = driver["read_shim_attempts"](path) required, _, failures = driver["shim_journal_failures"](attempts) self.assertEqual(required, ["cargo"]) self.assertEqual(len(failures), 1) self.assertIn("must not need", failures[0])
def test_a_symlinked_journal_is_not_followed(self) -> None: driver = load_driver() target = self.root / "real.jsonl" target.write_text('{"tool":"cargo","args":"","at":"now"}\n', encoding="utf-8") link = self.root / "link.jsonl" link.symlink_to(target) result = driver["read_shim_attempts"](link) self.assertFalse(result["present"])
def test_an_absent_journal_is_reported_as_absent(self) -> None: driver = load_driver() result = driver["read_shim_attempts"](self.root / "nope.jsonl") self.assertFalse(result["present"]) self.assertEqual(result["lines"], 0) self.assertFalse(result["truncated"])
class CleanupConditionTests(unittest.TestCase): def test_a_vetoed_cleanup_action_is_skipped_and_the_rest_still_run(self) -> None: calls: list[str] = [] keep = {"yes": True} cleanup = verify.Cleanup() cleanup.add("remove scratch", lambda: calls.append("removed"), condition=lambda: not keep["yes"]) cleanup.add("remove containers", lambda: calls.append("containers")) cleanup.run() self.assertEqual(calls, ["containers"])
def test_an_unconditional_cleanup_action_always_runs(self) -> None: calls: list[str] = [] cleanup = verify.Cleanup() cleanup.add("remove scratch", lambda: calls.append("removed")) cleanup.run() self.assertEqual(calls, ["removed"])
class DockerImageIdentityTests(unittest.TestCase): """The image actually used is identified, not only named by its recipe tag."""
def test_the_image_identity_is_recorded_from_the_daemon(self) -> None: with mock.patch.object(subprocess, "run") as runner: runner.return_value = mock.Mock(returncode=0, stdout=b"sha256:8325d5e8\n", stderr=b"") identity = verify.docker_image_identity("ultimate-tool-release-env") self.assertEqual(identity["id"], "sha256:8325d5e8") self.assertEqual(identity["tag"], "ultimate-tool-release-env") self.assertIn("provenance", identity)
def test_an_uninspectable_image_is_reported_not_hidden(self) -> None: with mock.patch.object(subprocess, "run") as runner: runner.return_value = mock.Mock(returncode=1, stdout=b"", stderr=b"no such image") identity = verify.docker_image_identity("missing") self.assertIsNone(identity["id"]) self.assertIn("no such image", identity["error"])
@unittest.skipUnless(DOCKER_GATE, "set ULTIMATE_TOOL_VERIFY_LOCAL_DOCKER=1 to run the container gates")class DockerGateTests(TemporaryDirectoryCase): """The two container gates, opt-in because they build and run images."""
def test_docker_is_available_before_claiming_anything(self) -> None: available, detail = verify.docker_available() self.assertTrue(available, detail)
def test_the_pinned_release_environment_image_is_present(self) -> None: self.assertTrue(verify.docker_image_present(verify.RELEASE_ENV_IMAGE))
def prebuilt_binaries() -> dict[str, Path] | None: """An already-built target directory, so this test never compiles anything.""" for directory in ("target/release", "target/debug"): candidates = {name: ROOT / directory / name for name in verify.BINARY_NAMES} if all(path.is_file() for path in candidates.values()): return candidates return None
@unittest.skipUnless(DOCKER_GATE, "set ULTIMATE_TOOL_VERIFY_LOCAL_DOCKER=1 to run the container gates")class IsolatedRuntimeGateTests(TemporaryDirectoryCase): """Run the real runtime gate against a synthetic, self-consistent package.
The binaries come from an already-built target directory, so this test never compiles anything. It proves the container flow itself works — runtime image, fail-on-invocation shims, the example's eight runs, the embedded service and its origin rules, and the read-out. It is explicitly **not** release evidence: the archive is synthesized here rather than produced by the packager, and the binaries are not the packaged ones. """
def build_synthetic_release(self, binaries: dict[str, Path]) -> Path: dist = self.root / "dist" dist.mkdir() example = ROOT / "examples" / "first-run" self.assertTrue(example.is_dir(), "the first-run example is missing")
binary_archive = dist / f"{BINARY_PREFIX}.tar.gz" with tarfile.open(binary_archive, "w:gz") as handle: for name, path in sorted(binaries.items()): info = handle.gettarinfo(str(path), arcname=f"{BINARY_PREFIX}/bin/{name}") info.mode = 0o755 with path.open("rb") as stream: handle.addfile(info, stream) for path in sorted(example.rglob("*")): if not path.is_file() or ".ultimate-tool" in path.parts: continue relative = path.relative_to(ROOT).as_posix() info = handle.gettarinfo(str(path), arcname=f"{BINARY_PREFIX}/{relative}") with path.open("rb") as stream: handle.addfile(info, stream) self.assertGreater(binary_archive.stat().st_size, 0)
source_members = {"Cargo.toml": b"[workspace]\n"} source_archive = dist / f"{SOURCE_PREFIX}.tar.gz" with tarfile.open(source_archive, "w:gz") as handle: for name, content in source_members.items(): info = tarfile.TarInfo(f"{SOURCE_PREFIX}/{name}") info.size = len(content) handle.addfile(info, io.BytesIO(content)) files = [ {"path": name, "sha256": sha256(content), "size": len(content)} for name, content in sorted(source_members.items()) ]
measured = { name: {"sha256": sha256(path.read_bytes()), "size": path.stat().st_size} for name, path in sorted(binaries.items()) } (dist / "release-manifest.json").write_text( json.dumps( { "schema": 1, "version": "0.2.0-preview.1", "development": "local-harness-selftest", "release_ready": False, "source_completeness": {"status": "incomplete_pending_third_party_notices"}, "artifacts": [ {"name": name, "kind": "host-binary", "path": f"bin/{name}", **record} for name, record in sorted(measured.items()) ], "source_archive": { "name": source_archive.name, "sha256": sha256(source_archive.read_bytes()), }, "archive": {"name": binary_archive.name, "sha256": sha256(binary_archive.read_bytes())}, "build_environment": { "base_image_digest": verify.PINNED_BASE_DIGEST, "rust_version": verify.PINNED_RUST_VERSION, "target_triple": verify.PINNED_TARGET_TRIPLE, }, } ) ) (dist / "source-manifest.json").write_text( json.dumps( { "schema": 1, "record_kind": "release_source_manifest_resolved", "archive": {"name": source_archive.name, "sha256": sha256(source_archive.read_bytes())}, "files": files, "aggregate_sha256": verify.aggregate_digest(files), "vendor": {"files": []}, } ) ) (dist / "SHA256SUMS").write_text( f"{sha256(binary_archive.read_bytes())} {binary_archive.name}\n" f"{sha256(source_archive.read_bytes())} {source_archive.name}\n" ) return dist
def test_the_runtime_gate_drives_the_binaries_in_isolation(self) -> None: binaries = prebuilt_binaries() if not binaries: self.skipTest("no prebuilt target/{release,debug} binaries to run") dist = self.build_synthetic_release(binaries) record = self.root / "runtime-record.json" code = verify.main( [ "--mode", "runtime", "--release-dir", str(dist), "--out", str(record), "--json", "--checkout-root", str(self.root / "unmounted" / "checkout"), ] ) self.assertTrue(record.is_file(), "the harness wrote no record") document = json.loads(record.read_text()) self.assertEqual(code, verify.EXIT_OK, document.get("failures")) self.assertEqual(document["verdict"], "verified")
observed = document["observed"] self.assertEqual(len(observed["runs"]), 8, observed["runs"]) self.assertEqual(sorted({run["outcome"] for run in observed["runs"]}), ["failed", "passed"]) self.assertEqual(len({run["run_id"] for run in observed["runs"]}), 8) self.assertEqual(len({run["artifact_sha256"] for run in observed["runs"]}), 1) for run in observed["runs"]: if run["expected_outcome"] == "passed": self.assertEqual(run["exit_code"], 0) else: self.assertEqual(run["exit_code"], 4) self.assertEqual(run["failure_code"], "crash")
crash = {entry["test"]: entry for entry in observed["crash_reports"]} self.assertEqual(crash["sdk-crash"]["state"], "resolved") self.assertEqual(crash["sdk-panic"]["unavailable_reason"], "site_not_instrumented")
service = observed["service"] self.assertTrue(service["url"].startswith("http://127.0.0.1:"), service["url"]) self.assertTrue(service["token_obtained"]) # The packaged archive carries no prior evidence, so the service must # see exactly the eight runs this gate just made through the CLI. self.assertEqual(service["runs_visible"], 8) self.assertTrue(service["terminated"], "the service was left running") statuses = {entry["name"]: entry.get("status") for entry in service["requests"]} self.assertEqual(statuses["get/api/v1/health"], 200) self.assertEqual(statuses["get-runs"], 200) self.assertEqual(statuses["get-bad-host"], 400) self.assertEqual(statuses["delete-no-token"], 403) self.assertEqual(statuses["delete-wrong-origin"], 403) self.assertEqual(statuses["delete-authorized"], 200) self.assertTrue(service["all_runs_visible"], service.get("visible_run_ids")) self.assertEqual(len(service["visible_run_ids"]), 8) for path in ("/", "/app.js", "/styles.css", "/diagnostic-report.js"): entry = next(item for item in service["requests"] if item["name"] == f"asset{path}") self.assertTrue(entry["marker_ok"], entry) self.assertGreater(entry["body_bytes"], 0)
restart = observed["service_restart"] self.assertTrue(restart["token_differs"]) self.assertTrue(restart["old_token_rejected"]) self.assertTrue(restart["new_token_accepted"]) self.assertEqual(restart["runs_visible"], 8) self.assertTrue(restart["ids_match_original"], restart.get("visible_run_ids")) self.assertTrue(restart["terminated"], "the restarted service was left running") self.assertEqual(service["seed_route_status"], 404) report_entry = next( item for item in service["requests"] if item["name"] == "get-diagnostic-report" ) self.assertEqual(report_entry["status"], 200) self.assertEqual(report_entry["report_api_schema_version"], service["api_schema_version"]) # The report's own contract schema is distinct from the outer API schema. self.assertEqual(report_entry["report_inner_schema_version"], 1) self.assertNotEqual( report_entry["report_inner_schema_version"], report_entry["report_api_schema_version"] ) self.assertTrue(report_entry["report_run_id_matches"]) self.assertTrue(report_entry["report_results_bound"])
self.assertTrue(observed["read_only"]["unchanged"], observed["read_only"]) self.assertEqual(observed["shim_attempts"]["required_absent_attempts"], 0) self.assertEqual(observed["shim_attempts"]["required_absent_tools"], []) self.assertEqual(document["shim_control"]["attempts_recorded"], 5) for tool, inspection in observed["absence"]["shims"].items(): self.assertTrue(inspection["is_shim"], f"{tool} was a real tool, not a shim") self.assertFalse(observed["absence"]["exists:/usr/local/cargo"]) self.assertFalse(observed["absence"]["exists:/var/run/docker.sock"]) self.assertFalse(document["isolation"]["checkout_mounted"]) self.assertEqual(document["runtime_environment"]["network"], "none") self.assertEqual(document["runtime_environment"]["provenance"], "harness_built_from_recipe") self.assertEqual(document["runtime_environment"]["base_reference"], verify.PINNED_BASE_REFERENCE) self.assertEqual(document["runtime_environment"]["base_digest"], verify.PINNED_BASE_DIGEST) self.assertEqual(document["runtime_environment"]["installed_by_recipe"], ["python3"])
class InvocationOwnershipTests(TemporaryDirectoryCase): """The caller may own container removal only by naming the run explicitly."""
def setUp(self) -> None: super().setUp() original = verify.INVOCATION self.addCleanup(setattr, verify, "INVOCATION", original)
def _main_without_docker(self, argv) -> int: with mock.patch.object(verify, "docker_available", return_value=(False, "no docker in this test")): return verify.main(argv)
def test_unsafe_invocation_ids_are_refused_before_anything_exists(self) -> None: base = ["--mode", "runtime", "--release-dir", str(self.root / "release")] for unsafe in ("../../etc", "UPPER", "with space", "a/b", "", "x" * 65, "-leading", "a.b"): with self.assertRaises(SystemExit) as caught: verify.main(base + ["--invocation-id", unsafe]) self.assertEqual(caught.exception.code, 2, unsafe) self.assertEqual(list(self.root.iterdir()), [], "nothing may be created for a refused id")
def test_a_safe_id_is_accepted_and_labels_the_run(self) -> None: code = self._main_without_docker([ "--mode", "runtime", "--release-dir", str(self.root / "release"), "--invocation-id", "ci-42-abcdef", ]) self.assertEqual(code, verify.EXIT_ENVIRONMENT) self.assertEqual(verify.INVOCATION, "ci-42-abcdef") self.assertEqual(verify.container_label(), f"{verify.CONTAINER_LABEL}.ci-42-abcdef")
def test_the_default_id_is_still_generated_when_none_is_given(self) -> None: code = self._main_without_docker([ "--mode", "runtime", "--release-dir", str(self.root / "release"), ]) self.assertEqual(code, verify.EXIT_ENVIRONMENT) self.assertTrue(verify.INVOCATION_ID.fullmatch(verify.INVOCATION), verify.INVOCATION)
def test_external_cleanup_requires_runtime_and_an_explicit_id(self) -> None: with self.assertRaises(SystemExit) as caught: verify.main([ "--mode", "source-rebuild", "--release-dir", str(self.root), "--invocation-id", "abc", "--external-container-cleanup", ]) self.assertEqual(caught.exception.code, 2) with self.assertRaises(SystemExit) as caught: verify.main([ "--mode", "runtime", "--release-dir", str(self.root), "--external-container-cleanup", ]) self.assertEqual(caught.exception.code, 2)
def test_external_cleanup_skips_container_removal_and_keeps_scratch_cleanup(self) -> None: for external, expected_removals in ((True, 0), (False, 1)): token = f"ci-{int(external)}-abcdef" argv = [ "--mode", "runtime", "--release-dir", str(self.root), "--work-dir", str(self.root), "--invocation-id", token, "--checkout-root", str(self.root / "unmounted" / "checkout"), ] if external: argv.append("--external-container-cleanup") args = verify.build_parser().parse_args(argv) verify.INVOCATION = verify.validate_invocation_id(token) cleanup, journal = verify.Cleanup(), verify.Journal() with mock.patch.object(verify, "docker_available", return_value=(True, "test")), \ mock.patch.object(verify, "verify_release_directory", return_value={"binary_archive": {"name": "a.tar"}, "problems": []}), \ mock.patch.object(verify, "remove_containers_with_label", return_value=[]) as remover, \ mock.patch.object(verify, "extract_verified_archive", side_effect=verify.HarnessError("stop here")): with self.assertRaises(verify.HarnessError): verify.run_runtime(args, journal, cleanup) # The registry runs at exit as well, which would fire the real label # cleanup once these stubs are gone; it is disposed of here instead. atexit.unregister(cleanup.run) self.assertEqual(remover.call_count, expected_removals, external) registered = " ".join(cleanup.as_json()) self.assertIn("scratch directory", registered, "scratch cleanup is always kept") if external: self.assertNotIn("containers labelled", registered) self.assertIn("--external-container-cleanup", " ".join(journal.notes)) else: self.assertIn("containers labelled", registered)
if __name__ == "__main__": unittest.main()