Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
117 kB · 2427 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428#!/usr/bin/env python3"""Tests for the local release packaging scaffolding.
These tests are pure, standard-library ``unittest``. They build synthetic ELF64x86-64 binaries and a synthetic repository in temporary directories, drive thereal command-line entry points of ``tools/package_release.py`` and``tools/check_release.py``, and assert the documented behaviour.
Because every artifact here is synthesised, a passing run proves the packagingschema, the ELF target checks, the allowlist and privacy enforcement, thearchive safety rules, the checksum chain and the publication gating. It doesnot prove that the pinned Ubuntu image builds these bytes, that the Rust-MOStoolchain links the demo PRGs, or that any C64 program executed. The pinnedimage is exercised only by structure checks against its Dockerfile.
A handful of tests drive a real temporary Git repository and are skipped when``git`` is unavailable."""
from __future__ import annotations
import astimport dataclassesimport gzipimport hashlibimport ioimport unittest.mockimport jsonimport osimport shutilimport structimport subprocessimport sysimport tarfileimport tempfileimport timeimport unittestfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"if str(TOOLS) not in sys.path: sys.path.insert(0, str(TOOLS))
import check_release # noqa: E402import package_release as packaging # noqa: E402
REAL_MANIFEST = TOOLS / "release_source_manifest.json"PINNED_DIGEST = "sha256:534baea6a22c03a63003dbc8dbe78fe34bc0d7e595d9a9dc9834884ff530eb55"BUILD_COMMAND = ["cargo", "build", "--locked", "--release", "-p", "ultimate-tool", "-p", "ultimate-tool-mcp"]RESERVED_VERSION = "0.2.0-preview.1"
# Constructed so this file never contains the literal sentinel text it plants.DEVICE_ID = "25" + "A6" + "0A"PRIVATE_ADDRESS = "192.168" + ".8.133"HOME_PATH = "/home/" + "maintainer" + "/projects"CREDENTIAL = "C64U_" + "PASSWORD" + ' = "' + "hunter2" + "hunter2" + '"'PLANTED_KEY = "api_" + "key" + ' = "' + "0123456789abcdef" + '"'
# ---------------------------------------------------------------------------# Synthetic ELF64 builders# ---------------------------------------------------------------------------
PT_LOAD = 1PT_INTERP = 3SHT_PROGBITS = 1SHT_STRTAB = 3SHT_DYNAMIC = 6SHT_NOTE = 7SHT_GNU_VERNEED = 0x6FFFFFFEEM_X86_64 = 62EM_AARCH64 = 183INTERPRETER = b"/lib64/ld-linux-x86-64.so.2\0"
def build_elf64( *, glibc_versions: tuple[str, ...] = ("GLIBC_2.34", "GLIBC_2.38"), dynamic: bool = True, machine: int = EM_X86_64, ei_class: int = 2, build_id: bytes = b"\x11" * 20, payload: bytes = b"ultimate-tool synthetic binary\n" * 4,) -> bytes: """Build a small but structurally real ELF64 whose version needs are known.""" dynstr = bytearray(b"\0libc.so.6\0") offsets = {} for version in glibc_versions: offsets[version] = len(dynstr) dynstr += version.encode("ascii") + b"\0" libc_offset = 1
if glibc_versions: verneed = bytearray() verneed += struct.pack("<HHIII", 1, len(glibc_versions), libc_offset, 16, 0) for index, version in enumerate(glibc_versions): next_offset = 16 if index + 1 < len(glibc_versions) else 0 verneed += struct.pack("<IHHII", 0x0BADC0DE, 0, index + 2, offsets[version], next_offset) else: verneed = bytearray()
note = struct.pack("<III", 4, len(build_id), 3) + b"GNU\0" * 1 + b"\0" * 0 note = struct.pack("<III", 4, len(build_id), 3) + b"GNU\0" + build_id note += b"\0" * ((4 - (len(note) % 4)) % 4)
dynamic = struct.pack("<QQ", 0, 0) * 4 if dynamic else b""
section_names = [b"\0", b".text\0", b".interp\0", b".gnu.version_r\0", b".dynstr\0"] if dynamic: section_names.append(b".dynamic\0") section_names += [b".note.gnu.build-id\0", b".shstrtab\0"]
shstrtab = bytearray() name_offsets = [] for name in section_names: name_offsets.append(len(shstrtab)) shstrtab += name
ehdr_size = 64 phdrs = [] body = bytearray() phnum = 2 phdr_size = phnum * 56 cursor = ehdr_size + phdr_size
def add_section(data: bytes) -> tuple[int, int]: nonlocal cursor offset = cursor body.extend(data) cursor += len(data) if cursor % 8: pad = 8 - (cursor % 8) body.extend(b"\0" * pad) cursor += pad return offset, len(data)
text_offset, text_size = add_section(payload) interp_offset, interp_size = add_section(INTERPRETER) verneed_offset, verneed_size = add_section(bytes(verneed)) dynstr_offset, dynstr_size = add_section(bytes(dynstr)) dynamic_offset = dynamic_size = None if dynamic: dynamic_offset, dynamic_size = add_section(dynamic) note_offset, note_size = add_section(note) shstrtab_offset, shstrtab_size = add_section(bytes(shstrtab))
phdrs.append(struct.pack("<IIQQQQQQ", PT_LOAD, 5, 0, 0x400000, 0x400000, cursor, cursor, 0x1000)) phdrs.append(struct.pack("<IIQQQQQQ", PT_INTERP, 4, interp_offset, interp_offset, interp_offset, interp_size, interp_size, 1))
sections = [(0, 0, 0, 0, 0, 0, 0, 0, 0, 0)]
def section( index: int, section_type: int, flags: int, offset: int, size: int, link: int = 0, info: int = 0, align: int = 1, entsize: int = 0, ) -> tuple: return (name_offsets[index], section_type, flags, 0, offset, size, link, info, align, entsize)
dynstr_index = 4 sections.append(section(1, SHT_PROGBITS, 6, text_offset, text_size, align=16)) sections.append(section(2, SHT_PROGBITS, 2, interp_offset, interp_size)) sections.append(section(3, SHT_GNU_VERNEED, 2, verneed_offset, verneed_size, link=dynstr_index, info=1, align=4)) sections.append(section(4, SHT_STRTAB, 2, dynstr_offset, dynstr_size)) if dynamic: sections.append( section(5, SHT_DYNAMIC, 3, dynamic_offset, dynamic_size, link=dynstr_index, align=8, entsize=16) ) note_name_index = 6 shstrtab_index = 7 else: note_name_index = 5 shstrtab_index = 6 sections.append(section(note_name_index, SHT_NOTE, 2, note_offset, note_size, align=4)) sections.append(section(shstrtab_index, SHT_STRTAB, 0, shstrtab_offset, shstrtab_size))
shoff = cursor shdrs = b"".join(struct.pack("<IIQQQQIIQQ", *section) for section in sections) shnum = len(sections) ehdr = struct.pack( "<4sBBBBB7xHHIQQQIHHHHHH", b"\x7fELF", ei_class, 1, 1, 0, 0, 3, machine, 1, 0, ehdr_size, shoff, 0, 64, 56, phnum, 64, shnum, shstrtab_index, ) return bytes(ehdr) + b"".join(phdrs) + bytes(body) + shdrs
def build_tar(path: Path, members: list[tuple[str, bytes]], *, symlink: str | None = None) -> None: with tarfile.open(path, "w:gz") as tar: for name, data in members: info = tarfile.TarInfo(name) info.size = len(data) info.mtime = 0 tar.addfile(info, __import__("io").BytesIO(data)) if symlink: info = tarfile.TarInfo(symlink) info.type = tarfile.SYMTYPE info.linkname = "/etc/passwd" tar.addfile(info)
# ---------------------------------------------------------------------------# Synthetic project# ---------------------------------------------------------------------------
CRATE_MANIFEST = """[package]name = "%s"version.workspace = trueedition = "2021"license = "GPL-3.0-or-later"
[[bin]]name = "%s"path = "src/main.rs""""
GPL_TEXT = b"""GNU GENERAL PUBLIC LICENSEVersion 3, 29 June 2007
Copyright (C) 2007 Free Software Foundation, Inc. <https://fsf.org/>"""
def synthetic_manifest(*, inventory_collected: bool = False) -> dict: """A compact test manifest with the same shape as the shipped one.""" inventory = [ { "component": "ultimate-tool first-party code", "scope": "both host binaries", "license": "GPL-3.0-or-later", "license_source": "LICENSE", "status": "collected", "collected_paths": ["LICENSE", "Cargo.toml"], }, { "component": "Rust crate dependencies", "scope": "linked into both host binaries", "license": "per crate", "license_source": "vendor/", "status": "collected" if inventory_collected else "conditional_on_vendor", "collected_paths": ["vendor/serde-1.0.229/LICENSE-MIT"] if inventory_collected else None, "retrieval": "cargo vendor --locked --versioned-dirs", }, { "component": "Rust standard library", "scope": "statically linked", "license": "Apache-2.0 OR MIT", "license_source": "not collected yet", "status": "collected" if inventory_collected else "pending_external", "collected_paths": ["Cargo.lock"] if inventory_collected else None, "retrieval": "collect the toolchain sources and notices", }, ] for entry in inventory: if entry.get("collected_paths") is None: entry.pop("collected_paths", None) return { "schema": 1, "record_kind": "release_source_manifest", "release": { "product": "ultimate-tool", "product_title": "Ultimate Tool", "release_type": "preview", "reserved_preview_version": RESERVED_VERSION, "license": "GPL-3.0-or-later", "license_file": "LICENSE", "binaries": [ { "name": "ultimate-tool", "kind": "host-binary", "package": "ultimate-tool", "directory": "crates/ultimate-tool-cli", "description": "cli", }, { "name": "ultimate-tool-mcp", "kind": "host-binary", "package": "ultimate-tool-mcp", "directory": "crates/ultimate-tool-mcp", "description": "mcp", }, ], "build_command": list(BUILD_COMMAND), "support_boundary": { "kind": "development-preview", "tested_platform": "Ubuntu 24.04 x86-64 (glibc 2.39)", "untested": ["other distributions"], "documented_limits": ["synthetic test boundary"], }, "runtime_requirements": [ {"name": "python3", "requirement": "Python 3", "needed_for": "the example helper", "distributed_in_archive": False} ], }, "build_environment": { "kind": "pinned-ubuntu", "dockerfile": "tools/release-env/Dockerfile", "image_tag": "ultimate-tool-release-env", "base_image": "ubuntu:24.04", "base_image_digest": PINNED_DIGEST, "base_image_digest_status": "pinned", "target_triple": "x86_64-unknown-linux-gnu", "os": "ubuntu", "os_version_id": "24.04", "glibc": "2.39", "rust_version": "1.93.0", "node_version": "22.23.2", "playwright_version": "1.64.0", "python_version": "3", "descriptor_schema": 1, "descriptor_name": "build-env.json", "descriptor_writer": "ultimate-tool-build-env", }, "source_paths": { "rules": ["explicit paths only", "hidden files excluded except crate configuration"], "include": [ {"path": "Cargo.toml", "kind": "workspace-manifest"}, {"path": "Cargo.lock", "kind": "workspace-lockfile"}, {"path": "LICENSE", "kind": "license"}, {"path": "crates/", "kind": "crate-source"}, {"path": "examples/", "kind": "example-inputs"}, {"path": "fixtures/", "kind": "fixture-source"}, {"path": "fixtures/smoke/.cargo/config.toml", "kind": "crate-config", "hidden": True}, {"path": "tools/package_release.py", "kind": "release-script"}, {"path": "tools/check_release.py", "kind": "release-script"}, {"path": "tests/", "kind": "test-source"}, {"path": "docs/RELEASING.md", "kind": "public-doc"}, ], "deny": [ {"path": "AGENTS.md", "reason": "internal"}, {"path": "README.md", "reason": "withheld until reviewed"}, {"path": "docs/INTERNAL.md", "reason": "contains acceptance evidence"}, {"path": "fixtures/*/build/", "reason": "generated outputs"}, {"path": "examples/first-run/.ultimate-tool/", "reason": "local run evidence"}, ], }, "privacy": { "scanner": "tools/package_release.py", "sentinel_policy": [ {"id": "private_key_block", "severity": "block"}, {"id": "credential_assignment", "severity": "block"}, {"id": "absolute_home_path", "severity": "block"}, {"id": "device_identity", "severity": "review_required"}, {"id": "private_network_address", "severity": "review_required"}, {"id": "private_run_evidence_reference", "severity": "allow"}, ], }, "vendor": { "required_for_complete_source": True, "directory": "vendor", "producer_command": ["cargo", "vendor", "--locked", "--versioned-dirs"], }, "package_layout": { "binary_archive_prefix": "ultimate-tool-{version}-linux-x86_64-gnu", "source_archive_prefix": "ultimate-tool-{version}-linux-x86_64-gnu-source", "development_infix": "-development-{label}", "archive_suffix": ".tar.gz", "generated_files": [{"name": "README.md", "role": "generated"}, {"name": "INSTALL.md", "role": "generated"}], "shipped_docs": ["docs/RELEASING.md"], "shipped_examples": ["examples/first-run"], "license_files": ["LICENSE"], }, "checks": { "required_binary_checks": ["binary_elf_target", "glibc_baseline"], "states": ["verified", "pending", "not_verified", "not_applicable"], "release_ready_requires": "all checks verified", }, "inventory": inventory, "external_prerequisites": [{"name": "Rust-MOS", "needed_for": "the fixture PRGs", "availability": "pinned image", "note": "not pinned here"}], }
class SyntheticProject: """A small repository-shaped tree with a synthetic manifest and binaries."""
def __init__(self, root: Path, *, version: str = RESERVED_VERSION, inventory_collected: bool = False) -> None: self.root = root self.version = version self.inventory_collected = inventory_collected self.manifest = synthetic_manifest(inventory_collected=inventory_collected) self.output = root.parent / "release-out" self.binary_dir = root.parent / "binaries"
def write(self) -> None: root = self.root (root / "crates/ultimate-tool-cli/src").mkdir(parents=True) (root / "crates/ultimate-tool-mcp/src").mkdir(parents=True) (root / "examples/first-run/build").mkdir(parents=True) (root / "examples/first-run/.ultimate-tool/runs").mkdir(parents=True) (root / "fixtures/smoke/.cargo").mkdir(parents=True) (root / "fixtures/smoke/build").mkdir(parents=True) (root / "fixtures/smoke/src").mkdir(parents=True) (root / "tests/python").mkdir(parents=True) (root / "docs").mkdir(parents=True) (root / ".git").mkdir(parents=True)
(root / "Cargo.toml").write_text( "[workspace]\n" 'members = ["crates/ultimate-tool-cli", "crates/ultimate-tool-mcp"]\n' 'exclude = ["fixtures/smoke"]\n' 'resolver = "2"\n\n' "[workspace.package]\n" 'edition = "2024"\n' f'version = "{self.version}"\n' 'license = "GPL-3.0-or-later"\n', encoding="utf-8", ) (root / "Cargo.lock").write_text('version = 4\n\n[[package]]\nname = "ultimate-tool"\nversion = "0.1.0"\n', encoding="utf-8") (root / "LICENSE").write_bytes(GPL_TEXT) (root / "crates/ultimate-tool-cli/Cargo.toml").write_text( CRATE_MANIFEST % ("ultimate-tool", "ultimate-tool"), encoding="utf-8" ) (root / "crates/ultimate-tool-mcp/Cargo.toml").write_text( CRATE_MANIFEST % ("ultimate-tool-mcp", "ultimate-tool-mcp"), encoding="utf-8" ) (root / "crates/ultimate-tool-cli/src/main.rs").write_text("fn main() {}\n", encoding="utf-8") (root / "crates/ultimate-tool-mcp/src/main.rs").write_text("fn main() {}\n", encoding="utf-8") (root / "fixtures/smoke/Cargo.toml").write_text('[package]\nname = "smoke"\nversion = "0.1.0"\n', encoding="utf-8") (root / "fixtures/smoke/.cargo/config.toml").write_text('[build]\ntarget = "mos-c64-none"\n', encoding="utf-8") (root / "fixtures/smoke/src/main.rs").write_text("fn main() {}\n", encoding="utf-8") (root / "fixtures/smoke/build/ultimate-tool-smoke.prg").write_bytes(b"\x01\x08" + b"\xAA" * 32) (root / "examples/first-run/c64dev.toml").write_text('schema_version = 11\n', encoding="utf-8") (root / "examples/first-run/build/ultimate-tool-smoke.prg").write_bytes(b"\x01\x08" + b"\xBB" * 16) (root / "examples/first-run/.ultimate-tool/runs/run.json").write_text("{}\n", encoding="utf-8") (root / "tests/python/test_smoke.py").write_text("def test_smoke():\n assert True\n", encoding="utf-8") (root / "docs/RELEASING.md").write_text("# Releasing\n", encoding="utf-8") (root / "docs/INTERNAL.md").write_text("device " + DEVICE_ID + "\n", encoding="utf-8") (root / "README.md").write_text("raw development readme\n", encoding="utf-8") (root / "AGENTS.md").write_text("internal instructions\n", encoding="utf-8") (root / ".gitignore").write_text("target/\n", encoding="utf-8") (root / "tools").mkdir(exist_ok=True) shutil.copyfile(TOOLS / "package_release.py", root / "tools/package_release.py") shutil.copyfile(TOOLS / "check_release.py", root / "tools/check_release.py") (root / "tools/release_source_manifest.json").write_text( json.dumps(self.manifest, indent=2), encoding="utf-8" ) for name in ("ultimate-tool", "ultimate-tool-mcp"): (root / name).write_bytes(b"")
def write_vendor(self, directory: Path, *, crates: tuple[str, ...] = ("serde-1.0.229",)) -> Path: directory.mkdir(parents=True, exist_ok=True) for crate in crates: crate_dir = directory / crate crate_dir.mkdir(parents=True, exist_ok=True) (crate_dir / "Cargo.toml").write_text(f'[package]\nname = "{crate}"\nversion = "1.0.0"\n', encoding="utf-8") (crate_dir / "LICENSE-MIT").write_text("MIT\n", encoding="utf-8") return directory
def write_binaries(self, *, glibc_versions: tuple[str, ...] = ("GLIBC_2.34", "GLIBC_2.38")) -> Path: self.binary_dir.mkdir(parents=True, exist_ok=True) for name in ("ultimate-tool", "ultimate-tool-mcp"): (self.binary_dir / name).write_bytes(build_elf64(glibc_versions=glibc_versions)) (self.binary_dir / name).chmod(0o755) return self.binary_dir
def write_build_env(self, **overrides) -> Path: expected = self.manifest["build_environment"] descriptor = { "schema": 1, "record_kind": "build_environment", "produced_by": "/usr/local/bin/ultimate-tool-build-env in tools/release-env/Dockerfile", "observed_at": "2026-10-11T00:00:00Z", "os": {"id": "ubuntu", "version_id": "24.04"}, "glibc": {"version": "2.39"}, "target_triple": "x86_64-unknown-linux-gnu", "toolchain": { "rustc": "1.93.0 (254b59607 2026-01-19)", "cargo": "1.93.0 (083ac5135 2025-12-15)", "python": "Python 3.12.3", "node": "v22.23.3", }, "base_image": { "reference": f"{expected['base_image']}@{expected['base_image_digest']}", "digest": expected["base_image_digest"], "pinned": True, }, "build_command": list(BUILD_COMMAND), } descriptor.update(overrides) self.binary_dir.mkdir(parents=True, exist_ok=True) path = self.binary_dir / "build-env.json" path.write_text(json.dumps(descriptor, indent=2), encoding="utf-8") return path
def write_manifest_file(self, manifest: dict | None = None) -> Path: path = self.root / "tools/release_source_manifest.json" path.write_text(json.dumps(manifest or self.manifest, indent=2), encoding="utf-8") return path
def tmp_path(self, name: str) -> Path: return self.root.parent / name
def write_rebuild_evidence(self, source_manifest: dict, **overrides) -> Path: """A clean-rebuild evidence document bound to the prepared source archive.""" outputs = [] for binary in ("ultimate-tool", "ultimate-tool-mcp"): path = self.binary_dir / binary outputs.append( {"name": binary, "size": path.stat().st_size, "sha256": packaging.sha256_file(path)} ) evidence = { "schema": 1, "record_kind": "release_rebuild_evidence", "source_archive": { "name": source_manifest["archive"]["name"], "sha256": source_manifest["archive"]["sha256"], }, "source_aggregate_sha256": source_manifest["aggregate_sha256"], "environment": { "base_image_digest": self.manifest["build_environment"]["base_image_digest"], "rustc": "1.93.0 (254b59607 2026-01-19)", "target_triple": "x86_64-unknown-linux-gnu", }, "commands": [ {"argv": list(BUILD_COMMAND), "exit_code": 0, "cwd": "extracted source archive"} ], "outputs": outputs, } evidence.update(overrides) path = self.root.parent / "rebuild-evidence.json" path.write_text(json.dumps(evidence, indent=2), encoding="utf-8") return path
def development_package_with_evidence(self) -> None: """A development package whose clean-rebuild evidence is registered.""" self.write_binaries() self.write_build_env() vendor = self.write_vendor(self.root.parent / "vendor") self.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") source_manifest = json.loads((self.output / "source-manifest.json").read_text(encoding="utf-8")) evidence = self.write_rebuild_evidence(source_manifest) self.run_packager( "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(self.binary_dir), "--development", "synthetic", "--rebuild-status", "verified", "--rebuild-evidence", str(evidence), )
def run_packager(self, *arguments: str, expect: int = 0) -> subprocess.CompletedProcess: command = [ sys.executable, str(TOOLS / "package_release.py"), "--source-root", str(self.root), "--source-manifest", str(self.root / "tools/release_source_manifest.json"), "--output", str(self.output), *arguments, ] result = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False) if expect is not None and result.returncode != expect: raise AssertionError( f"packager exited {result.returncode}, expected {expect}\nstdout: {result.stdout}\nstderr: {result.stderr}" ) return result
def run_checker(self, *arguments: str, expect: int | None = None) -> subprocess.CompletedProcess: command = [ sys.executable, str(TOOLS / "check_release.py"), "--release-dir", str(self.output), "--source-root", str(self.root), *arguments, ] result = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False) if expect is not None and result.returncode != expect: raise AssertionError( f"checker exited {result.returncode}, expected {expect}\nstdout: {result.stdout}\nstderr: {result.stderr}" ) return result
def prepare(self, *extra: str, expect: int = 0) -> subprocess.CompletedProcess: vendor = self.write_vendor(self.root.parent / "vendor") return self.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", *extra, expect=expect, )
def full_package(self, *extra: str, expect: int = 0) -> subprocess.CompletedProcess: self.write_binaries() self.write_build_env() vendor = self.write_vendor(self.root.parent / "vendor") return self.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(self.binary_dir), "--development", "synthetic", *extra, expect=expect, )
class PackageTestCase(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory(prefix="ut-release-test-") self.tmp = Path(self._tmp.name) self.repo = self.tmp / "project" self.repo.mkdir() self.addCleanup(self._tmp.cleanup)
def project(self, **kwargs) -> SyntheticProject: project = SyntheticProject(self.repo, **kwargs) project.write() return project
def git(self, *arguments: str, cwd: Path | None = None) -> subprocess.CompletedProcess: return subprocess.run( ["git", *arguments], cwd=cwd or self.repo, capture_output=True, text=True, check=False, timeout=60 )
# ---------------------------------------------------------------------------# Scanner# ---------------------------------------------------------------------------
class ScannerTests(unittest.TestCase): def test_detects_each_sentinel(self) -> None: cases = { "private_key_block": b"-----BEGIN RSA " + b"PRIVATE KEY-----\nMIIE\n", "credential_assignment": CREDENTIAL.encode(), "device_identity": ("device " + DEVICE_ID + " responded").encode(), "private_network_address": ("receiver at " + PRIVATE_ADDRESS).encode(), "absolute_home_path": (HOME_PATH + "/ultimate-tool").encode(), "private_run_evidence_reference": b"root/" + b"." + b"ultimate-tool" + b"/runs/x", } for identifier, payload in cases.items(): with self.subTest(identifier=identifier): self.assertIn(identifier, packaging.scan_bytes(payload))
def test_chunked_scan_finds_split_pattern(self) -> None: payload = b"x" * (packaging.SENTINEL_CHUNK - 4) + b"-----BEGIN RSA " + b"PRIVATE KEY-----" with tempfile.NamedTemporaryFile(delete=False) as handle: handle.write(payload) path = Path(handle.name) self.addCleanup(path.unlink) self.assertIn("private_key_block", packaging.scan_file(path))
def test_scanning_large_payloads_stays_linear(self) -> None: """A one-megabyte payload must be scanned in well under a second.
An unbounded quantifier before the credential keyword once made this quadratic, which stalled the whole packaging run. """ payloads = { "unrelated_bytes": b"x" * (1 << 20), "uppercase_after_header": b"-----BEGIN " + b"A" * (1 << 20), "home_prefix": b"/home/" + b"a" * (1 << 20), "keyword_then_filler": b"password" + b"a" * (1 << 20), "underscores": b"_" * (1 << 20), "spaces_after_keyword": b"secret_key" + b" " * (1 << 20), } for name, payload in payloads.items(): start = time.monotonic() packaging.scan_bytes(payload) elapsed = time.monotonic() - start with self.subTest(payload=name): self.assertLess(elapsed, 1.0, f"{name} took {elapsed:.3f}s")
def test_scanner_does_not_flag_its_own_sources(self) -> None: for path in (TOOLS / "package_release.py", TOOLS / "check_release.py"): with self.subTest(path=path.name): self.assertEqual(packaging.scan_bytes(path.read_bytes()), [])
def test_credential_scan_ignores_short_and_non_assignments(self) -> None: self.assertEqual(packaging.scan_bytes(b"password = short\n"), []) self.assertEqual(packaging.scan_bytes(b"the token cannot cross the HTTP interface\n"), []) self.assertEqual(packaging.scan_bytes(b"192.168.x.x placeholder\n"), []) self.assertEqual(packaging.scan_bytes(b"suppaftp = { version = \"=10.0.1\" }\n"), []) # A plain identifier assignment is code, not a credential. This is the # exact shape found in the first-party runners. self.assertEqual(packaging.scan_bytes(b"let recovery_password = password.map(str::to_owned);\n"), [])
def test_suppaftp_version_is_not_a_private_address(self) -> None: self.assertNotIn("private_network_address", packaging.scan_bytes(b'version = "10.0.1"\n'))
class PatternTests(unittest.TestCase): def test_directory_pattern(self) -> None: pattern = packaging.compile_path_pattern("fixtures/*/build/") self.assertTrue(pattern.match("fixtures/smoke/build/out.prg")) self.assertTrue(pattern.match("fixtures/smoke/build")) self.assertFalse(pattern.match("fixtures/smoke/src/main.rs"))
def test_double_star_pattern(self) -> None: pattern = packaging.compile_path_pattern("docs/**/internal/*.md") self.assertTrue(pattern.match("docs/a/b/internal/x.md")) self.assertTrue(pattern.match("docs/internal/x.md")) self.assertFalse(pattern.match("docs/a/b/internal/x.txt"))
def test_exact_pattern(self) -> None: pattern = packaging.compile_path_pattern("Cargo.toml") self.assertTrue(pattern.match("Cargo.toml")) self.assertFalse(pattern.match("crates/Cargo.toml"))
# ---------------------------------------------------------------------------# ELF measurement# ---------------------------------------------------------------------------
class ElfTests(unittest.TestCase): def measure(self, data: bytes, baseline: str = "2.39") -> dict: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "ultimate-tool" path.write_bytes(data) return packaging.measure_elf(path, baseline)
def test_measures_x86_64_gnu_binary(self) -> None: record = self.measure(build_elf64()) self.assertEqual(record["class"], "ELF64") self.assertEqual(record["machine"], "x86_64") self.assertEqual(record["machine_value"], EM_X86_64) self.assertEqual(record["type"], "DYN") self.assertEqual(record["linkage"], "dynamic") self.assertEqual(record["interpreter"], "/lib64/ld-linux-x86-64.so.2") self.assertEqual(record["glibc_max_required"], "GLIBC_2.38") self.assertTrue(record["within_baseline"]) self.assertEqual(record["build_id"], "11" * 20)
def test_rejects_glibc_newer_than_baseline(self) -> None: with self.assertRaises(packaging.CheckFailure) as caught: self.measure(build_elf64(glibc_versions=("GLIBC_2.39", "GLIBC_2.41"))) self.assertIn("GLIBC_2.41", str(caught.exception))
def test_rejects_wrong_machine(self) -> None: with self.assertRaises(packaging.CheckFailure) as caught: self.measure(build_elf64(machine=EM_AARCH64)) self.assertIn("x86_64", str(caught.exception))
def test_rejects_32_bit_class(self) -> None: with self.assertRaises(packaging.CheckFailure) as caught: self.measure(build_elf64(ei_class=1)) self.assertIn("not ELF64", str(caught.exception))
def test_rejects_non_elf(self) -> None: with self.assertRaises(packaging.CheckFailure): self.measure(b"#!/bin/sh\necho hi\n")
def test_static_binary_without_version_needs(self) -> None: record = self.measure(build_elf64(dynamic=False, glibc_versions=())) self.assertEqual(record["linkage"], "static") self.assertIsNone(record["glibc_max_required"]) self.assertTrue(record["within_baseline"])
def test_dynamic_without_interpreter_is_rejected(self) -> None: data = bytearray(build_elf64()) data[120:124] = struct.pack("<I", PT_LOAD) with self.assertRaises(packaging.CheckFailure) as caught: self.measure(bytes(data)) self.assertIn("interpreter", str(caught.exception))
# ---------------------------------------------------------------------------# Archive writing and safety# ---------------------------------------------------------------------------
class ArchiveTests(PackageTestCase): def test_deterministic_archive_bytes(self) -> None: source = self.tmp / "payload.txt" source.write_text("hello\n", encoding="utf-8") first = self.tmp / "first.tar.gz" second = self.tmp / "second.tar.gz" members = [("pkg/README.md", b"# hi\n", 0o644), ("pkg/payload.txt", source, 0o644)] packaging.write_archive(first, members) packaging.write_archive(second, members) self.assertEqual(packaging.sha256_file(first), packaging.sha256_file(second))
def test_audit_reports_safe_archive(self) -> None: target = self.tmp / "safe.tar.gz" packaging.write_archive(target, [("pkg/a", b"a", 0o644)]) audit = packaging.audit_tar(target) self.assertEqual(audit["tar_safety"], "verified") self.assertEqual(audit["members"], 1)
def test_audit_rejects_symlink_member(self) -> None: target = self.tmp / "unsafe.tar.gz" build_tar(target, [("pkg/a", b"a")], symlink="pkg/link") with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target) self.assertIn("link member", str(caught.exception))
def test_audit_rejects_traversal_member(self) -> None: target = self.tmp / "traversal.tar.gz" build_tar(target, [("../escape", b"a")]) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target) self.assertIn("parent-directory segment", str(caught.exception))
def test_audit_rejects_terminal_parent_segment(self) -> None: target = self.tmp / "terminal.tar.gz" build_tar(target, [("pkg/..", b"a")]) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target) self.assertIn("parent-directory segment", str(caught.exception))
def test_audit_rejects_duplicate_members(self) -> None: target = self.tmp / "duplicate.tar.gz" build_tar(target, [("pkg/a", b"one"), ("pkg/a", b"two")]) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target) self.assertIn("duplicate member names", str(caught.exception))
def test_audit_enforces_injected_member_limit(self) -> None: target = self.tmp / "many.tar.gz" build_tar(target, [("pkg/a", b"a"), ("pkg/b", b"b"), ("pkg/c", b"c")]) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target, packaging.Limits(max_members=2)) self.assertIn("more than 2 members", str(caught.exception))
def test_audit_enforces_injected_size_limits(self) -> None: target = self.tmp / "big.tar.gz" build_tar(target, [("pkg/a", b"x" * 1024)]) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target, packaging.Limits(max_member_bytes=16)) self.assertIn("above the 16 byte member limit", str(caught.exception)) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target, packaging.Limits(max_archive_bytes=8)) self.assertIn("above the 8 byte limit", str(caught.exception))
def test_write_archive_refuses_unsafe_member_name(self) -> None: target = self.tmp / "out.tar.gz" with self.assertRaises(packaging.CheckFailure) as caught: packaging.write_archive(target, [("../escape", b"a", 0o644)]) self.assertIn("parent-directory segment", str(caught.exception)) self.assertFalse(target.exists())
def test_write_archive_refuses_injected_member_limit(self) -> None: target = self.tmp / "out.tar.gz" with self.assertRaises(packaging.CheckFailure) as caught: packaging.write_archive( target, [("pkg/a", b"a", 0o644), ("pkg/b", b"b", 0o644)], packaging.Limits(max_members=1) ) self.assertIn("above the limit", str(caught.exception)) self.assertFalse(target.exists())
# ---------------------------------------------------------------------------# Allowlist and privacy enforcement# ---------------------------------------------------------------------------
class AllowlistTests(PackageTestCase): def test_selection_respects_deny_and_hidden_rules(self) -> None: project = self.project() files, report = packaging.collect_source_files(project.root, project.manifest) paths = {entry["path"] for entry in files} self.assertIn("Cargo.toml", paths) self.assertIn("fixtures/smoke/.cargo/config.toml", paths) self.assertNotIn("AGENTS.md", paths) self.assertNotIn("README.md", paths) self.assertNotIn("docs/INTERNAL.md", paths) self.assertNotIn(".gitignore", paths) self.assertFalse(any(path.startswith("fixtures/smoke/build/") for path in paths)) self.assertFalse(any(".ultimate-tool" in path for path in paths)) self.assertGreater(report["denied"], 0) self.assertGreater(report["hidden"], 0)
def test_hidden_entry_requires_crate_config_kind(self) -> None: project = self.project() project.manifest["source_paths"]["include"].append( {"path": "fixtures/smoke/.hidden.toml", "kind": "fixture-source", "hidden": True} ) (project.root / "fixtures/smoke/.hidden.toml").write_text("x\n", encoding="utf-8") with self.assertRaises(packaging.CheckFailure) as caught: packaging.collect_source_files(project.root, project.manifest) self.assertIn("cargo configuration", str(caught.exception))
def test_missing_allowlisted_path_is_an_error(self) -> None: project = self.project() (project.root / "Cargo.lock").unlink() with self.assertRaises(packaging.CheckFailure) as caught: packaging.collect_source_files(project.root, project.manifest) self.assertIn("missing from the source root", str(caught.exception))
def test_duplicate_include_entry_is_rejected(self) -> None: project = self.project() project.manifest["source_paths"]["include"].append({"path": "Cargo.toml", "kind": "workspace-manifest"}) with self.assertRaises(packaging.CheckFailure) as caught: packaging.collect_source_files(project.root, project.manifest) self.assertIn("duplicate include entry", str(caught.exception))
def test_symlink_in_allowlist_is_rejected(self) -> None: project = self.project() link = project.root / "crates/ultimate-tool-cli/src/link.rs" link.symlink_to(project.root / "Cargo.toml") with self.assertRaises(packaging.CheckFailure) as caught: packaging.collect_source_files(project.root, project.manifest) self.assertIn("symlink", str(caught.exception))
def test_workspace_version_disagreement_is_rejected(self) -> None: project = self.project() (project.root / "crates/ultimate-tool-mcp/Cargo.toml").write_text( '[package]\nname = "ultimate-tool-mcp"\nversion = "0.1.0"\n', encoding="utf-8" ) with self.assertRaises(packaging.CheckFailure) as caught: packaging.read_workspace_version(project.root) self.assertIn("versioned together", str(caught.exception))
class PrivacyTests(PackageTestCase): def test_blocking_secret_is_never_packaged(self) -> None: project = self.project() (project.root / "crates/ultimate-tool-cli/src/main.rs").write_text( CREDENTIAL + "\n", encoding="utf-8" ) result = project.prepare(expect=1) self.assertIn("privacy sentinel policy blocked packaging", result.stderr) self.assertIn("crates/ultimate-tool-cli/src/main.rs", result.stderr) archive = ( project.output / f"ultimate-tool-{RESERVED_VERSION}-linux-x86_64-gnu-development-synthetic-source.tar.gz" ) self.assertFalse(archive.exists())
def test_private_key_is_never_packaged(self) -> None: project = self.project() (project.root / "docs/RELEASING.md").write_text( "-----BEGIN OPENSSH " + "PRIVATE KEY-----\n", encoding="utf-8" ) result = project.prepare(expect=1) self.assertIn("private_key_block", result.stderr)
def test_absolute_home_path_is_never_packaged(self) -> None: project = self.project() (project.root / "tests/python/test_smoke.py").write_text( f'PATH = "{HOME_PATH}/ultimate-tool"\n', encoding="utf-8" ) result = project.prepare(expect=1) self.assertIn("absolute_home_path", result.stderr)
def test_device_identifier_is_recorded_for_review(self) -> None: project = self.project() (project.root / "examples/first-run/c64dev.toml").write_text( f'schema_version = 11\n# {"device " + DEVICE_ID}\n', encoding="utf-8" ) project.prepare() manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) findings = manifest["privacy"]["findings"] self.assertEqual(len(findings), 1) self.assertEqual(findings[0]["sentinel"], "device_identity") self.assertEqual(findings[0]["severity"], "review_required") self.assertEqual(findings[0]["path"], "examples/first-run/c64dev.toml") self.assertEqual(manifest["privacy"]["blocking_findings"], 0)
def test_local_evidence_directory_never_enters_the_archive(self) -> None: project = self.project() project.full_package() archive = project.output / ( "ultimate-tool-" + RESERVED_VERSION + "-linux-x86_64-gnu-development-synthetic.tar.gz" ) with tarfile.open(archive, "r:gz") as tar: names = tar.getnames() self.assertFalse(any(".ultimate-tool" in name for name in names)) self.assertFalse(any(".git" in name.split("/") for name in names))
# ---------------------------------------------------------------------------# Packaging end to end# ---------------------------------------------------------------------------
class PrepareSourceTests(PackageTestCase): def test_source_archive_and_manifest(self) -> None: project = self.project() project.prepare() archive = project.output / ( "ultimate-tool-" + RESERVED_VERSION + "-linux-x86_64-gnu-development-synthetic-source.tar.gz" ) self.assertTrue(archive.exists()) manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["record_kind"], "release_source_manifest_resolved") self.assertEqual(manifest["archive"]["sha256"], packaging.sha256_file(archive)) self.assertEqual(manifest["vendor"]["status"], "provided") self.assertEqual(manifest["vendor"]["file_count"], 2) self.assertEqual(manifest["source_completeness"]["status"], "incomplete_pending_third_party_notices") prefix = manifest["archive"]["prefix"] with tarfile.open(archive, "r:gz") as tar: names = set(tar.getnames()) for entry in manifest["files"]: member = f"{prefix}/{entry['path']}" self.assertIn(member, names) handle = tar.extractfile(member) self.assertEqual(hashlib.sha256(handle.read()).hexdigest(), entry["sha256"]) provenance = json.loads(tar.extractfile(f"{prefix}/SOURCE-PROVENANCE.json").read()) self.assertEqual(provenance["record_kind"], "release_source_provenance") self.assertIsNone(provenance["archive"]["sha256"] if "archive" in provenance else None) self.assertNotIn(HOME_PATH, json.dumps(provenance))
def test_missing_vendor_marks_source_incomplete(self) -> None: project = self.project() project.run_packager("--prepare-source", "--no-vendor", "--development", "synthetic") manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["source_completeness"]["status"], "incomplete_missing_vendored_crate_sources")
def test_dirty_tree_without_development_label_is_refused(self) -> None: project = self.project() project.root.joinpath("untracked.txt").write_text("x\n", encoding="utf-8") (project.root / ".git").rmdir() result = project.run_packager("--prepare-source", "--vendor-dir", str(project.write_vendor(project.root.parent / "vendor")), expect=1) self.assertIn("--development", result.stderr)
def test_vendor_directory_is_copied_with_license_files(self) -> None: project = self.project() project.prepare() archive = project.output / ( "ultimate-tool-" + RESERVED_VERSION + "-linux-x86_64-gnu-development-synthetic-source.tar.gz" ) with tarfile.open(archive, "r:gz") as tar: names = tar.getnames() self.assertTrue(any(name.endswith("vendor/serde-1.0.229/LICENSE-MIT") for name in names)) self.assertTrue(any(name.endswith("vendor-sources.toml") for name in names))
class PackTests(PackageTestCase): def test_pack_requires_prepared_source(self) -> None: project = self.project() project.write_binaries() project.write_build_env() result = project.run_packager( "--pack", "--binary-dir", str(project.binary_dir), "--development", "synthetic", expect=1, ) self.assertIn("source-manifest.json", result.stderr)
def test_full_package_records_measured_artifacts(self) -> None: project = self.project() project.full_package() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["record_kind"], "release_manifest") self.assertEqual(manifest["schema_version"], 1) self.assertEqual(manifest["version"], RESERVED_VERSION) self.assertTrue(manifest["development"]) self.assertEqual(manifest["development_label"], "synthetic") binary_archive = project.output / manifest["archive"]["name"] self.assertEqual(manifest["archive"]["sha256"], packaging.sha256_file(binary_archive)) artifacts = {entry["name"]: entry for entry in manifest["artifacts"]} self.assertEqual(artifacts["ultimate-tool"]["sha256"], packaging.sha256_file(project.binary_dir / "ultimate-tool")) self.assertEqual(artifacts["ultimate-tool"]["elf"]["machine"], "x86_64") self.assertEqual(artifacts["ultimate-tool"]["elf"]["glibc_max_required"], "GLIBC_2.38") source_artifact = next(entry for entry in manifest["artifacts"] if entry["kind"] == "source-archive") self.assertTrue(source_artifact["always_distributed_with_binaries"]) self.assertEqual(source_artifact["sha256"], packaging.sha256_file(project.output / manifest["source_archive"]["name"])) self.assertFalse(manifest["release_ready"]) self.assertIn("development package labeled 'synthetic'", " ".join(manifest["publication"]["blocking"])) checks = {item["check"]: item for item in manifest["verification"]} self.assertEqual(checks["binary_elf_target"]["state"], "verified") self.assertEqual(checks["build_environment_observed"]["state"], "verified") self.assertEqual(checks["annotated_version_tag"]["state"], "pending") self.assertEqual(checks["annotated_version_tag"]["severity"] if "severity" in checks["annotated_version_tag"] else "pending", "pending") self.assertEqual(checks["third_party_notices_collected"]["state"], "pending") sums = (project.output / "SHA256SUMS").read_text(encoding="utf-8") self.assertIn(manifest["archive"]["name"], sums) self.assertIn(manifest["source_archive"]["name"], sums) self.assertIn("release-manifest.json", sums) with tarfile.open(binary_archive, "r:gz") as tar: names = tar.getnames() prefix = manifest["archive"]["prefix"] self.assertIn(f"{prefix}/bin/ultimate-tool", names) self.assertIn(f"{prefix}/bin/ultimate-tool-mcp", names) self.assertIn(f"{prefix}/LICENSE", names) self.assertIn(f"{prefix}/INSTALL.md", names) self.assertIn(f"{prefix}/README.md", names) self.assertIn(f"{prefix}/SOURCE-AND-LICENSES.md", names) self.assertIn(f"{prefix}/docs/RELEASING.md", names) self.assertIn(f"{prefix}/examples/first-run/c64dev.toml", names) inner = json.loads(tar.extractfile(f"{prefix}/release-manifest.json").read()) self.assertIsNone(inner["archive"]["sha256"]) self.assertIn("self_reference", inner["archive"]) install = project.output / "release-manifest.json" self.assertTrue(install.exists())
def test_binary_archive_omits_denied_example_files(self) -> None: project = self.project() project.full_package() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) with tarfile.open(project.output / manifest["archive"]["name"], "r:gz") as tar: names = tar.getnames() self.assertFalse(any(".ultimate-tool" in name for name in names))
def test_host_glibc_artifact_is_refused_even_in_development(self) -> None: project = self.project() project.write_binaries(glibc_versions=("GLIBC_2.39", "GLIBC_2.41")) project.write_build_env() vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", expect=1, ) self.assertIn("GLIBC_2.41", result.stderr)
def test_missing_binary_is_refused(self) -> None: project = self.project() project.write_binaries() project.write_build_env() (project.binary_dir / "ultimate-tool-mcp").unlink() vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", expect=1, ) self.assertIn("never compiles", result.stderr)
def test_output_directory_must_be_new(self) -> None: project = self.project() project.output.mkdir(parents=True) (project.output / "existing.txt").write_text("keep me\n", encoding="utf-8") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=2) self.assertIn("not empty", result.stderr) self.assertEqual((project.output / "existing.txt").read_text(encoding="utf-8"), "keep me\n")
def test_dry_run_writes_nothing(self) -> None: project = self.project() result = project.run_packager("--prepare-source", "--dry-run", "--development", "synthetic") self.assertIn("dry run", result.stdout) self.assertFalse(project.output.exists())
@unittest.skipUnless(shutil.which("git"), "git is required")class TaggedReleaseTests(PackageTestCase): def init_repo(self, project: SyntheticProject) -> None: (project.root / ".git").rmdir() self.git("init", "--quiet") self.git("config", "user.email", "test@example.invalid") self.git("config", "user.name", "Release Test") self.git("add", "-A") self.git("commit", "--quiet", "-m", "initial") self.git("tag", "-a", f"v{RESERVED_VERSION}", "-m", "preview")
def ready_project(self) -> SyntheticProject: project = self.project(inventory_collected=True) self.init_repo(project) project.manifest = synthetic_manifest(inventory_collected=True) project.write_manifest_file() self.git("add", "-A") self.git("commit", "--quiet", "-m", "collected inventory") self.git("tag", "-f", "-a", f"v{RESERVED_VERSION}", "-m", "preview") project.write_binaries() project.write_build_env() return project
def prepare_then_pack( self, project: SyntheticProject, tag: str, *, rebuild: bool = True, extra: tuple[str, ...] = () ) -> subprocess.CompletedProcess: """The documented two-stage flow: prepare the source, then pack the binaries.""" vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--tag", tag) source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) arguments = [ "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--tag", tag, *extra, ] if rebuild: evidence = project.write_rebuild_evidence(source_manifest) arguments += ["--rebuild-status", "verified", "--rebuild-evidence", str(evidence)] return project.run_packager(*arguments)
def test_release_ready_package_and_preflight(self) -> None: project = self.ready_project() self.prepare_then_pack(project, f"v{RESERVED_VERSION}") manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) self.assertTrue(manifest["release_ready"], manifest["publication"]["blocking"]) self.assertEqual(manifest["source"]["tree_state"], "clean_tagged") self.assertEqual(manifest["source"]["tag_type"], "annotated") self.assertEqual(manifest["source"]["tag_commit"], manifest["source"]["revision"]) self.assertEqual(manifest["source_completeness"]["status"], "verified_by_clean_rebuild") self.assertEqual(manifest["source_completeness"]["clean_rebuild"], "verified") checks = {item["check"]: item for item in manifest["verification"]} self.assertEqual(checks["mandatory_artifacts"]["state"], "verified") result = project.run_checker("--mode", "release", expect=0) self.assertIn("verdict: passed", result.stdout)
def test_rebuild_claim_without_bound_evidence_is_refused(self) -> None: project = self.ready_project() vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--tag", f"v{RESERVED_VERSION}") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) unbounded = project.write_rebuild_evidence(source_manifest, source_archive={ "name": source_manifest["archive"]["name"], "sha256": "0" * 64, }) result = project.run_packager( "--pack", "--binary-dir", str(project.binary_dir), "--tag", f"v{RESERVED_VERSION}", "--rebuild-status", "verified", "--rebuild-evidence", str(unbounded), expect=1, ) self.assertIn("is not bound to the source archive hash", result.stderr)
def test_rebuild_claim_needs_the_documented_command_and_outputs(self) -> None: project = self.ready_project() vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--tag", f"v{RESERVED_VERSION}") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) wrong_command = project.write_rebuild_evidence( source_manifest, commands=[{"argv": ["cargo", "build"], "exit_code": 1}] ) result = project.run_packager( "--pack", "--binary-dir", str(project.binary_dir), "--tag", f"v{RESERVED_VERSION}", "--rebuild-status", "verified", "--rebuild-evidence", str(wrong_command), expect=1, ) self.assertIn("records exit code 1", result.stderr) self.assertIn("does not list the documented command", result.stderr)
def test_rebuild_status_without_evidence_is_a_usage_error(self) -> None: project = self.ready_project() vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--tag", f"v{RESERVED_VERSION}", "--rebuild-status", "verified", expect=2, ) self.assertIn("--rebuild-evidence", result.stderr)
def test_non_ubuntu_environment_is_refused_without_development(self) -> None: project = self.ready_project() descriptor = json.loads(project.write_build_env().read_text(encoding="utf-8")) descriptor["os"] = {"id": "debian", "version_id": "13"} (project.binary_dir / "build-env.json").write_text(json.dumps(descriptor), encoding="utf-8") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--tag", f"v{RESERVED_VERSION}", expect=1, ) self.assertIn("pinned environment", result.stderr)
def test_unverified_environment_is_labeled_untested_in_development(self) -> None: project = self.ready_project() project.write_build_env(toolchain={"rustc": "1.92.0", "cargo": "1.92.0"}) vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", ) manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["build"]["environment_status"], "untested_development") self.assertTrue(manifest["development"]) checks = {item["check"]: item for item in manifest["verification"]} self.assertEqual(checks["build_environment_observed"]["state"], "not_verified") self.assertIn("not verified as the pinned environment", " ".join(manifest["publication"]["blocking"]))
def test_lightweight_tag_is_refused(self) -> None: project = self.ready_project() self.git("tag", "-d", f"v{RESERVED_VERSION}") self.git("tag", f"v{RESERVED_VERSION}") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--tag", f"v{RESERVED_VERSION}", expect=1, ) self.assertIn("annotated tag", result.stderr)
def test_tag_name_must_match_the_workspace_version(self) -> None: project = self.ready_project() self.git("tag", "-a", "v9.9.9", "-m", "wrong") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--tag", "v9.9.9", expect=1, ) self.assertIn("does not name the workspace version", result.stderr)
def test_tag_pointing_at_another_commit_is_refused(self) -> None: project = self.ready_project() self.git("commit", "--quiet", "--allow-empty", "-m", "advance") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--tag", f"v{RESERVED_VERSION}", expect=1, ) self.assertIn("points at commit", result.stderr)
def test_dirty_tagged_tree_is_refused(self) -> None: project = self.ready_project() (project.root / "crates/ultimate-tool-cli/src/main.rs").write_text("fn main() { /* dirty */ }\n", encoding="utf-8") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--tag", f"v{RESERVED_VERSION}", expect=1, ) self.assertIn("dirty", result.stderr)
def test_release_mode_blocks_a_development_package(self) -> None: project = self.project() project.full_package() result = project.run_checker("--mode", "release", expect=1) self.assertIn("development package", result.stdout) self.assertIn("verdict: failed", result.stdout)
def test_reserved_version_change_blocks_preflight(self) -> None: project = self.ready_project() (project.root / "Cargo.toml").write_text( (project.root / "Cargo.toml").read_text(encoding="utf-8").replace(RESERVED_VERSION, "0.3.0-preview.1"), encoding="utf-8", ) self.git("add", "-A") self.git("commit", "--quiet", "-m", "bump") self.git("tag", "-f", "-a", "v0.3.0-preview.1", "-m", "preview") self.prepare_then_pack(project, "v0.3.0-preview.1") manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) self.assertFalse(manifest["release_ready"]) self.assertIn("not the reserved preview version", " ".join(manifest["publication"]["blocking"])) project.run_checker("--mode", "release", expect=1)
# ---------------------------------------------------------------------------# Independent verification# ---------------------------------------------------------------------------
class BoundAndSafetyTests(PackageTestCase): def test_vendor_symlink_is_refused(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") (vendor / "sneaky").symlink_to("/etc/passwd") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=1 ) self.assertIn("symbolic link", result.stderr)
def test_vendor_fifo_is_refused(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") os.mkfifo(vendor / "pipe") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=1 ) self.assertIn("not a regular file", result.stderr)
def test_vendor_sentinel_blocks_before_any_archive(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") (vendor / "serde-1.0.229/Cargo.toml").write_text( PLANTED_KEY + "\n", encoding="utf-8" ) result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=1 ) self.assertIn("vendored crate sources contain disallowed content", result.stderr) self.assertFalse(any(project.output.glob("*.tar.gz")))
def test_output_directory_rejects_a_symlinked_leaf(self) -> None: project = self.project() real = project.root.parent / "real-output" real.mkdir() project.output.symlink_to(real) vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=1 ) self.assertIn("symbolic link", result.stderr)
def test_output_directory_rejects_unrelated_contents(self) -> None: project = self.project() project.output.mkdir(parents=True) (project.output / "notes.txt").write_text("mine\n", encoding="utf-8") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=2 ) self.assertIn("does not hold a prepared source stage", result.stderr) self.assertEqual((project.output / "notes.txt").read_text(encoding="utf-8"), "mine\n")
def test_fifo_allowlisted_file_is_refused_without_blocking(self) -> None: project = self.project() os.mkfifo(project.root / "crates/ultimate-tool-cli/src/pipe.rs") result = project.run_packager( "--prepare-source", "--development", "synthetic", expect=1 ) self.assertIn("not a regular file", result.stderr)
def test_injected_member_limit_is_enforced_end_to_end(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", "--max-members", "1", expect=1, ) self.assertIn("above the limit", result.stderr)
def test_validate_release_api_reports_structures(self) -> None: project = self.project() project.full_package() development = check_release.validate_release( project.output, source_root=project.root, mode="development", require_ready=False ) self.assertEqual(development["verdict"], "passed") self.assertFalse(development["ready"]) self.assertFalse(development["release_ready_as_recorded"]) self.assertTrue(any(item["check"] == "mandatory_artifacts" for item in development["checks"])) release = check_release.validate_release( project.output, source_root=project.root, mode="release", require_ready=True ) self.assertEqual(release["verdict"], "failed") self.assertTrue(release["blockers"]) self.assertTrue(any("dev" in item for item in release["blockers"])) self.assertTrue(any(item["check"] == "binary_remeasured" and item["state"] == "verified" for item in release["checks"]))
def test_validate_release_rejects_an_unknown_mode(self) -> None: with self.assertRaises(ValueError): check_release.validate_release(self.tmp, mode="production")
def test_validate_release_performs_no_network_access(self) -> None: source = (TOOLS / "check_release.py").read_text(encoding="utf-8") for module in ("socket", "urllib", "http.client", "requests"): with self.subTest(module=module): self.assertNotIn(module, source)
def test_limits_are_recorded_in_the_release_manifest(self) -> None: project = self.project() project.full_package() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["limits"], dataclasses.asdict(packaging.DEFAULT_LIMITS))
class RebuildEvidenceTests(PackageTestCase): def test_evidence_requires_the_packager_environment(self) -> None: project = self.project() project.write_binaries() project.write_build_env() vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) wrong_environment = project.write_rebuild_evidence( source_manifest, environment={"base_image_digest": "sha256:" + "0" * 64, "rustc": "1.92.0", "target_triple": "x"}, ) result = project.run_packager( "--pack", "--binary-dir", str(project.binary_dir), "--development", "synthetic", "--rebuild-status", "verified", "--rebuild-evidence", str(wrong_environment), expect=1, ) self.assertIn("pinned base image", result.stderr) self.assertIn("rustc 1.93.0", result.stderr)
def test_evidence_must_match_the_packaged_outputs(self) -> None: project = self.project() project.write_binaries() project.write_build_env() vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) outputs = [ {"name": "ultimate-tool", "size": 1, "sha256": "0" * 64}, {"name": "ultimate-tool-mcp", "size": 1, "sha256": "0" * 64}, ] evidence = project.write_rebuild_evidence(source_manifest, outputs=outputs) result = project.run_packager( "--pack", "--binary-dir", str(project.binary_dir), "--development", "synthetic", "--rebuild-status", "verified", "--rebuild-evidence", str(evidence), expect=1, ) self.assertIn("does not match the measured", result.stderr)
class DescriptorWriterTests(unittest.TestCase): WRITER = TOOLS / "release-env/ultimate-tool-build-env"
def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory(prefix="ut-build-env-") self.tmp = Path(self._tmp.name) self.addCleanup(self._tmp.cleanup)
def run_writer(self, *arguments: str) -> subprocess.CompletedProcess: return subprocess.run( ["sh", str(self.WRITER), *arguments], capture_output=True, text=True, timeout=60, check=False, cwd=self.tmp, )
@unittest.skipUnless(shutil.which("sh") and shutil.which("rustc") and shutil.which("python3"), "needs sh, rustc and python3") def test_writer_writes_once_and_refuses_to_overwrite(self) -> None: target = self.tmp / "build-env.json" first = self.run_writer(str(target)) self.assertEqual(first.returncode, 0, first.stderr) descriptor = json.loads(target.read_text(encoding="utf-8")) self.assertEqual(descriptor["record_kind"], "build_environment") original = target.read_bytes() second = self.run_writer(str(target)) self.assertNotEqual(second.returncode, 0) self.assertIn("refusing to overwrite", second.stderr) self.assertEqual(target.read_bytes(), original) replaced = self.run_writer(str(target), "--replace") self.assertEqual(replaced.returncode, 0, replaced.stderr)
@unittest.skipUnless(shutil.which("sh") and shutil.which("rustc") and shutil.which("python3"), "needs sh, rustc and python3") def test_writer_never_follows_a_symlinked_leaf(self) -> None: victim = self.tmp / "victim.json" victim.write_text("keep me\n", encoding="utf-8") target = self.tmp / "build-env.json" target.symlink_to(victim) result = self.run_writer(str(target)) self.assertNotEqual(result.returncode, 0) self.assertEqual(victim.read_text(encoding="utf-8"), "keep me\n")
@unittest.skipUnless(shutil.which("sh") and shutil.which("rustc") and shutil.which("python3"), "needs sh, rustc and python3") def test_actual_descriptor_validates_against_the_manifest(self) -> None: os_release = {} for line in Path("/etc/os-release").read_text(encoding="utf-8").splitlines(): key, _, value = line.partition("=") os_release[key] = value.strip().strip('"') if os_release.get("ID") != "ubuntu" or not os_release.get("VERSION_ID", "").startswith("24.04"): self.skipTest(f"host is {os_release.get('ID')} {os_release.get('VERSION_ID')}, not the documented baseline") target = self.tmp / "build-env.json" result = self.run_writer(str(target)) self.assertEqual(result.returncode, 0, result.stderr) descriptor = json.loads(target.read_text(encoding="utf-8")) manifest = json.loads(REAL_MANIFEST.read_text(encoding="utf-8")) # The writer records the full version line; the packager must accept it. self.assertTrue(descriptor["toolchain"]["rustc"].startswith("rustc 1.93.0")) outcome = packaging.validate_build_environment(descriptor, manifest, development=False) self.assertEqual(outcome["status"], "verified", outcome["problems"])
class ReviewerGateTests(PackageTestCase): """The mandatory review cases: evidence binding, shipped content, bounds."""
# -- HIGH 1: registered rebuild evidence --------------------------------- def test_missing_registered_evidence_blocks_the_gate(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") project.write_binaries() project.write_build_env() project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) evidence = project.write_rebuild_evidence(source_manifest) project.run_packager( "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", "--rebuild-status", "verified", "--rebuild-evidence", str(evidence), ) registered = project.output / "release-rebuild-evidence.json" self.assertTrue(registered.exists()) result = project.run_checker("--json").stdout checks = {item["check"]: item for item in json.loads(result)["checks"]} self.assertEqual(checks["clean_extracted_source_rebuild"]["state"], "verified") registered.unlink() checks = {item["check"]: item for item in json.loads(project.run_checker("--json", expect=1).stdout)["checks"]} self.assertEqual(checks["clean_extracted_source_rebuild"]["state"], "not_verified") self.assertIn("clean_extracted_source_rebuild", project.run_checker("--mode", "release", expect=1).stdout)
def test_tampered_registered_evidence_fails_preflight(self) -> None: project = self.project() project.development_package_with_evidence() registered = project.output / "release-rebuild-evidence.json" registered.write_bytes(registered.read_bytes() + b"\n") checks = { item["check"]: item for item in json.loads(project.run_checker("--json", expect=1).stdout)["checks"] } self.assertEqual(checks["clean_extracted_source_rebuild"]["state"], "not_verified")
def test_evidence_bound_to_another_archive_is_refused(self) -> None: project = self.project() project.development_package_with_evidence() registered = project.output / "release-rebuild-evidence.json" document = json.loads(registered.read_text(encoding="utf-8")) document["source_archive"]["sha256"] = "0" * 64 payload = (json.dumps(document, indent=2, sort_keys=True) + "\n").encode() registered.write_bytes(payload) manifest_path = project.output / "release-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["source_completeness"]["evidence_file_sha256"] = hashlib.sha256(payload).hexdigest() manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") checks = { item["check"]: item for item in json.loads(project.run_checker("--json", expect=1).stdout)["checks"] } self.assertEqual(checks["clean_extracted_source_rebuild"]["state"], "not_verified") self.assertIn("not bound", checks["clean_extracted_source_rebuild"]["detail"])
def test_registered_evidence_name_must_stay_inside_the_release_directory(self) -> None: project = self.project() project.development_package_with_evidence() manifest_path = project.output / "release-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["source_completeness"]["evidence_file"] = "/etc/passwd" manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") output = project.run_checker("--json", expect=1).stdout checks = {item["check"]: item for item in json.loads(output)["checks"]} self.assertEqual(checks["clean_extracted_source_rebuild"]["state"], "not_verified") self.assertIn("missing or unsafe", checks["clean_extracted_source_rebuild"]["detail"])
# -- HIGH 2: shipped content -------------------------------------------- def test_hidden_file_beside_an_example_is_refused(self) -> None: project = self.project() (project.root / "examples/first-run/.env").write_text("SECRET=1\n", encoding="utf-8") project.write_binaries() project.write_build_env() vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", expect=1, ) self.assertIn("hidden entry in a shipped directory", result.stderr)
def test_private_corpus_file_beside_an_example_is_refused(self) -> None: project = self.project() (project.root / "examples/first-run/.private").mkdir() (project.root / "examples/first-run/.private/run.json").write_text("{}\n", encoding="utf-8") project.write_binaries() project.write_build_env() vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", expect=1, ) self.assertIn("hidden entry in a shipped directory", result.stderr)
def test_credential_beside_an_example_blocks_the_binary_archive(self) -> None: project = self.project() (project.root / "examples/first-run/notes.txt").write_text(CREDENTIAL + "\n", encoding="utf-8") project.write_binaries() project.write_build_env() vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", expect=1, ) self.assertIn("disallowed content", result.stderr)
def test_local_evidence_directory_is_skipped_not_fatal(self) -> None: project = self.project() project.full_package() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) with tarfile.open(project.output / manifest["archive"]["name"], "r:gz") as tar: names = tar.getnames() self.assertFalse(any(".ultimate-tool" in name for name in names))
def test_vendor_review_findings_reach_the_gate(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") (vendor / "serde-1.0.229/README.md").write_text(f"pinned device {DEVICE_ID}\n", encoding="utf-8") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) self.assertGreaterEqual(source_manifest["privacy"]["review_required_findings"], 1) self.assertTrue( any(item["path"].startswith("vendor/") for item in source_manifest["privacy"]["findings"]) )
# -- Medium 4: decompressed / header bounds ------------------------------- def test_counted_stream_stops_at_the_bound(self) -> None: class RecordingHandle: def __init__(self, data: bytes) -> None: self._buffer = io.BytesIO(data) self.bytes_read = 0 self.reads = 0
def read(self, size: int = -1) -> bytes: chunk = self._buffer.read(size) self.bytes_read += len(chunk) self.reads += 1 return chunk
def tell(self) -> int: return self._buffer.tell()
def seek(self, offset: int, whence: int = 0) -> int: return self._buffer.seek(offset, whence)
handle = RecordingHandle(b"x" * 4096) reader = packaging.CountingReader(handle, 12, "instrumented stream") self.assertEqual(len(reader.read(8)), 8) with self.assertRaises(packaging.CheckFailure) as caught: reader.read(8) self.assertIn("decompressed limit", str(caught.exception)) consumed_at_breach = handle.bytes_read with self.assertRaises(packaging.CheckFailure): reader.read(8) self.assertEqual(handle.bytes_read, consumed_at_breach, "no content was read after the breach")
def test_audit_stops_at_the_decompressed_bound(self) -> None: target = self.tmp / "bomb.tar.gz" build_tar(target, [("pkg/a", b"a" * 4096), ("pkg/b", b"b" * 4096), ("pkg/c", b"c" * 4096)]) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target, packaging.Limits(max_total_uncompressed_bytes=8192)) self.assertIn("decompressed limit", str(caught.exception)) self.assertIn("no further content was read", str(caught.exception))
def test_pax_extension_headers_count_toward_the_header_bound(self) -> None: target = self.tmp / "pax.tar.gz" with tarfile.open(target, "w:gz", format=tarfile.PAX_FORMAT) as tar: info = tarfile.TarInfo("pkg/" + "long-name-" * 20) info.size = 4 info.mtime = 0 tar.addfile(info, io.BytesIO(b"data")) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target, packaging.Limits(max_header_bytes=512)) self.assertIn("header and extension bytes", str(caught.exception))
# -- Medium 7: exclusive output ------------------------------------------ def test_archive_write_refuses_to_overwrite(self) -> None: target = self.tmp / "once.tar.gz" members = [("pkg/a", b"a", 0o644)] packaging.write_archive(target, members) digest = packaging.sha256_file(target) with self.assertRaises(packaging.CheckFailure) as caught: packaging.write_archive(target, [("pkg/a", b"changed", 0o644)]) self.assertIn("refusing to overwrite", str(caught.exception)) self.assertEqual(packaging.sha256_file(target), digest)
def test_foreign_source_archive_suffix_is_refused(self) -> None: project = self.project() project.output.mkdir(parents=True) (project.output / "ultimate-tool-9.9.9-linux-x86_64-gnu-source.tar.gz").write_bytes(b"foreign") (project.output / "source-manifest.json").write_text("{}\n", encoding="utf-8") vendor = project.write_vendor(project.root.parent / "vendor") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=2 ) self.assertIn("does not hold a prepared source stage", result.stderr)
# -- Medium 5: snapshot inputs ------------------------------------------- def test_snapshot_and_materializer_paths_are_allowlisted(self) -> None: manifest = json.loads(REAL_MANIFEST.read_text(encoding="utf-8")) files, _ = packaging.collect_source_files(ROOT, manifest) paths = {entry["path"] for entry in files} snapshot = json.loads((ROOT / "examples/first-run/snapshot.json").read_text(encoding="utf-8"))
def referenced(node) -> list[str]: found: list[str] = [] if isinstance(node, dict): for key, value in node.items(): if key in ("path", "source", "artifact") and isinstance(value, str): found.append(value) else: found.extend(referenced(value)) elif isinstance(node, list): for item in node: found.extend(referenced(item)) return found
for candidate in referenced(snapshot): relative = candidate[2:] if candidate.startswith("./") else candidate if relative.startswith("/") or ".." in Path(relative).parts: continue if (ROOT / relative).exists(): with self.subTest(path=relative): self.assertIn(relative, paths) for required in ( "examples/first-run/tools/materialize_demo.py", "examples/first-run/tools/prepare_demo_snapshot.py", "examples/first-run/sources/fixtures/rust-mos-smoke/.cargo/config.toml", "examples/first-run/sources/fixtures/rust-mos-smoke/link.ld", "examples/first-run/sources/crates/ultimate-tool-c64/src/lib.rs", ): with self.subTest(required=required): self.assertIn(required, paths)
class ReviewerGateGitTests(TaggedReleaseTests): def test_dirty_paths_are_component_aware(self) -> None: project = self.ready_project() sibling = project.root / "dist-private" sibling.mkdir() (sibling / "notes.txt").write_text("not the output directory\n", encoding="utf-8") state = packaging.resolve_source_state( project.root, None, True, exclude_prefixes=[project.root / "dist"] ) self.assertTrue(state["dirty"], state) self.assertTrue(any(path.startswith("dist-private") for path in state["dirty_paths"]), state["dirty_paths"]) inside = packaging.resolve_source_state( project.root, None, True, exclude_prefixes=[project.root / "dist-private"] ) self.assertFalse(any(path.startswith("dist-private") for path in inside["dirty_paths"]))
class CheckReleaseTests(PackageTestCase): def test_development_package_passes_integrity(self) -> None: project = self.project() project.full_package() result = project.run_checker(expect=0) self.assertIn("verdict: passed", result.stdout) report = json.loads(project.run_checker("--json").stdout) states = {item["check"]: item["state"] for item in report["checks"]} self.assertEqual(states["binary_remeasured"], "verified") self.assertEqual(states["checksums"], "verified") self.assertEqual(states["archive_safety"], "verified") self.assertEqual(states["source_archive_contents"], "verified") self.assertEqual(states["annotated_version_tag"], "pending")
def test_tampered_archive_fails_integrity(self) -> None: project = self.project() project.full_package() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) archive = project.output / manifest["archive"]["name"] payload = bytearray(archive.read_bytes()) payload[len(payload) // 2] ^= 0xFF archive.write_bytes(bytes(payload)) result = project.run_checker(expect=1) self.assertIn("FAIL", result.stdout)
def test_tampered_manifest_hash_fails_integrity(self) -> None: project = self.project() project.full_package() path = project.output / "release-manifest.json" manifest = json.loads(path.read_text(encoding="utf-8")) manifest["source_archive"]["sha256"] = "0" * 64 path.write_text(json.dumps(manifest), encoding="utf-8") result = project.run_checker(expect=1) self.assertIn("source archive hash disagrees", result.stdout)
def test_replaced_binary_fails_remeasurement(self) -> None: project = self.project() project.full_package() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) archive = project.output / manifest["archive"]["name"] members = [] with tarfile.open(archive, "r:gz") as tar: for member in tar.getmembers(): handle = tar.extractfile(member) if member.isreg() else None data = handle.read() if handle else b"" if member.name.endswith("bin/ultimate-tool"): data = build_elf64(glibc_versions=("GLIBC_2.34", "GLIBC_2.38"), payload=b"swapped payload\n") member = tarfile.TarInfo(member.name) member.size = len(data) member.mode = 0o755 members.append((member, data)) rebuild = project.tmp_path("swapped.tar.gz") with tarfile.open(rebuild, "w:gz") as tar: for member, data in members: if member.isreg(): import io
tar.addfile(member, io.BytesIO(data)) shutil.copyfile(rebuild, archive) sums = project.output / "SHA256SUMS" lines = [ f"{packaging.sha256_file(project.output / name)} {name}" for name in (manifest["archive"]["name"], manifest["source_archive"]["name"], "release-manifest.json") ] sums.write_text("\n".join(lines) + "\n", encoding="utf-8") result = project.run_checker(expect=1) self.assertIn("binary_remeasured", result.stdout)
def test_unrecorded_privacy_finding_fails(self) -> None: project = self.project() project.full_package() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) source_manifest_path = project.output / "source-manifest.json" source_manifest = json.loads(source_manifest_path.read_text(encoding="utf-8")) archive = project.output / manifest["source_archive"]["name"] prefix = manifest["source_archive"]["prefix"] planted = f'# {"device " + DEVICE_ID}\n'.encode() members = [] with tarfile.open(archive, "r:gz") as tar: for member in tar.getmembers(): handle = tar.extractfile(member) if member.isreg() else None data = handle.read() if handle else b"" if member.name == f"{prefix}/examples/first-run/c64dev.toml": data = planted member = tarfile.TarInfo(member.name) member.size = len(data) member.mode = 0o644 members.append((member, data)) with tarfile.open(archive, "w:gz") as tar: for member, data in members: if member.isreg(): import io
tar.addfile(member, io.BytesIO(data)) digest = packaging.sha256_file(archive) source_manifest["archive"]["sha256"] = digest source_manifest_path.write_text(json.dumps(source_manifest), encoding="utf-8") manifest["source_archive"]["sha256"] = digest (project.output / "release-manifest.json").write_text(json.dumps(manifest), encoding="utf-8") (project.output / "SHA256SUMS").write_text( "\n".join( [ f"{packaging.sha256_file(project.output / manifest['archive']['name'])} {manifest['archive']['name']}", f"{digest} {manifest['source_archive']['name']}", f"{packaging.sha256_file(project.output / 'release-manifest.json')} release-manifest.json", ] ) + "\n", encoding="utf-8", ) result = project.run_checker(expect=1) self.assertIn("does not record", result.stdout)
def test_report_file_is_written(self) -> None: project = self.project() project.full_package() target = project.root.parent / "preflight.json" project.run_checker("--report", str(target), "--mode", "release", expect=1) report = json.loads(target.read_text(encoding="utf-8")) self.assertEqual(report["mode"], "release") self.assertEqual(report["verdict"], "failed") self.assertTrue(report["blockers"])
# ---------------------------------------------------------------------------# Repository-level checks# ---------------------------------------------------------------------------
class ShippedScaffoldingTests(unittest.TestCase): def test_release_manifest_is_well_formed(self) -> None: manifest = json.loads(REAL_MANIFEST.read_text(encoding="utf-8")) self.assertEqual(manifest["schema"], 1) self.assertEqual(manifest["record_kind"], "release_source_manifest") self.assertEqual(manifest["release"]["reserved_preview_version"], RESERVED_VERSION) self.assertEqual(manifest["build_environment"]["base_image_digest"], PINNED_DIGEST) self.assertEqual(manifest["release"]["build_command"], BUILD_COMMAND) self.assertEqual(manifest["build_environment"]["rust_version"], "1.93.0") self.assertEqual(manifest["build_environment"]["node_version"], "22.23.3") self.assertEqual(manifest["build_environment"]["playwright_version"], "1.64.0") self.assertIn("node:22.23.3-bookworm-slim@sha256:", manifest["build_environment"]["tool_images"]["node"]) self.assertIn("rust:1.93.0-slim-bookworm@sha256:", manifest["build_environment"]["tool_images"]["rust"]) self.assertEqual(manifest["privacy"]["sentinel_policy"][0]["severity"], "block") policy = {entry["id"]: entry["severity"] for entry in manifest["privacy"]["sentinel_policy"]} self.assertEqual(set(policy), {identifier for identifier, _ in packaging._SENTINEL_PATTERNS}) self.assertTrue( any(entry["status"] == "pending_external" for entry in manifest["inventory"]), "the shipped inventory must still gate source completeness until the toolchain notices are collected", )
def test_every_allowlisted_path_exists(self) -> None: manifest = json.loads(REAL_MANIFEST.read_text(encoding="utf-8")) for entry in manifest["source_paths"]["include"]: with self.subTest(path=entry["path"]): self.assertTrue((ROOT / entry["path"]).exists(), f"{entry['path']} is missing from the repository")
def test_denied_private_sources_are_not_allowlisted(self) -> None: manifest = json.loads(REAL_MANIFEST.read_text(encoding="utf-8")) denied = {entry["path"] for entry in manifest["source_paths"]["deny"]} included = {entry["path"] for entry in manifest["source_paths"]["include"]} self.assertIn("AGENTS.md", denied) self.assertIn("README.md", denied) self.assertFalse(denied & included)
def test_allowlist_covers_the_workspace_and_its_fixtures(self) -> None: manifest = json.loads(REAL_MANIFEST.read_text(encoding="utf-8")) included = {entry["path"] for entry in manifest["source_paths"]["include"]} self.assertIn("crates/", included) self.assertIn("fixtures/", included) self.assertIn("sdk/", included) self.assertIn("examples/", included) self.assertIn("tests/", included) self.assertIn("fixtures/rust-mos-smoke/.cargo/config.toml", included) info = packaging.read_workspace_version(ROOT) self.assertEqual(len(info["members"]), 5, info["members"]) self.assertEqual(info["version"], manifest["release"]["reserved_preview_version"]) self.assertEqual(info["license"], "GPL-3.0-or-later") files, _ = packaging.collect_source_files(ROOT, manifest) paths = {entry["path"] for entry in files} for member in info["members"]: with self.subTest(member=member): self.assertIn(f"{member}/Cargo.toml", paths) self.assertIn("fixtures/rust-mos-smoke/Cargo.lock", paths) self.assertIn("fixtures/checkpoint-demo/link.ld", paths) self.assertIn("sdk/ultimate-tool-riscv/runtime/u64ii/linker.ld", paths) self.assertIn("examples/first-run/tools/materialize_demo.py", paths) self.assertIn("examples/first-run/outputs/build/ultimate-tool-smoke.prg", paths) self.assertNotIn("examples/first-run/build/ultimate-tool-smoke.prg", paths) self.assertTrue(paths.isdisjoint({"AGENTS.md", "README.md", "c64dev.toml"}))
def test_full_check_image_explicitly_installs_and_checks_rust_components(self) -> None: dockerfile = (TOOLS / "release-env/Dockerfile").read_text(encoding="utf-8") self.assertIn("RUN rustup component add --toolchain 1.93.0 rustfmt clippy", dockerfile) self.assertIn("cargo +1.93.0 fmt --version", dockerfile) self.assertIn("cargo +1.93.0 clippy --version", dockerfile) self.assertLess(dockerfile.index("rustup component add"), dockerfile.index("USER ${BUILD_UID}:${BUILD_GID}"))
def test_dockerfile_matches_the_manifest_pin(self) -> None: dockerfile = (TOOLS / "release-env/Dockerfile").read_text(encoding="utf-8") manifest = json.loads(REAL_MANIFEST.read_text(encoding="utf-8")) build = manifest["build_environment"] self.assertIn(build["base_image_digest"], dockerfile) self.assertIn(f"ubuntu:24.04@{build['base_image_digest']}", dockerfile) self.assertIn("1.93.0", dockerfile) self.assertIn(build["node_version"], dockerfile) self.assertIn(build["tool_images"]["rust"], dockerfile) self.assertIn(build["tool_images"]["node"], dockerfile) for reference in (build["tool_images"]["rust"], build["tool_images"]["node"]): with self.subTest(reference=reference): self.assertIn("@sha256:", reference) # A dangling npm or missing Playwright tooling would break `make check-all` # inside the image, so the module tree and the browser setup are required. self.assertIn("/usr/local/lib/node_modules", dockerfile) self.assertNotIn("COPY --from=node-toolchain /usr/local/bin/npm", dockerfile) self.assertIn("npm/bin/npm-cli.js", dockerfile) self.assertIn("npm/bin/npx-cli.js", dockerfile) self.assertIn("npm ci", dockerfile) self.assertIn("playwright install", dockerfile) self.assertIn("with-deps chromium", dockerfile) self.assertIn("USER ${BUILD_UID}:${BUILD_GID}", dockerfile) self.assertNotIn("USER builder", dockerfile) self.assertIn("--user", dockerfile) # The build context is the repository root, so every COPY path is rooted # there and no context relative to tools/release-env is assumed. self.assertIn("COPY tools/release-env/ultimate-tool-build-env", dockerfile) self.assertIn("COPY package.json package-lock.json", dockerfile) self.assertIn("-f tools/release-env/Dockerfile", dockerfile) self.assertNotIn("COPY ultimate-tool-build-env", dockerfile) ignore = (ROOT / ".dockerignore").read_text(encoding="utf-8") self.assertTrue(ignore.startswith("#") or "*" in ignore) for pattern in ("*", "!package.json", "!package-lock.json", "!tools/release-env/Dockerfile"): with self.subTest(pattern=pattern): self.assertIn(pattern, ignore) self.assertNotIn("!.git", ignore) self.assertNotIn("!.ultimate-tool", ignore)
def test_build_env_writer_records_observed_values_only(self) -> None: script = (TOOLS / "release-env/ultimate-tool-build-env").read_text(encoding="utf-8") self.assertIn("record_kind\": \"build_environment", script) self.assertIn("rustc -vV", script) self.assertIn("ldd --version", script) self.assertIn("/etc/os-release", script) self.assertNotIn("release_source_manifest.json", script)
def test_tools_use_only_the_standard_library(self) -> None: local = {"package_release", "check_release", "runtime_source_bundle", "verify_release_local", "build_release_source"} for path in sorted(TOOLS.glob("*.py")): tree = ast.parse(path.read_text(encoding="utf-8")) for node in ast.walk(tree): modules: list[str] = [] if isinstance(node, ast.Import): modules = [alias.name for alias in node.names] elif isinstance(node, ast.ImportFrom) and node.level == 0: modules = [node.module or ""] for module in modules: top = module.split(".")[0] if not top: continue with self.subTest(path=path.name, module=module): self.assertTrue( top in sys.stdlib_module_names or top in local, f"{path.name} imports non-standard module {module}", )
def test_releasing_document_covers_the_required_steps(self) -> None: document = (ROOT / "docs/RELEASING.md").read_text(encoding="utf-8") for anchor in ( "--prepare-source", "--pack", "cargo build --locked --release -p ultimate-tool -p ultimate-tool-mcp", "tools/release-env/Dockerfile", PINNED_DIGEST, "annotated", "SHA256SUMS", "check_release.py", "publish_tangled_artifacts.py", ): with self.subTest(anchor=anchor): self.assertIn(anchor, document) for disclaimer in ("not verified", "development"): self.assertIn(disclaimer, document.lower())
class MandatoryReviewTests(PackageTestCase): """The mandatory cases: audit gating, read bounds, fixture policy, contract."""
# -- failed archive audit gates every dependent read ---------------------- def test_failed_audit_makes_zero_downstream_reads(self) -> None: project = self.project() project.development_package_with_evidence() manifest = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) binary = project.output / manifest["archive"]["name"] calls: list[str] = [] real = packaging.open_bounded_tar
def counting(path, limits=packaging.DEFAULT_LIMITS): calls.append(path.name) return real(path, limits)
# Control: how many times a *successful* audit opens its own archive. with unittest.mock.patch.object(packaging, "open_bounded_tar", counting): check_release.validate_release( project.output, source_root=project.root, mode="development", require_ready=False ) baseline = calls.count(manifest["archive"]["name"]) self.assertGreaterEqual(baseline, 1)
# Replace the binary archive with a duplicate-member archive so its audit fails. build_tar(binary, [("dup/a", b"one"), ("dup/a", b"two")]) calls.clear() with unittest.mock.patch.object(packaging, "open_bounded_tar", counting): result = check_release.validate_release( project.output, source_root=project.root, mode="development", require_ready=False ) checks = {item["check"]: item for item in result["checks"]} for name in ("binary_archive_members", "binary_remeasured", "in_archive_manifest"): with self.subTest(check=name): self.assertEqual(checks[name]["state"], "not_verified") self.assertIn("did not pass its audit", checks[name]["detail"]) # The failed archive is never opened again beyond its own audit passes. self.assertLessEqual(calls.count(manifest["archive"]["name"]), baseline)
# -- per-header early refusal ------------------------------------------- def test_large_member_reads_are_chunked_not_refused(self) -> None: """An ordinary multi-megabyte member must still read normally.""" class Recording: def __init__(self, data: bytes) -> None: self._buffer = io.BytesIO(data) self.requests: list[int] = []
def read(self, size: int = -1) -> bytes: self.requests.append(size) return self._buffer.read(size)
def tell(self) -> int: return self._buffer.tell()
def seek(self, offset: int, whence: int = 0) -> int: return self._buffer.seek(offset, whence)
payload = b"x" * (11 << 20) # the real runtime bundle is ~11 MB handle = Recording(payload) reader = packaging.CountingReader(handle, 64 << 20, "instrumented", 64 << 10) data = reader.read(len(payload)) self.assertEqual(len(data), len(payload)) self.assertTrue(all(size <= (64 << 10) for size in handle.requests), max(handle.requests)) self.assertLessEqual(reader.max_requested, 64 << 10)
def test_oversized_extension_header_is_refused_before_decompression(self) -> None: target = self.tmp / "crafted-pax.tar.gz" declared = 64 << 20 block = bytearray(b"\0" * 512) block[0:6] = b"pax.h\0" block[100:108] = b"0000644\0" block[108:116] = b"0000000\0" block[116:124] = b"0000000\0" block[124:136] = ("%011o\0" % declared).encode() block[136:148] = b"00000000000\0" block[148:156] = b" " * 8 block[156:157] = b"x" # PAX extended header record with gzip.open(target, "wb") as handle: handle.write(bytes(block)) with self.assertRaises(packaging.CheckFailure) as caught: packaging.audit_tar(target) self.assertIn("refused before decompression", str(caught.exception)) self.assertIn("extension limit", str(caught.exception))
def test_prescan_accepts_a_legitimate_pax_archive(self) -> None: target = self.tmp / "legit-pax.tar.gz" with tarfile.open(target, "w:gz", format=tarfile.PAX_FORMAT) as tar: info = tarfile.TarInfo("pkg/" + "long-name-" * 20) info.size = 4 info.mtime = 0 tar.addfile(info, io.BytesIO(b"data")) audit = packaging.audit_tar(target) self.assertEqual(audit["members"], 1)
# -- vendored public fixture policy ------------------------------------- def _vendor_project_with_review(self, planted: str, *, tamper_after: bool = False): project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") crate = vendor / "serde-1.0.229" planted_path = crate / "tests" / "fixture.rs" planted_path.parent.mkdir(parents=True, exist_ok=True) planted_path.write_text(planted, encoding="utf-8") digest = packaging.sha256_file(planted_path) review = { "schema": 1, "record_kind": "vendored_fixture_review", "review": "pending", "entries": [ { "path": "vendor/serde-1.0.229/tests/fixture.rs", "sentinel": "private_network_address", "sha256": digest, "upstream": "serde-1.0.229", } ], } review_path = project.root / "tools/release-vendor-fixture-review.json" review_path.write_text(json.dumps(review, indent=2), encoding="utf-8") project.manifest["privacy"]["vendored_fixture_review_file"] = "tools/release-vendor-fixture-review.json" project.manifest["privacy"]["vendored_fixture_review_sha256"] = packaging.sha256_file(review_path) project.write_manifest_file() if tamper_after: planted_path.write_text(planted + "// changed\n", encoding="utf-8") return project, vendor
def test_reviewed_fixture_is_pending_not_allowed(self) -> None: project, vendor = self._vendor_project_with_review("let host = " + '"' + PRIVATE_ADDRESS + '";\n') project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) findings = source_manifest["privacy"]["findings"] exempted = [item for item in findings if item["severity"] == "exempt_vendored_public_fixture"] self.assertEqual(len(exempted), 1, findings) self.assertEqual(exempted[0]["review"], "pending") self.assertEqual(source_manifest["privacy"]["vendored_fixture_pending_review"], 1)
def test_changed_bytes_at_a_reviewed_path_are_refused(self) -> None: project, vendor = self._vendor_project_with_review( "let host = " + '"' + PRIVATE_ADDRESS + '";\n', tamper_after=True ) result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=1 ) self.assertIn("review the crate again instead of widening the exemption", result.stderr)
def test_unlisted_vendored_sentinel_is_refused(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") (vendor / "serde-1.0.229/notes.rs").write_text(CREDENTIAL + "\n", encoding="utf-8") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=1 ) self.assertIn("vendored crate sources contain disallowed content", result.stderr)
def test_unlisted_vendored_review_finding_is_recorded_not_exempt(self) -> None: """A review-required sentinel in an unlisted vendored file is recorded, never exempted.""" project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") (vendor / "serde-1.0.229/notes.rs").write_text(f"device {DEVICE_ID}\n", encoding="utf-8") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) findings = source_manifest["privacy"]["findings"] self.assertTrue(any(item["path"].endswith("notes.rs") for item in findings), findings) self.assertFalse( any(item["severity"] == "exempt_vendored_public_fixture" for item in findings), findings ) self.assertGreaterEqual(source_manifest["privacy"]["review_required_findings"], 1)
def test_private_planted_file_can_never_inherit_an_exemption(self) -> None: project, vendor = self._vendor_project_with_review("let host = " + '"' + PRIVATE_ADDRESS + '";\n') (vendor / "serde-1.0.229/Cargo.toml").write_text(CREDENTIAL + "\n", encoding="utf-8") result = project.run_packager( "--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic", expect=1 ) self.assertIn("disallowed content", result.stderr)
# -- shared rebuild contract ------------------------------------------- def test_rebuild_contract_rejects_future_schema_wrong_size_and_duplicates(self) -> None: facts = { "expected_archive": {"name": "a.tar.gz", "sha256": "a" * 64}, "expected_aggregate": "b" * 64, "expected_binaries": {"ultimate-tool": {"sha256": "c" * 64, "size": 10}}, "expected_environment": { "base_image_digest": "sha256:" + "d" * 64, "rust_version": "1.93.0", "target_triple": "x86_64-unknown-linux-gnu", }, "commands_required": [packaging.BUILD_COMMAND], } good = { "schema": 1, "record_kind": "release_rebuild_evidence", "source_archive": {"name": "a.tar.gz", "sha256": "a" * 64}, "source_aggregate_sha256": "b" * 64, "environment": { "base_image_digest": "sha256:" + "d" * 64, "rustc": "1.93.0 (abc)", "target_triple": "x86_64-unknown-linux-gnu", }, "commands": [{"argv": list(packaging.BUILD_COMMAND), "exit_code": 0}], "outputs": [{"name": "ultimate-tool", "sha256": "c" * 64, "size": 10}], } self.assertEqual(packaging.validate_rebuild_document(good, **facts), [])
for label, mutate, expected in ( ("future schema", lambda doc: doc.update({"schema": 2}), "must be schema 1"), ( "wrong kind", lambda doc: doc.update({"record_kind": "release_rebuild_evidence_v2"}), "must be schema 1", ), ("wrong size", lambda doc: doc["outputs"][0].update({"size": 11}), "does not match the measured size"), ("wrong hash", lambda doc: doc["outputs"][0].update({"sha256": "e" * 64}), "does not match the measured hash"), ( "duplicate outputs", lambda doc: doc.update({"outputs": doc["outputs"] + [dict(doc["outputs"][0])]}), "duplicate outputs", ), ("missing output", lambda doc: doc.update({"outputs": []}), "no rebuilt outputs"), ("missing command field", lambda doc: doc["commands"][0].pop("exit_code"), "missing argv or exit_code"), ): with self.subTest(case=label): document = json.loads(json.dumps(good)) mutate(document) problems = packaging.validate_rebuild_document(document, **facts) self.assertTrue(problems, label) self.assertTrue(any(expected in problem for problem in problems), problems)
if __name__ == "__main__": unittest.main()