Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
223 kB · 4823 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804380538063807380838093810381138123813381438153816381738183819382038213822382338243825382638273828382938303831383238333834383538363837383838393840384138423843384438453846384738483849385038513852385338543855385638573858385938603861386238633864386538663867386838693870387138723873387438753876387738783879388038813882388338843885388638873888388938903891389238933894389538963897389838993900390139023903390439053906390739083909391039113912391339143915391639173918391939203921392239233924392539263927392839293930393139323933393439353936393739383939394039413942394339443945394639473948394939503951395239533954395539563957395839593960396139623963396439653966396739683969397039713972397339743975397639773978397939803981398239833984398539863987398839893990399139923993399439953996399739983999400040014002400340044005400640074008400940104011401240134014401540164017401840194020402140224023402440254026402740284029403040314032403340344035403640374038403940404041404240434044404540464047404840494050405140524053405440554056405740584059406040614062406340644065406640674068406940704071407240734074407540764077407840794080408140824083408440854086408740884089409040914092409340944095409640974098409941004101410241034104410541064107410841094110411141124113411441154116411741184119412041214122412341244125412641274128412941304131413241334134413541364137413841394140414141424143414441454146414741484149415041514152415341544155415641574158415941604161416241634164416541664167416841694170417141724173417441754176417741784179418041814182418341844185418641874188418941904191419241934194419541964197419841994200420142024203420442054206420742084209421042114212421342144215421642174218421942204221422242234224422542264227422842294230423142324233423442354236423742384239424042414242424342444245424642474248424942504251425242534254425542564257425842594260426142624263426442654266426742684269427042714272427342744275427642774278427942804281428242834284428542864287428842894290429142924293429442954296429742984299430043014302430343044305430643074308430943104311431243134314431543164317431843194320432143224323432443254326432743284329433043314332433343344335433643374338433943404341434243434344434543464347434843494350435143524353435443554356435743584359436043614362436343644365436643674368436943704371437243734374437543764377437843794380438143824383438443854386438743884389439043914392439343944395439643974398439944004401440244034404440544064407440844094410441144124413441444154416441744184419442044214422442344244425442644274428442944304431443244334434443544364437443844394440444144424443444444454446444744484449445044514452445344544455445644574458445944604461446244634464446544664467446844694470447144724473447444754476447744784479448044814482448344844485448644874488448944904491449244934494449544964497449844994500450145024503450445054506450745084509451045114512451345144515451645174518451945204521452245234524452545264527452845294530453145324533453445354536453745384539454045414542454345444545454645474548454945504551455245534554455545564557455845594560456145624563456445654566456745684569457045714572457345744575457645774578457945804581458245834584458545864587458845894590459145924593459445954596459745984599460046014602460346044605460646074608460946104611461246134614461546164617461846194620462146224623462446254626462746284629463046314632463346344635463646374638463946404641464246434644464546464647464846494650465146524653465446554656465746584659466046614662466346644665466646674668466946704671467246734674467546764677467846794680468146824683468446854686468746884689469046914692469346944695469646974698469947004701470247034704470547064707470847094710471147124713471447154716471747184719472047214722472347244725472647274728472947304731473247334734473547364737473847394740474147424743474447454746474747484749475047514752475347544755475647574758475947604761476247634764476547664767476847694770477147724773477447754776477747784779478047814782478347844785478647874788478947904791479247934794479547964797479847994800480148024803480448054806480748084809481048114812481348144815481648174818481948204821482248234824#!/usr/bin/env python3"""Package a local Ultimate Tool preview release from already-built binaries.
The operator builds the two host binaries outside this tool, in the pinnedenvironment described by ``tools/release-env/Dockerfile``::
cargo build --locked --release -p ultimate-tool -p ultimate-tool-mcp
This tool never compiles anything. It verifies what was built, collects theallowlisted corresponding source, and writes a release directory that``tools/check_release.py`` re-verifies independently.
Two entry modes:
``--prepare-source`` Resolve ``tools/release_source_manifest.json``, vendor the host crate sources with the real ``cargo vendor`` command, and write the corresponding source archive plus ``source-manifest.json``.
``--pack`` Measure both built binaries as ELF64 x86-64 GNU artifacts, validate the observed build-environment descriptor, assemble the binary archive with the license, generated install instructions, shipped docs and examples, and write ``release-manifest.json`` and ``SHA256SUMS``.
Both modes may be combined. Packaging a dirty or untagged tree requires anexplicit ``--development`` label; such a package can never pass the publicationpreflight in ``tools/check_release.py``.
What this tool does not do: it does not build, does not tag, does not push, doesnot publish, does not contact the network (except that ``cargo vendor`` fetchescrate sources while preparing the source archive), and does not touch a device.
Two narrow, optional options wire the verified third-party runtimecorresponding-source bundle into a package::
--runtime-sources-report PATH collector report (record_kind release_runtime_sources) --runtime-sources-dir PATH directory holding the verified runtime archive
They must be given together. When they are, the packager reads the runtimearchive through the shared bounded, no-follow reader, ships the ``runtime-sources/``tree inside the host source archive and the ``runtime-notices/`` tree beside thebinaries, and resolves the pending third-party inventory entries from the verifiedcoverage. When they are absent the inventory is left exactly as the staticmanifest records it: the pending entries are retained and source completenessstays incomplete. The packager never approves the coverage mapping on a human'sbehalf, and release readiness stays false either way.
Exit codes: 0 success, 1 a check failed or a policy blocked packaging, 2 usageerror, 3 internal error."""
from __future__ import annotations
import argparseimport gzipimport hashlibimport importlib.utilimport ioimport jsonimport osimport reimport shutilimport statimport subprocessimport sysimport tarfileimport tempfileimport timeimport tomllibfrom contextlib import contextmanagerfrom dataclasses import asdict, dataclassfrom pathlib import Pathfrom typing import Callable
SCHEMA = 1PRODUCT = "ultimate-tool"EXIT_OK = 0EXIT_FAILED = 1EXIT_USAGE = 2EXIT_INTERNAL = 3
BUILD_COMMAND = [ "cargo", "build", "--locked", "--release", "-p", "ultimate-tool", "-p", "ultimate-tool-mcp",]BINARY_NAMES = ("ultimate-tool", "ultimate-tool-mcp")SOURCE_MANIFEST_RELATIVE = "tools/release_source_manifest.json"REGISTERED_EVIDENCE_NAME = "release-rebuild-evidence.json"ARCHIVE_MODE_FILE = 0o644ARCHIVE_MODE_EXECUTABLE = 0o755
# Directories and suffixes that are never archived, whatever the allowlist says.ALWAYS_EXCLUDED_DIRS = frozenset({".git", ".ultimate-tool", "node_modules", "__pycache__"})ALWAYS_EXCLUDED_SUFFIXES = (".pyc",)
TARGET_TRIPLE = "x86_64-unknown-linux-gnu"EM_X86_64 = 62ELFTYPE_EXEC = 2ELFTYPE_DYN = 3SHT_DYNAMIC = 6SHT_NOTE = 7SHT_GNU_VERNEED = 0x6FFFFFFEPT_INTERP = 3
# Privacy sentinels. Each literal is split so that this file never contains the# pattern text it searches for, and the test suite asserts the scanner does not# flag its own implementation._SENTINEL_PATTERNS = ( ( "private_key_block", re.compile(b"-" * 5 + b"BEGIN [A-Z ]{0,32}" + b"PRIVATE" + b" KEY" + b"-" * 5), ), ( # The keyword leads the match so the pattern stays linear on a large run # of unrelated bytes; every quantifier is bounded. An assignment whose # target merely names where a secret lives (…_ENV, …_VAR, …_NAME) is a # variable name, not a credential value. "credential_assignment", re.compile( rb"(?i)(?:password|passwd|client_secret|secret_key|api_key|access_token|auth_token)" rb"(?!_(?:env|var|name|path|file|key_name))" rb"[A-Za-z0-9_]{0,64}\s*[:=]\s*[\"'][^\"'\s]{6,}[\"']" ), ), ( "device_identity", re.compile(rb"\b" + b"25" + b"A6" + b"0A" + rb"\b"), ), ( "private_network_address", re.compile( rb"\b(?:" rb"192\.168|" rb"10\.(?:[0-9]{1,3})|" rb"172\.(?:1[6-9]|2[0-9]|3[01])" rb")\.[0-9]{1,3}\.[0-9]{1,3}\b" ), ), ( "absolute_home_path", re.compile(rb"/home/" + rb"[A-Za-z0-9._-]{1,64}" + rb"/"), ), ( "private_run_evidence_reference", re.compile(rb"\." + rb"ultimate-tool" + rb"/runs/"), ),)SENTINEL_CHUNK = 1 << 20SENTINEL_OVERLAP = 512
@dataclass(frozen=True)class Limits: """Every bound this tool places on a read, an archive, or a tree.
The defaults are generous for a real checkout. Tests inject small values so each bound is exercised cheaply, and an exceeded bound always fails closed: a partial result is never reported as success. """
max_manifest_bytes: int = 8 << 20 max_archive_bytes: int = 4 << 30 max_members: int = 200_000 max_member_bytes: int = 1 << 30 max_total_uncompressed_bytes: int = 8 << 30 max_header_bytes: int = 64 << 20 max_member_header_bytes: int = 64 << 10 max_single_read: int = 64 << 10 max_file_bytes: int = 1 << 30 max_name_bytes: int = 4096 max_vendor_files: int = 200_000 max_vendor_bytes: int = 4 << 30 max_evidence_bytes: int = 1 << 20
DEFAULT_LIMITS = Limits()
def unsafe_member_name(name: str, limits: Limits = DEFAULT_LIMITS) -> str | None: """Return a reason when an archive member name is unsafe, else ``None``.""" if not name: return "empty member name" if len(name.encode("utf-8", "surrogateescape")) > limits.max_name_bytes: return f"member name longer than {limits.max_name_bytes} bytes" if name.startswith("/"): return "absolute member name" if name.startswith("./") or "/./" in name or name.endswith("/."): return "member name contains a current-directory segment" if name == ".." or name.startswith("../") or "/../" in name or name.endswith("/.."): return "member name contains a parent-directory segment" if "\\" in name or "\0" in name: return "member name contains a backslash or NUL" if "//" in name: return "member name contains an empty path segment" return None
def check_ancestors(path: Path) -> None: """Refuse a path whose parent chain contains a symlink.
The path itself is not required to exist. Nothing is resolved: a symlinked ancestor is reported rather than silently followed. """ parts = path.absolute().parts current = Path(parts[0]) for part in parts[1:]: current = current / part try: info = current.lstat() except FileNotFoundError: continue except OSError as error: raise CheckFailure(f"{current} could not be inspected: {error}") from error if stat.S_ISLNK(info.st_mode): raise CheckFailure(f"{current} is a symbolic link; refusing to work through it")
def is_prepared_output_entry(name: str, expected_source_archive: str | None = None) -> bool: """True when a top-level entry belongs to a prepared source stage.""" return ( name in ("source-manifest.json", "SHA256SUMS", "vendor") or (expected_source_archive is not None and name == expected_source_archive) or name.startswith(".pack-") )
def prepare_output_directory(path: Path, expected_source_archive: str | None = None) -> None: """Create the output directory or accept a valid prepared continuation.
``--pack`` continues into the directory ``--prepare-source`` wrote, so a directory holding exactly the prepared artifacts for *this* version is valid. A foreign archive with a similar suffix is not: only the expected source archive name is accepted. Anything else is refused, and nothing is ever overwritten. """ check_ancestors(path) try: info = path.lstat() except FileNotFoundError: info = None if info is not None: if stat.S_ISLNK(info.st_mode): raise CheckFailure(f"{path} is a symbolic link; pass a new directory") if not stat.S_ISDIR(info.st_mode): raise CheckFailure(f"{path} exists and is not a directory") entries = sorted(entry.name for entry in path.iterdir() if not entry.name.startswith(".pack-")) unexpected = [name for name in entries if not is_prepared_output_entry(name, expected_source_archive)] if unexpected: raise UsageError( f"{path} is not empty and does not hold a prepared source stage; pass a new directory so " f"nothing is overwritten (unexpected: {', '.join(unexpected[:5])})" ) return path.mkdir(parents=True, exist_ok=True) info = path.lstat() if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): raise CheckFailure(f"{path} was not created as a real directory")
@contextmanagerdef open_regular_readonly(path: Path, *, max_bytes: int, description: str, check_link_ancestors: bool = True): """Open a regular file for reading without following a symbolic link.
The descriptor is anchored with ``O_NOFOLLOW`` and ``O_NONBLOCK`` so a fifo, device node or last-moment symlink cannot block or redirect the read, and ``fstat`` confirms the opened descriptor is a regular file of a bounded size. The parent chain is checked too, so a path that is only reachable through a symlinked ancestor is refused. Tests can inject a small ``max_bytes``. """ if check_link_ancestors: check_ancestors(path) flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) try: descriptor = os.open(path, flags) except OSError as error: raise CheckFailure(f"{description} could not be opened safely ({path}): {error}") from error try: info = os.fstat(descriptor) if not stat.S_ISREG(info.st_mode): raise CheckFailure(f"{description} is not a regular file ({stat.filemode(info.st_mode)}): {path}") if info.st_size > max_bytes: raise CheckFailure(f"{description} is {info.st_size} bytes, above the {max_bytes} byte limit: {path}") with os.fdopen(descriptor, "rb") as handle: descriptor = -1 yield handle finally: if descriptor >= 0: os.close(descriptor)
def read_regular_bytes(path: Path, *, max_bytes: int, description: str) -> bytes: """Read a whole regular file with a running cap.
The cap is enforced while reading as well as from ``fstat``, so a file that grows after it was opened (or a stream that never reports EOF) cannot push the read past the bound. """ chunks: list[bytes] = [] total = 0 with open_regular_readonly(path, max_bytes=max_bytes, description=description) as handle: while True: chunk = handle.read(min(SENTINEL_CHUNK, max_bytes - total + 1)) if not chunk: break total += len(chunk) if total > max_bytes: raise CheckFailure(f"{description} grew past the {max_bytes} byte limit while being read: {path}") chunks.append(chunk) return b"".join(chunks)
def anchored_sha256(path: Path, *, max_bytes: int, description: str) -> str: """Hash a regular file with the same bounds as ``read_regular_bytes``.""" digest = hashlib.sha256() total = 0 with open_regular_readonly(path, max_bytes=max_bytes, description=description) as handle: while True: chunk = handle.read(min(SENTINEL_CHUNK, max_bytes - total + 1)) if not chunk: break total += len(chunk) if total > max_bytes: raise CheckFailure(f"{description} grew past the {max_bytes} byte limit while being read: {path}") digest.update(chunk) return digest.hexdigest()
def version_token(value: object, tool: str) -> str | None: """Extract a ``major.minor.patch`` version from a tool version string.
The pinned image's descriptor writer records the full version line, for example ``rustc 1.93.0 (254b59607 2026-01-19)``, while a hand-written descriptor may record the bare ``1.93.0``. Both are accepted, and the tool name prefix is optional rather than required. """ if value is None: return None match = re.search(r"(?<![0-9A-Za-z])([0-9]+(?:\.[0-9]+){1,2})", str(value)) if not match: return None return match.group(1)
class UsageError(Exception): """Invalid invocation."""
class CheckFailure(Exception): """A verification or policy check failed; packaging must not proceed."""
def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest()
def sha256_file(path: Path) -> str: digest = hashlib.sha256() with open(path, "rb") as handle: for chunk in iter(lambda: handle.read(1 << 20), b""): digest.update(chunk) return digest.hexdigest()
def utc_now() -> str: epoch = os.environ.get("SOURCE_DATE_EPOCH") if epoch: try: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(int(epoch))) except ValueError as error: # pragma: no cover - defensive raise UsageError(f"SOURCE_DATE_EPOCH is not an integer: {epoch}") from error return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
# ---------------------------------------------------------------------------# Path patterns and allowlist resolution# ---------------------------------------------------------------------------
def compile_path_pattern(pattern: str) -> re.Pattern: """Compile a manifest path pattern.
``*`` matches one path segment, ``**`` matches any number of segments, and a trailing ``/`` means "this directory and everything below it". """ text = pattern.strip() if not text: raise UsageError("empty path pattern in the release source manifest") trailing = text.endswith("/") if trailing: text = text[:-1] parts: list[str] = [] index = 0 while index < len(text): char = text[index] if char == "*": if text[index : index + 2] == "**": if text[index + 2 : index + 3] == "/": parts.append("(?:.*/)?") index += 3 continue parts.append(".*") index += 2 continue parts.append("[^/]*") index += 1 continue if char == "?": parts.append("[^/]") index += 1 continue parts.append(re.escape(char)) index += 1 body = "".join(parts) suffix = "(?:/.*)?" if trailing else "" return re.compile("^" + body + suffix + "$")
def _load_json(path: Path, limits: Limits = DEFAULT_LIMITS, description: str = "JSON document") -> dict: try: data = read_regular_bytes(path, max_bytes=limits.max_manifest_bytes, description=description) except CheckFailure: raise try: return json.loads(data) except json.JSONDecodeError as error: raise CheckFailure(f"{path} is not valid JSON: {error}") from error
def load_source_manifest(path: Path) -> dict: manifest = _load_json(path) if manifest.get("schema") != SCHEMA or manifest.get("record_kind") != "release_source_manifest": raise CheckFailure( f"{path} must be schema {SCHEMA} release_source_manifest; " f"found schema {manifest.get('schema')!r} record_kind {manifest.get('record_kind')!r}" ) for key in ("release", "build_environment", "source_paths", "privacy", "vendor", "package_layout", "inventory"): if key not in manifest: raise CheckFailure(f"{path} is missing required section {key!r}") return manifest
def sentinel_policy(manifest: dict) -> dict[str, str]: """Sentinel severities: block, review_required, or allow.""" policy = {} entries = manifest.get("privacy", {}).get("sentinel_policy") if isinstance(entries, list): for entry in entries: if not isinstance(entry, dict) or "id" not in entry: raise CheckFailure("privacy.sentinel_policy entries must be objects with an id") severity = entry.get("severity") if severity not in ("block", "review_required", "allow"): raise CheckFailure(f"privacy.sentinel_policy[{entry['id']}] has invalid severity {severity!r}") policy[entry["id"]] = severity known = {identifier for identifier, _ in _SENTINEL_PATTERNS} unknown = sorted(set(policy) - known) if unknown: raise CheckFailure(f"privacy.sentinel_policy names unknown sentinel ids: {', '.join(unknown)}") for identifier in known: policy.setdefault(identifier, "block") return policy
def scan_bytes(data: bytes) -> list[str]: """Return the sentinel identifiers matching ``data``.""" return [identifier for identifier, pattern in _SENTINEL_PATTERNS if pattern.search(data)]
def scan_file(path: Path, limits: Limits = DEFAULT_LIMITS) -> list[str]: """Stream a file and return the sentinel identifiers it matches.""" hits: list[str] = [] carry = b"" with open_regular_readonly(path, max_bytes=limits.max_file_bytes, description="file to scan") as handle: while True: chunk = handle.read(SENTINEL_CHUNK) if not chunk: break window = carry + chunk for identifier in scan_bytes(window): if identifier not in hits: hits.append(identifier) carry = window[-SENTINEL_OVERLAP:] return hits
def _hidden_components(relative: str) -> list[str]: return [part for part in relative.split("/") if part.startswith(".")]
def _is_always_excluded(relative: str) -> bool: parts = relative.split("/") if any(part in ALWAYS_EXCLUDED_DIRS for part in parts): return True # Preserve this actual adapter source module, not arbitrary build output # below a `src` directory or a second nested generated target directory. adapter_source = ["crates", "ultimate-tool-core", "src", "target"] for index, part in enumerate(parts): if part == "target" and parts[:index + 1] != adapter_source: return True return relative.endswith(ALWAYS_EXCLUDED_SUFFIXES)
def _check_regular_file(path: Path, relative: str) -> None: try: info = path.lstat() except FileNotFoundError as error: raise CheckFailure(f"allowlisted path disappeared while packaging: {relative}") from error if stat.S_ISLNK(info.st_mode): raise CheckFailure( f"allowlisted path is a symlink and is never archived: {relative}" ) if not stat.S_ISREG(info.st_mode): raise CheckFailure( f"allowlisted path is not a regular file ({stat.filemode(info.st_mode)}): {relative}" )
def _walk_allowlisted_directory(root: Path, relative_dir: str, deny_patterns: list[re.Pattern], report: dict) -> list[str]: """Collect files under an allowlisted directory, pruning excluded trees.
Excluded and denied directories are pruned rather than filtered afterwards, because a checkout can contain very large build and evidence trees that must never be walked. """ found: list[str] = [] pending = [relative_dir] while pending: current = pending.pop() base = root / current try: entries = sorted(os.scandir(base), key=lambda item: item.name) except OSError as error: raise CheckFailure(f"allowlisted directory could not be read: {current} ({error})") from error for entry in entries: relative = f"{current}/{entry.name}" if entry.name.startswith("."): report["hidden"] += 1 report.setdefault("hidden_paths", []).append(relative) continue if _is_always_excluded(relative): report["always_excluded"] += 1 continue if any(pattern.match(relative) for pattern in deny_patterns): report["denied"] += 1 continue if entry.is_dir(follow_symlinks=False): pending.append(relative) continue found.append(relative) return found
def collect_source_files( source_root: Path, manifest: dict, limits: Limits = DEFAULT_LIMITS) -> tuple[list[dict], dict]: """Resolve the allowlist into an ordered list of files with hashes.
Returns ``(files, report)`` where ``report`` counts what was skipped and why. """ include_entries = manifest["source_paths"]["include"] deny_patterns = [compile_path_pattern(entry["path"]) for entry in manifest["source_paths"]["deny"]] seen_entries: set[str] = set() selected: dict[str, str] = {} report = {"include_entries": len(include_entries), "denied": 0, "hidden": 0, "always_excluded": 0}
for entry in include_entries: relative_entry = entry["path"].strip() kind = entry.get("kind", "unknown") hidden = bool(entry.get("hidden")) if relative_entry in seen_entries: raise CheckFailure(f"duplicate include entry in the release source manifest: {relative_entry}") seen_entries.add(relative_entry) approved_build_inputs = { ".dockerignore", ".gitattributes", ".gitignore", ".tangled/workflows/smoke.yml", ".tangled/workflows/check.yml", ".tangled/workflows/emulator.yml", ".tangled/workflows/dns-diagnostic.yml", } if hidden and kind != "crate-config" and not (kind == "build-config" and relative_entry in approved_build_inputs): raise CheckFailure( f"include entry {relative_entry} is marked hidden; only per-crate cargo configuration " "or exact approved build/CI inputs may name a hidden path" ) if _hidden_components(relative_entry) and not hidden: raise CheckFailure(f"include entry {relative_entry} names a hidden component without hidden: true")
target = source_root / relative_entry if not target.exists() and not target.is_symlink(): raise CheckFailure(f"allowlisted path is missing from the source root: {relative_entry}")
if target.is_dir() and not target.is_symlink(): for relative in sorted(_walk_allowlisted_directory(source_root, relative_entry.rstrip("/"), deny_patterns, report)): hidden_parts = _hidden_components(relative) if hidden_parts and not (hidden and relative == relative_entry): report["hidden"] += 1 continue if _is_always_excluded(relative): report["always_excluded"] += 1 continue if any(pattern.match(relative) for pattern in deny_patterns): report["denied"] += 1 continue _check_regular_file(source_root / relative, relative) selected[relative] = kind else: _check_regular_file(target, relative_entry) if _is_always_excluded(relative_entry): raise CheckFailure(f"include entry {relative_entry} is always excluded from archives") if any(pattern.match(relative_entry) for pattern in deny_patterns): raise CheckFailure(f"include entry {relative_entry} is denied by the release source manifest") selected[relative_entry] = kind
files = [] for relative in sorted(selected): path = source_root / relative info = path.stat() if info.st_size > limits.max_file_bytes: raise CheckFailure( f"allowlisted file {relative} is {info.st_size} bytes, above the {limits.max_file_bytes} byte limit" ) files.append( { "path": relative, "kind": selected[relative], "size": info.st_size, "sha256": anchored_sha256(path, max_bytes=limits.max_file_bytes, description="source file"), "executable": bool(info.st_mode & stat.S_IXUSR), } ) return files, report
def aggregate_digest(entries: list[dict], key: str = "sha256") -> str: digest = hashlib.sha256() for entry in sorted(entries, key=lambda item: item["path"]): digest.update(entry["path"].encode("utf-8")) digest.update(b"\0") digest.update(str(entry[key]).encode("ascii")) digest.update(b"\n") return digest.hexdigest()
def review_exemptions(manifest: dict) -> dict[tuple[str, str], dict]: """Reviewed exemptions for files whose purpose is to contain sentinel text.
An exemption never hides a finding: it is recorded with its reason in the resolved source manifest, the release manifest and the distribution documents. It must also declare which severity it overrides, and the declared severity must match the policy. An exemption that overrides a ``block`` sentinel is recorded as blocking content and keeps the package out of release readiness, so an exemption can never turn a blocking finding into a silently publishable package. """ mapping: dict[tuple[str, str], dict] = {} entries = manifest.get("privacy", {}).get("review_exemptions") or [] for entry in entries: path = entry.get("path") identifiers = entry.get("ids") or [] reason = entry.get("reason") override = entry.get("overrides") if not path or not identifiers or not reason: raise CheckFailure("privacy.review_exemptions entries need path, ids and reason") if override not in ("review_required", "block"): raise CheckFailure( f"privacy.review_exemptions entry for {path} must declare overrides: review_required or block" ) for identifier in identifiers: mapping[(path, identifier)] = {"reason": reason, "overrides": override} return mapping
def scan_source_tree( source_root: Path, files: list[dict], policy: dict[str, str], exemptions: dict[tuple[str, str], dict] | None = None, limits: Limits = DEFAULT_LIMITS,) -> dict: """Scan every selected file and enforce the sentinel policy.""" exemptions = exemptions or {} findings: list[dict] = [] counts: dict[str, int] = {} for entry in files: hits = scan_file(source_root / entry["path"], limits) for identifier in hits: counts[identifier] = counts.get(identifier, 0) + 1 severity = policy.get(identifier, "block") exemption = exemptions.get((entry["path"], identifier)) finding = { "path": entry["path"], "sentinel": identifier, "severity": severity, } if exemption: if exemption["overrides"] != severity: raise CheckFailure( f"the exemption for {entry['path']} [{identifier}] declares overrides=" f"{exemption['overrides']!r} but the policy severity is {severity!r}" ) finding["severity"] = "exempt" if severity == "review_required" else "exempt_blocking" finding["exemption"] = exemption["reason"] findings.append(finding) blocked = [item for item in findings if item["severity"] == "block"] review = [item for item in findings if item["severity"] == "review_required"] blocking_exemptions = [item for item in findings if item["severity"] == "exempt_blocking"] if blocked: listing = "\n".join(f" {item['path']} [{item['sentinel']}]" for item in blocked[:20]) raise CheckFailure( "privacy sentinel policy blocked packaging. These files contain disallowed content; " "sanitize them or record a reviewed severity change in " f"{SOURCE_MANIFEST_RELATIVE}.\n{listing}" ) return { "files_scanned": len(files), "patterns": [identifier for identifier, _ in _SENTINEL_PATTERNS], "policy": {identifier: policy[identifier] for identifier in sorted(policy)}, "counts": {identifier: counts[identifier] for identifier in sorted(counts)}, "findings": findings, "exemptions": [item for item in findings if item["severity"] in ("exempt", "exempt_blocking")], "blocking_exemptions": blocking_exemptions, "blocking_exemption_count": len(blocking_exemptions), "blocking_findings": 0, "review_required_findings": len(review), }
# ---------------------------------------------------------------------------# ELF measurement# ---------------------------------------------------------------------------
def _u16(data: bytes, offset: int) -> int: return int.from_bytes(data[offset : offset + 2], "little")
def _u32(data: bytes, offset: int) -> int: return int.from_bytes(data[offset : offset + 4], "little")
def _u64(data: bytes, offset: int) -> int: return int.from_bytes(data[offset : offset + 8], "little")
def _cstring(data: bytes, offset: int) -> str: end = data.find(b"\0", offset) if end < 0: end = len(data) return data[offset:end].decode("utf-8", "replace")
def _version_tuple(text: str) -> tuple[int, ...]: parts = text.split("_", 1)[-1].split(".") numbers = [] for part in parts: digits = re.match(r"\d+", part) if not digits: break numbers.append(int(digits.group(0))) return tuple(numbers)
def measure_elf(path: Path, baseline_glibc: str, limits: Limits = DEFAULT_LIMITS) -> dict: """Measure one artifact and return its observed ELF record.
The record is derived only from the file's own bytes; nothing here trusts a claim made by the build or by the operator. The file is opened without following a symlink and is bounded before it is read. """ data = read_regular_bytes(path, max_bytes=limits.max_file_bytes, description=f"binary {path.name}") record: dict = { "path": path.name, "size": len(data), "sha256": sha256_bytes(data), } if len(data) < 64 or data[:4] != b"\x7fELF": raise CheckFailure(f"{path.name} is not an ELF file (bad magic)") ei_class = data[4] ei_data = data[5] e_type = _u16(data, 16) e_machine = _u16(data, 18) e_phoff = _u64(data, 32) e_shoff = _u64(data, 40) e_phentsize = _u16(data, 54) e_phnum = _u16(data, 56) e_shentsize = _u16(data, 58) e_shnum = _u16(data, 60) e_shstrndx = _u16(data, 62) record.update( { "class": "ELF64" if ei_class == 2 else f"ELFCLASS{ei_class}", "data": "LSB" if ei_data == 1 else f"ELFDATA{ei_data}", "type": {ELFTYPE_EXEC: "EXEC", ELFTYPE_DYN: "DYN"}.get(e_type, f"TYPE{e_type}"), "machine": "x86_64" if e_machine == EM_X86_64 else f"MACHINE{e_machine}", "machine_value": e_machine, "os_abi": data[7], } ) problems = [] if ei_class != 2: problems.append(f"{path.name} is not ELF64") if ei_data != 1: problems.append(f"{path.name} is not little-endian") if e_machine != EM_X86_64: problems.append(f"{path.name} machine is {e_machine}, expected {EM_X86_64} (x86_64)") if e_type not in (ELFTYPE_EXEC, ELFTYPE_DYN): problems.append(f"{path.name} type is {e_type}, expected ET_EXEC or ET_DYN")
program_headers = [] for index in range(e_phnum): offset = e_phoff + index * e_phentsize if offset + 56 > len(data): break program_headers.append( { "type": _u32(data, offset), "offset": _u64(data, offset + 8), "filesz": _u64(data, offset + 32), } ) interpreter = None for header in program_headers: if header["type"] == PT_INTERP: interpreter = _cstring(data, header["offset"]) record["interpreter"] = interpreter
sections = [] for index in range(e_shnum): offset = e_shoff + index * e_shentsize if offset + 64 > len(data): break sections.append( { "name_offset": _u32(data, offset), "type": _u32(data, offset + 4), "offset": _u64(data, offset + 24), "size": _u64(data, offset + 32), "link": _u32(data, offset + 40), } ) has_dynamic = any(section["type"] == SHT_DYNAMIC for section in sections) record["linkage"] = "dynamic" if has_dynamic else "static" if has_dynamic and not interpreter: problems.append(f"{path.name} is dynamically linked but declares no program interpreter")
build_id = None for section in sections: if section["type"] != SHT_NOTE: continue offset = section["offset"] end = offset + section["size"] while offset + 12 <= end and offset + 12 <= len(data): namesz = _u32(data, offset) descsz = _u32(data, offset + 4) note_type = _u32(data, offset + 8) name_start = offset + 12 name = data[name_start : name_start + namesz].rstrip(b"\0") desc_start = name_start + ((namesz + 3) // 4) * 4 desc = data[desc_start : desc_start + descsz] if name == b"GNU" and note_type == 3 and desc: build_id = desc.hex() break offset = desc_start + ((descsz + 3) // 4) * 4 if build_id: break record["build_id"] = build_id
required: list[str] = [] version_section = None for section in sections: if section["type"] == SHT_GNU_VERNEED: version_section = section break if version_section is not None: string_table = b"" # Version requirement names live in the section named by sh_link. for index, section in enumerate(sections): if index == version_section["link"]: string_table = data[section["offset"] : section["offset"] + section["size"]] break offset = version_section["offset"] end = offset + version_section["size"] while offset + 16 <= end and offset + 16 <= len(data): vn_cnt = _u16(data, offset + 2) vn_aux = _u32(data, offset + 8) vn_next = _u32(data, offset + 12) aux = offset + vn_aux for _ in range(vn_cnt): if aux + 16 > len(data): break name_offset = _u32(data, aux + 8) aux_next = _u32(data, aux + 12) if string_table: text = _cstring(string_table, name_offset) if text and text not in required: required.append(text) if aux_next == 0: break aux += aux_next if vn_next == 0: break offset += vn_next
glibc_required = sorted( (text for text in required if text.startswith("GLIBC_")), key=_version_tuple, ) record["version_requirements"] = sorted(required) record["glibc_required"] = glibc_required record["glibc_max_required"] = glibc_required[-1] if glibc_required else None record["glibc_baseline"] = baseline_glibc within = True if glibc_required: needed = _version_tuple(glibc_required[-1]) baseline = _version_tuple("GLIBC_" + baseline_glibc) within = needed <= baseline record["within_baseline"] = within if not within: problems.append( f"{path.name} requires glibc {record['glibc_max_required']}, " f"newer than the documented baseline {baseline_glibc}" ) if problems: raise CheckFailure("; ".join(problems)) return record
def verify_binaries(binary_dir: Path, manifest: dict) -> dict: """Measure both required binaries and fail closed on anything unexpected.""" baseline = manifest["build_environment"]["glibc"] declared = {entry["name"]: entry for entry in manifest["release"]["binaries"]} records = {} failures: list[str] = [] for name in BINARY_NAMES: if name not in declared: raise CheckFailure(f"release source manifest does not declare a binary named {name}") path = binary_dir / name if not path.exists(): raise CheckFailure( f"expected the built binary {name} in {binary_dir}; run the documented build command " "in the pinned environment first (this tool never compiles)" ) if path.is_symlink() or not path.is_file(): raise CheckFailure(f"{path} is not a regular file") try: record = measure_elf(path, baseline) except CheckFailure as error: failures.append(str(error)) continue record["package"] = declared[name]["package"] record["package_directory"] = declared[name]["directory"] records[name] = record if failures: raise CheckFailure("binary verification failed:\n" + "\n".join(f" {item}" for item in failures)) return records
def validate_build_environment(build_env: dict | None, manifest: dict, development: bool) -> dict: """Validate the observed build-environment descriptor written by the pinned image.""" expected = manifest["build_environment"] required_rust = expected["rust_version"] problems: list[str] = [] if build_env is None: problems.append( "no build-environment descriptor was supplied (expected " f"{expected['descriptor_name']} next to the binaries or via --build-env)" ) environment = None else: environment = build_env if environment.get("schema") != SCHEMA or environment.get("record_kind") != "build_environment": problems.append( "the build-environment descriptor must be schema 1 build_environment; got " f"schema {environment.get('schema')!r} record_kind {environment.get('record_kind')!r}" ) os_record = environment.get("os") or {} if os_record.get("id") != expected["os"] or str(os_record.get("version_id")) != expected["os_version_id"]: problems.append( "the build-environment descriptor does not describe the documented baseline " f"{expected['os']} {expected['os_version_id']}: got {os_record.get('id')!r} " f"{os_record.get('version_id')!r}" ) glibc = str((environment.get("glibc") or {}).get("version", "")) if not glibc.startswith(expected["glibc"]): problems.append(f"the build environment reports glibc {glibc or 'unknown'}, expected {expected['glibc']}") if environment.get("target_triple") != expected["target_triple"]: problems.append( f"the build environment reports target {environment.get('target_triple')!r}, " f"expected {expected['target_triple']!r}" ) toolchain = environment.get("toolchain") or {} for tool in ("rustc", "cargo"): reported = version_token(toolchain.get(tool), tool) if reported != required_rust: problems.append( f"the build environment reports {tool} " f"{toolchain.get(tool)!r}, expected {required_rust}" ) base_image = environment.get("base_image") or {} reference = str(base_image.get("reference", "")) digest = base_image.get("digest") if not base_image.get("pinned") or not digest: problems.append("the build environment base image is not recorded as digest-pinned") if reference != f"{expected['base_image']}@{expected['base_image_digest']}": problems.append( "the build environment base image reference does not match the pinned reference " f"{expected['base_image']}@{expected['base_image_digest']}" ) if environment.get("build_command") != manifest["release"]["build_command"]: problems.append("the build environment recorded a different build command than the manifest declares") if not environment.get("observed_at"): problems.append("the build-environment descriptor has no observed_at timestamp") if not environment.get("produced_by"): problems.append("the build-environment descriptor does not record what produced it")
if problems and not development: raise CheckFailure( "the build environment could not be verified as the documented pinned environment:\n" + "\n".join(f" {item}" for item in problems) + "\nRe-run the build in tools/release-env/Dockerfile and write the descriptor with " "/usr/local/bin/ultimate-tool-build-env, or pass --development LABEL to label these " "binaries untested." ) if not problems: status = "verified" elif development: status = "untested_development" else: # pragma: no cover - guarded above status = "unverified" return { "status": status, "environment": environment, "problems": problems, "expected": { "base_image": expected["base_image"], "base_image_digest": expected["base_image_digest"], "os": expected["os"], "os_version_id": expected["os_version_id"], "glibc": expected["glibc"], "rust_version": required_rust, "target_triple": expected["target_triple"], }, }
# ---------------------------------------------------------------------------# Source tree state# ---------------------------------------------------------------------------
def _git(source_root: Path, *arguments: str) -> tuple[int, str]: try: result = subprocess.run( ["git", *arguments], cwd=source_root, capture_output=True, text=True, timeout=60, check=False, ) except FileNotFoundError: return (127, "") except subprocess.TimeoutExpired: return (124, "") return (result.returncode, result.stdout.strip())
def resolve_source_state(source_root: Path, tag: str | None, development: bool, exclude_prefixes: list[Path]) -> dict: """Read the Git state and enforce the packaging rules for it.""" code, revision = _git(source_root, "rev-parse", "HEAD") in_git = code == 0 and bool(revision) state = { "revision": revision if in_git else None, "in_git_worktree": in_git, "dirty": None, "dirty_paths": [], "tag": tag, "tag_type": None, "tag_object": None, "tag_commit": None, "tree_state": "unknown", } if not in_git: state["tree_state"] = "no_git_metadata" if tag: raise CheckFailure("--tag was given but the source root is not a Git worktree") if not development: raise CheckFailure( "the source root has no Git metadata, so the source revision and dirty state cannot be " "recorded; pass --development LABEL to package an explicitly development-only archive" ) return state
code, status_output = _git(source_root, "status", "--porcelain") if code != 0: raise CheckFailure("git status failed; refusing to package an unverifiable worktree") dirty_paths = [] for line in status_output.splitlines(): entry = line[3:].strip() if " -> " in entry: entry = entry.split(" -> ", 1)[1] candidate = (source_root / entry).resolve() if any( candidate == prefix.resolve() or prefix.resolve() in candidate.parents for prefix in exclude_prefixes ): continue dirty_paths.append(entry) state["dirty"] = bool(dirty_paths) state["dirty_paths"] = sorted(dirty_paths)
if tag: code, tag_type = _git(source_root, "cat-file", "-t", tag) if code != 0: raise CheckFailure(f"tag {tag!r} does not exist in this repository") if tag_type != "tag": raise CheckFailure(f"tag {tag!r} is a {tag_type or 'unknown'} object, not an annotated tag") _, tag_object = _git(source_root, "rev-parse", tag) _, tag_commit = _git(source_root, "rev-list", "-n", "1", tag) state["tag_type"] = "annotated" state["tag_object"] = tag_object state["tag_commit"] = tag_commit version = read_workspace_version(source_root)["version"] if tag != f"v{version}": raise CheckFailure( f"tag {tag!r} does not name the workspace version {version!r}; expected v{version}" ) if tag_commit != revision: raise CheckFailure( f"tag {tag!r} points at commit {tag_commit}, but the source root is at {revision}" ) if state["dirty"]: raise CheckFailure( "the worktree is dirty, so an annotated release tag cannot describe it:\n" + "\n".join(f" {path}" for path in state["dirty_paths"][:20]) ) state["tree_state"] = "clean_tagged" else: state["tree_state"] = "dirty" if state["dirty"] else "clean_untagged"
if state["tree_state"] != "clean_tagged" and not development: raise CheckFailure( f"the source tree state is {state['tree_state']}; a package from a dirty or untagged tree " "must be labeled with --development LABEL and can never pass the publication preflight" ) return state
def read_workspace_version(source_root: Path) -> dict: """Read the workspace version and verify the member crates agree with it.""" manifest_path = source_root / "Cargo.toml" try: import tomllib except ModuleNotFoundError: # pragma: no cover - Python 3.10 fallback tomllib = None # type: ignore[assignment] text = manifest_path.read_text(encoding="utf-8") if tomllib is not None: document = tomllib.loads(text) else: # pragma: no cover - Python 3.10 fallback document = {} section = None for line in text.splitlines(): stripped = line.strip() if stripped.startswith("[") and stripped.endswith("]"): section = stripped.strip("[]") document.setdefault(section.split(".")[0], {}) continue if section == "workspace.package" and "=" in stripped: key, _, value = stripped.partition("=") document.setdefault("workspace", {}).setdefault("package", {})[key.strip()] = value.strip().strip('"') workspace = document.get("workspace", {}) version = workspace.get("package", {}).get("version") license_value = workspace.get("package", {}).get("license") members = workspace.get("members", []) if not version: raise CheckFailure(f"{manifest_path} has no [workspace.package] version") versions = {} disagreeing = [] for member in members: member_manifest = source_root / member / "Cargo.toml" if not member_manifest.exists(): raise CheckFailure(f"workspace member {member} has no Cargo.toml") if tomllib is not None: member_document = tomllib.loads(member_manifest.read_text(encoding="utf-8")) package = member_document.get("package", {}) declared = package.get("version") inherited = isinstance(declared, dict) and declared.get("workspace") is True else: # pragma: no cover - Python 3.10 fallback member_text = member_manifest.read_text(encoding="utf-8") inherited = "version.workspace = true" in member_text match = re.search(r'^version\s*=\s*"([^"]+)"', member_text, re.MULTILINE) declared = None if inherited else (match.group(1) if match else None) versions[member] = "workspace" if inherited else declared if not inherited and declared != version: disagreeing.append(f"{member} declares {declared!r}") if disagreeing: raise CheckFailure( f"workspace crates must be versioned together at {version}: " + ", ".join(disagreeing) ) return { "version": version, "license": license_value, "members": members, "member_versions": versions, }
# ---------------------------------------------------------------------------# Archives# ---------------------------------------------------------------------------
def _tar_info(name: str, size: int, mode: int) -> tarfile.TarInfo: info = tarfile.TarInfo(name) info.size = size info.mtime = 0 info.mode = mode info.uid = 0 info.gid = 0 info.uname = "root" info.gname = "root" info.type = tarfile.REGTYPE return info
# ---------------------------------------------------------------------------# Stable inputs: an archive member is only written from bytes that match the# metadata recorded for it# ---------------------------------------------------------------------------
#: Optional deterministic hook used by the tests. When it is set,#: ``write_archive`` calls it with ``(member_name, path)`` immediately before a#: file member's bytes are read for the archive. A test uses it to change a file#: after the manifest hashed it, and prove the packager refuses to publish an#: archive whose bytes no longer match the metadata recorded for them._member_read_hook: Callable[[str, Path], None] | None = None
def set_member_read_hook(hook: Callable[[str, Path], None] | None) -> None: """Install or clear the pre-read hook the release tests use for fault injection.""" global _member_read_hook _member_read_hook = hook
def member_diverged( name: str, expectation: tuple[int, str], actual_size: int, actual_sha256: str | None) -> CheckFailure: """Describe an input that changed between hashing and archive writing.
The message names the member and the exact divergence — a different size or a different digest — and states that nothing was published, so a caller can tell a source tree that changed during packaging apart from a corrupt or foreign archive. """ expected_size, expected_sha256 = expectation if actual_size != expected_size: detail = f"it is now {actual_size} bytes, but {expected_size} bytes were recorded when it was hashed" elif actual_sha256 is not None and actual_sha256 != expected_sha256: detail = ( f"it now hashes to {actual_sha256[:16]}…, but {expected_sha256[:16]}… was recorded when it " "was hashed" ) else: detail = "it no longer matches the metadata recorded when it was hashed" return CheckFailure( f"source member {name} changed while the archive was being prepared: {detail}. " "No archive or manifest was published, because every archived member is streamed directly " "from the bytes that are verified against the recorded size and SHA-256." )
class _VerifyingReader: """Stream a member into the archive while hashing and counting its bytes.
Exactly the bytes handed to the tar writer are hashed, so a file that is replaced after the manifest hashed it cannot slip unmatched content into a published archive. The reader never returns more than ``target_size`` bytes: a file that ends early raises instead of silently truncating the member, and memory use stays bounded to one read buffer. """
def __init__(self, handle, *, name: str, target_size: int, max_bytes: int) -> None: self._handle = handle self._name = name self._target_size = target_size self._max_bytes = max_bytes self._digest = hashlib.sha256() self.total = 0
def read(self, size: int = -1) -> bytes: remaining = self._target_size - self.total if remaining <= 0: return b"" wanted = remaining if size is None or size < 0 else min(size, remaining) chunk = self._handle.read(wanted) if not chunk: raise CheckFailure( f"{self._name} ended after {self.total} bytes while the archive was being written, " f"but {self._target_size} bytes were recorded for it; the file changed during preparation" ) self.total += len(chunk) if self.total > self._max_bytes: raise CheckFailure( f"{self._name} grew past the {self._max_bytes} byte limit while being archived" ) self._digest.update(chunk) return chunk
def sha256(self) -> str: return self._digest.hexdigest()
def write_archive( destination: Path, members: list[tuple[str, Path | bytes, int]], limits: Limits = DEFAULT_LIMITS, expected: dict[str, tuple[int, str]] | None = None,) -> dict: """Write a deterministic gzip-compressed tar archive.
Member order is sorted, mtime is pinned to zero, ownership is pinned to root, and the gzip header carries no timestamp, so identical inputs produce identical bytes. Member count, per-member size and total size are bounded before anything is written.
``expected`` maps a member name to the ``(size, sha256)`` recorded for it when its input was hashed. A file member named there is streamed through a reader that hashes exactly the bytes written to the archive, and the member is refused — before the archive is linked into place — if the streamed size or digest diverges from the recorded value. A file that changed between hashing and this read therefore fails the preparation instead of shipping unmatched content. """ check_ancestors(destination) if len(members) > limits.max_members: raise CheckFailure(f"{destination.name} would have {len(members)} members, above the limit") total = 0 for name, source, _mode in members: reason = unsafe_member_name(name, limits) if reason: raise CheckFailure(f"refusing to write an archive member ({reason}): {name}") expectation = expected.get(name) if expected else None if isinstance(source, bytes): size = len(source) elif expectation is not None: # Trust the recorded size, not a fresh stat, so a file that already # changed cannot distort the pre-flight total. size = expectation[0] else: size = source.stat().st_size if size > limits.max_member_bytes: raise CheckFailure(f"refusing to write {name}: {size} bytes is above the per-member limit") total += size if total > limits.max_total_uncompressed_bytes: raise CheckFailure(f"{destination.name} would expand to {total} bytes, above the limit") if destination.exists() or destination.is_symlink(): raise CheckFailure(f"refusing to overwrite the existing file {destination.name}") workdir = destination.parent workdir.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=workdir, prefix=".pack-tar-", delete=False) as handle: tar_path = Path(handle.name) with tempfile.NamedTemporaryFile(dir=workdir, prefix=".pack-gz-", delete=False) as handle: staged_path = Path(handle.name) try: with tarfile.open(tar_path, "w", format=tarfile.PAX_FORMAT) as tar: for name, source, mode in sorted(members, key=lambda item: item[0]): expectation = expected.get(name) if expected else None if isinstance(source, bytes): if expectation is not None: actual_sha256 = sha256_bytes(source) if len(source) != expectation[0] or actual_sha256 != expectation[1]: raise member_diverged(name, expectation, len(source), actual_sha256) tar.addfile(_tar_info(name, len(source), mode), io.BytesIO(source)) else: if _member_read_hook is not None: _member_read_hook(name, source) with open_regular_readonly( source, max_bytes=limits.max_member_bytes, description="archive input" ) as payload: live_size = os.fstat(payload.fileno()).st_size if expectation is not None and live_size != expectation[0]: raise member_diverged(name, expectation, live_size, None) # The member's declared size is the recorded size when one # is known, so the reader hands the tar writer exactly the # bytes the manifest promised. declared_size = expectation[0] if expectation is not None else live_size reader = _VerifyingReader( payload, name=name, target_size=declared_size, max_bytes=limits.max_member_bytes, ) tar.addfile(_tar_info(name, declared_size, mode), reader) if expectation is not None: final_size = os.fstat(payload.fileno()).st_size if final_size != expectation[0]: raise member_diverged(name, expectation, final_size, None) actual_sha256 = reader.sha256() if actual_sha256 != expectation[1]: raise member_diverged(name, expectation, reader.total, actual_sha256) with open(staged_path, "wb") as raw, gzip.GzipFile( fileobj=raw, mode="wb", mtime=0, compresslevel=9, filename="" ) as compressed: with open(tar_path, "rb") as uncompressed: shutil.copyfileobj(uncompressed, compressed, 1 << 20) try: os.link(staged_path, destination) except FileExistsError as error: raise CheckFailure(f"refusing to overwrite the existing file {destination.name}") from error finally: tar_path.unlink(missing_ok=True) staged_path.unlink(missing_ok=True) return audit_tar(destination, limits)
class CountingReader: """Count decompressed bytes as a tar reader consumes them.
The count covers everything the tar reader takes from the stream, which includes member data, PAX and GNU extension headers, and block padding, so a crafted archive cannot expand past the bound through extension headers. The transport is stopped as soon as the bound is crossed, so no later member content is read or scanned. """
def __init__(self, handle, limit: int, description: str, max_single_read: int = 64 << 10) -> None: self._handle = handle self._limit = limit self._description = description self._max_single_read = max_single_read self.consumed = 0 self.max_requested = 0
def _account(self, count: int) -> None: self.consumed += count if self.consumed > self._limit: raise CheckFailure( f"{self._description} expanded past the {self._limit} byte decompressed limit; " "no further content was read" )
def _budget(self, size: int) -> int: """Bound the *total* bytes this call may consume before it delegates.""" remaining = self._limit - self.consumed if remaining <= 0: raise CheckFailure( f"{self._description} is already past the {self._limit} byte decompressed limit; " "no further content was read" ) if size is None or size < 0: return remaining + 1 return min(size, remaining + 1)
def read(self, size: int = -1) -> bytes: """Satisfy a read by chunking, so legitimate large members still work.
The requested amount is assembled from bounded chunks, so an ordinary multi-megabyte member is read normally while nothing huge is ever asked of the decompressor in one call. """ total = self._budget(size) chunks: list[bytes] = [] remaining = total while remaining > 0: chunk = self._handle.read(min(remaining, self._max_single_read)) if not chunk: break self._account(len(chunk)) self.max_requested = max(self.max_requested, len(chunk)) chunks.append(chunk) remaining -= len(chunk) return b"".join(chunks)
def readinto(self, buffer) -> int: data = self.read(len(buffer)) buffer[: len(data)] = data return len(data)
def seek(self, offset: int, whence: int = 0) -> int: base = self._handle.tell() if whence == 1 else (0 if whence == 0 else None) if base is not None and base + offset > self._limit: raise CheckFailure( f"{self._description} would seek past the {self._limit} byte decompressed limit; " "no further content was read" ) position = self._handle.seek(offset, whence) self.consumed = max(self.consumed, position) return position
def tell(self) -> int: return self._handle.tell()
def readable(self) -> bool: return True
def seekable(self) -> bool: return True
def _tar_declared_size(block: bytes) -> int: """Read a tar header block's declared size field (octal or base-256).""" field = block[124:136] if field[:1] == b"\x80": # base-256 numeric value = 0 for byte in field[1:]: value = (value << 8) | byte return value text = field.split(b"\0", 1)[0].strip() if not text: return 0 try: return int(text, 8) except ValueError: return 0
def prescan_extension_headers(handle, limits: Limits = DEFAULT_LIMITS) -> int: """Walk the raw tar stream and refuse an oversized extension header early.
PAX and GNU extension records (typeflags ``x``, ``g``, ``L``, ``K``) declare their length in the 512-byte header block that precedes them. Reading that one block is enough to refuse a record that declares more than the extension bound, so a crafted archive is rejected *before* the payload is decompressed or allocated. Ordinary members only have their declared sizes bounded; their content is read later in bounded chunks. """ offset = 0 consumed_here = 0 while True: header = handle.read(512) consumed_here += len(header) if len(header) < 512: return consumed_here if header.strip(b"\0") == b"": return consumed_here typeflag = header[156:157] declared = _tar_declared_size(header) blocks = (declared + 511) // 512 if typeflag in (b"x", b"g", b"L", b"K"): if declared > limits.max_member_header_bytes: raise CheckFailure( f"a tar extension header declares {declared} bytes, above the " f"{limits.max_member_header_bytes} byte extension limit; refused before decompression" ) elif declared > limits.max_member_bytes: raise CheckFailure( f"a tar member declares {declared} bytes, above the {limits.max_member_bytes} byte member limit" ) offset += 512 + blocks * 512 consumed_here += blocks * 512 if blocks: remaining = blocks * 512 while remaining > 0: chunk = handle.read(min(remaining, limits.max_single_read)) if not chunk: return consumed_here consumed_here += len(chunk) remaining -= len(chunk) if offset > limits.max_total_uncompressed_bytes: raise CheckFailure("the archive expands past the decompressed limit during the header prescan")
@contextmanagerdef open_bounded_tar(path: Path, limits: Limits = DEFAULT_LIMITS): """Open a gzip tar for reading with bounded, no-follow, counted reads.
A raw header prescan runs first on its own decompression pass: it refuses an oversized PAX or GNU extension record before any payload is read, and bounds every member's declared size. The file is then reopened on a fresh descriptor and the same bounds are enforced again while tarfile parses the stream in bounded chunks, so ordinary large members are still read normally. """ with open_regular_readonly( path, max_bytes=limits.max_archive_bytes, description=f"archive {path.name}" ) as prescan_handle: prescan_reader = CountingReader( gzip.GzipFile(fileobj=prescan_handle, mode="rb"), limits.max_total_uncompressed_bytes, f"archive {path.name} prescan", limits.max_single_read, ) prescan_extension_headers(prescan_reader, limits) with open_regular_readonly( path, max_bytes=limits.max_archive_bytes, description=f"archive {path.name}" ) as handle: compressed = gzip.GzipFile(fileobj=handle, mode="rb") counting = CountingReader( compressed, limits.max_total_uncompressed_bytes, f"archive {path.name}", limits.max_single_read, ) with tarfile.open(fileobj=counting, mode="r") as tar: yield tar
def audit_tar(path: Path, limits: Limits = DEFAULT_LIMITS) -> dict: """Audit archive members for unsafe, duplicated or oversized entries.
Every bound fails closed: an archive that exceeds a limit or repeats a member name is rejected rather than partially inspected. """ kinds: dict[str, int] = {} unsafe: list[str] = [] duplicates: list[str] = [] seen: set[str] = set() count = 0 total_bytes = 0 header_bytes = 0 try: with open_bounded_tar(path, limits) as tar: for member in tar: count += 1 if count > limits.max_members: raise CheckFailure(f"archive {path.name} has more than {limits.max_members} members") header_bytes += max(0, member.offset_data - member.offset) if member.offset_data - member.offset > limits.max_member_header_bytes: raise CheckFailure( f"archive {path.name} member {member.name} declares " f"{member.offset_data - member.offset} header/extension bytes, above the " f"{limits.max_member_header_bytes} byte per-member limit" ) if header_bytes > limits.max_header_bytes: raise CheckFailure( f"archive {path.name} header and extension bytes exceed the " f"{limits.max_header_bytes} byte limit" ) total_bytes += member.size if member.size > limits.max_member_bytes: raise CheckFailure( f"archive {path.name} member {member.name} is {member.size} bytes, " f"above the {limits.max_member_bytes} byte limit" ) if total_bytes > limits.max_total_uncompressed_bytes: raise CheckFailure( f"archive {path.name} expands beyond the {limits.max_total_uncompressed_bytes} byte limit" ) name = member.name if name in seen: duplicates.append(name) seen.add(name) kind = member.type.decode("ascii", "replace") kinds[kind] = kinds.get(kind, 0) + 1 reason = unsafe_member_name(name, limits) if reason: unsafe.append(f"{reason}: {name}") if member.issym() or member.islnk(): unsafe.append(f"link member: {name} -> {member.linkname}") if not (member.isreg() or member.isdir()): unsafe.append(f"non-regular member ({kind}): {name}") except CheckFailure: raise except (tarfile.TarError, gzip.BadGzipFile, EOFError, OSError, ValueError) as error: raise CheckFailure(f"archive {path.name} could not be read: {type(error).__name__}: {error}") from error if duplicates: raise CheckFailure( f"archive {path.name} contains duplicate member names:\n" + "\n".join(f" {name}" for name in sorted(set(duplicates))[:20]) ) if unsafe: raise CheckFailure(f"archive {path.name} contains unsafe members:\n" + "\n".join(f" {item}" for item in unsafe)) return { "name": path.name, "sha256": sha256_file(path), "size": path.stat().st_size, "members": count, "uncompressed_bytes": total_bytes, "member_kinds": kinds, "tar_safety": "verified", "determinism": "sorted members, mtime 0, uid/gid 0, gzip mtime 0", }
def read_archive_bytes(path: Path, member_name: str, limits: Limits = DEFAULT_LIMITS) -> bytes: with open_bounded_tar(path, limits) as tar: handle = tar.extractfile(member_name) if handle is None: raise CheckFailure(f"{path.name} has no member {member_name}") return handle.read()
# ---------------------------------------------------------------------------# Vendor# ---------------------------------------------------------------------------
def run_cargo_vendor(source_root: Path, vendor_dir: Path, timeout: int) -> tuple[str, str]: command = [ "cargo", "vendor", "--locked", "--versioned-dirs", "--manifest-path", str(source_root / "Cargo.toml"), str(vendor_dir), ] try: result = subprocess.run(command, capture_output=True, text=True, timeout=timeout, check=False) except FileNotFoundError as error: raise CheckFailure("cargo is not available; run --prepare-source in the pinned image") from error except subprocess.TimeoutExpired as error: raise CheckFailure(f"cargo vendor did not finish within {timeout}s") from error if result.returncode != 0: raise CheckFailure( "cargo vendor failed with exit code " f"{result.returncode}:\n{(result.stderr or result.stdout).strip()[:2000]}" ) return " ".join(command), result.stdout
# The source-replacement configuration a packaged vendored tree needs. The# directory is deliberately archive-relative: the source archive is extracted on# another machine, so a directory naming the machine that produced the package# points at nothing there and breaks the offline rebuild.VENDOR_SOURCE_NAME = "vendored-sources"VENDOR_CONFIG_DIRECTORY = "vendor"VENDOR_CONFIG_TABLE_KEYS = ("replace-with", "directory")
def _toml_source_key(name) -> str: """A TOML key for a source name, or an explicit refusal.
Only the characters cargo actually prints in a source name are accepted, and a name that needs quoting is quoted rather than escaped by guesswork. """ if not isinstance(name, str) or not name or any(ord(ch) < 0x20 or ch in '"\\' for ch in name): raise CheckFailure(f"the vendor source configuration names a source this packager will not serialize: {name!r}") return name if re.fullmatch(r"[A-Za-z0-9_-]+", name) else f'"{name}"'
def _toml_source_value(value) -> str: """A TOML basic string for a source name, validated the same way.""" _toml_source_key(value) return f'"{value}"'
def parse_vendor_config(snippet: str) -> dict: """Parse the ``cargo vendor`` source-replacement configuration into named fields.
Returns the vendored source name, its directory, and the replacement mapping. The snippet is read as TOML and every field is named explicitly, so nothing is inferred from substrings of the text. A configuration this packager does not understand — an unsupported key, no replacement, a replacement pointing at a source other than the vendored one, no vendored directory, or more than one — is refused rather than rewritten on a guess. """ text = snippet.strip() if not text: raise CheckFailure( "cargo vendor printed no source replacement configuration; refusing to archive a " "configuration that does not reflect the observed replacements" ) try: document = tomllib.loads(text) except tomllib.TOMLDecodeError as error: raise CheckFailure(f"the vendor source configuration is not valid TOML: {error}") from error sources = document.get("source") if set(document) != {"source"} or not isinstance(sources, dict) or not sources: raise CheckFailure( "the vendor source configuration must be one [source.*] table per replaced registry" ) replacements: dict[str, str] = {} directories: dict[str, str] = {} for name, table in sources.items(): if not isinstance(table, dict) or not table: raise CheckFailure(f"the vendor source configuration has an empty [source.{name}] table") unsupported = sorted(set(table) - set(VENDOR_CONFIG_TABLE_KEYS)) if unsupported: raise CheckFailure( f"[source.{name}] carries unsupported key(s) {unsupported}; refusing to rewrite a " "configuration this packager does not understand" ) if "directory" in table: if "replace-with" in table: raise CheckFailure(f"[source.{name}] declares both a directory and a replacement") value = table["directory"] if not isinstance(value, str) or not value: raise CheckFailure(f"[source.{name}] records no directory") directories[name] = value else: target = table["replace-with"] if not isinstance(target, str) or not target: raise CheckFailure(f"[source.{name}] records no replacement target") replacements[name] = target if len(directories) != 1: raise CheckFailure( "the vendor source configuration declares " + ("no vendored source directory" if not directories else "more than one vendored source directory") ) vendored, directory = next(iter(directories.items())) if not replacements: raise CheckFailure( "the vendor source configuration replaces no source, so the vendored tree would be unused" ) for name, target in sorted(replacements.items()): if target != vendored: raise CheckFailure( f"[source.{name}] replaces with {target!r} but the only vendored source is {vendored!r}; " "refusing to drop or rewrite a replacement this packager cannot preserve" ) return {"vendored": vendored, "directory": directory, "replacements": replacements}
def render_vendor_config(vendored: str, replacements: dict[str, str]) -> str: """Serialize the vendored source configuration with the archive-relative directory.""" lines: list[str] = [] for name in sorted(replacements): lines.append(f"[source.{_toml_source_key(name)}]") lines.append(f"replace-with = {_toml_source_value(vendored)}") lines.append("") lines.append(f"[source.{_toml_source_key(vendored)}]") lines.append(f"directory = {_toml_source_value(VENDOR_CONFIG_DIRECTORY)}") return "\n".join(lines) + "\n"
def normalize_generated_vendor_config(snippet: str, vendor_dir: Path) -> tuple[str, dict]: """Make the observed ``cargo vendor`` configuration portable, without guessing.
``cargo vendor`` prints the source-replacement block for the environment it ran in and echoes the vendor directory exactly as it was passed, which in the pinned image is an absolute path such as ``/work/target/<candidate>/vendor``. That path means nothing after the source archive is extracted elsewhere, so the single directory field is rewritten to the archive-relative ``vendor`` the package ships — by parsing the observed output as TOML and naming that exact field, never by substituting text.
Everything else cargo printed is retained: every registry the run replaced still maps to the same vendored source, and the observed replacements are recorded as provenance. The printed directory must be the vendor directory this packager just produced; a snippet this cannot normalize raises instead of being dropped or replaced by a canned block.
Returns the portable configuration and the provenance record describing what was rewritten. The recorded provenance never carries the host path itself: it notes that the directory was absolute and was rewritten, not where it pointed. """ parsed = parse_vendor_config(snippet) observed = Path(parsed["directory"]) if observed.resolve() != Path(vendor_dir).resolve(): raise CheckFailure( f"cargo vendor printed the source directory {parsed['directory']!r}, which is not the vendor " "directory this packager produced" ) rendered = render_vendor_config(parsed["vendored"], parsed["replacements"]) rendered_directory = parse_vendor_config(rendered)["directory"] if rendered_directory != VENDOR_CONFIG_DIRECTORY or Path(rendered_directory).is_absolute(): raise CheckFailure( "the portable vendor source configuration does not name the archive-relative vendor directory" ) if Path(vendor_dir).is_absolute() and str(vendor_dir) in rendered: raise CheckFailure("the portable vendor source configuration still names the host vendor directory") provenance = { "source": "observed cargo vendor stdout, parsed as TOML", "vendored_source": parsed["vendored"], "replacements": {name: parsed["vendored"] for name in sorted(parsed["replacements"])}, "directory_field": f"{parsed['vendored']}.directory", "archive_directory": VENDOR_CONFIG_DIRECTORY, "directory_rewritten": parsed["directory"] != VENDOR_CONFIG_DIRECTORY, "observed_directory_absolute": observed.is_absolute(), } return rendered, provenance
def assert_portable_vendor_config(snippet: str) -> dict: """Refuse to archive a vendored-source configuration that names a host path.
The last gate before the bytes are written, and it covers every origin of the snippet: whatever produced it, the archived configuration must name the archive-relative directory the package actually extracts to. """ parsed = parse_vendor_config(snippet) if parsed["directory"] != VENDOR_CONFIG_DIRECTORY: raise CheckFailure( f"the vendored source configuration names {parsed['directory']!r}, not the archive-relative " f"{VENDOR_CONFIG_DIRECTORY!r}" ) if Path(parsed["directory"]).is_absolute(): raise CheckFailure("the vendored source configuration names an absolute path") return parsed
def review_vendor_fixtures(vendor_dir: Path, limits: Limits = DEFAULT_LIMITS) -> dict: """Report every sentinel hit in a vendored tree, with exact file hashes.
This is the *input* to a human review of public upstream crate fixtures. It exempts nothing: it produces the explicit path, sentinel and SHA-256 list a reviewer must accept, and tools/check_release.py plus the packager refuse any vendored sentinel that is absent from the accepted list. """ check_ancestors(vendor_dir) entries = [] for path in sorted(vendor_dir.rglob("*")): info = path.lstat() if not stat.S_ISREG(info.st_mode): continue relative = path.relative_to(vendor_dir).as_posix() hits = scan_file(path, limits) if not hits: continue digest = anchored_sha256(path, max_bytes=limits.max_file_bytes, description="vendored fixture") for identifier in hits: entries.append( { "path": f"vendor/{relative}", "sentinel": identifier, "sha256": digest, "upstream": relative.split("/")[0], "size": info.st_size, } ) return { "schema": SCHEMA, "record_kind": "vendored_fixture_review", "produced_by": "tools/package_release.py --review-vendor-fixtures", "review": "pending", "note": ( "Explicit list of sentinel hits inside vendored public crates, each bound to the exact file " "SHA-256. Accepting this document exempts only these exact bytes and keeps the human review gate " "pending; any other vendored sentinel, or any change to these bytes, is still refused." ), "entries": entries, }
def load_vendor_fixture_review(path: Path, limits: Limits = DEFAULT_LIMITS) -> dict[tuple[str, str], dict]: """Load the reviewed per-file fixture list, bound to its recorded SHA-256.""" document = _load_json(path, limits, "vendored fixture review") if document.get("schema") != SCHEMA or document.get("record_kind") != "vendored_fixture_review": raise CheckFailure( f"{path.name} must be schema {SCHEMA} vendored_fixture_review; got " f"schema {document.get('schema')!r} record_kind {document.get('record_kind')!r}" ) mapping: dict[tuple[str, str], dict] = {} for entry in document.get("entries") or []: relative = entry.get("path") sentinel = entry.get("sentinel") digest = entry.get("sha256") if not relative or not sentinel or not digest: raise CheckFailure(f"{path.name} has an entry without path, sentinel and sha256") if not relative.startswith("vendor/"): raise CheckFailure(f"{path.name} exempts a path outside the vendored tree: {relative}") if len(digest) != 64: raise CheckFailure(f"{path.name} entry for {relative} does not carry a SHA-256 digest") mapping[(relative, sentinel)] = { "sha256": digest, "upstream": entry.get("upstream"), "review": document.get("review", "pending"), } if not mapping: raise CheckFailure(f"{path.name} lists no reviewed vendored fixtures") return mapping
def vendored_fixture_exemptions(manifest: dict, source_root: Path | None = None) -> dict[tuple[str, str], dict]: """Load the reviewed per-file fixture list named by the release manifest.
The list is a separate, explicitly reviewed document whose own SHA-256 is recorded in the manifest, so accepting it is a deliberate, auditable act. No pattern is weakened: an entry exempts one path with one sentinel at one exact file hash, and everything else in the vendored tree is still blocked. """ privacy = manifest.get("privacy", {}) name = privacy.get("vendored_fixture_review_file") if not name: return {} if Path(name).is_absolute() or ".." in Path(name).parts: raise CheckFailure(f"privacy.vendored_fixture_review_file must be a repository-relative path: {name!r}") root = source_root or Path(__file__).resolve().parent.parent path = root / name recorded = privacy.get("vendored_fixture_review_sha256") actual = anchored_sha256(path, max_bytes=DEFAULT_LIMITS.max_manifest_bytes, description="vendored fixture review") if not recorded or recorded != actual: raise CheckFailure( f"{name} does not match the SHA-256 recorded in the release source manifest " f"(recorded {str(recorded)[:16]}…, file {actual[:16]}…); review the fixture list again" ) return load_vendor_fixture_review(path)
def collect_vendor(vendor_dir: Path, manifest: dict | None = None, policy: dict[str, str] | None = None, exemptions: dict[tuple[str, str], dict] | None = None, limits: Limits = DEFAULT_LIMITS, fixture_exemptions: dict[tuple[str, str], dict] | None = None) -> dict: """Collect and verify an operator-supplied or generated vendor tree.
The tree gets the same treatment as first-party source: no symlink, fifo or other non-regular entry anywhere, no symlinked ancestor, a bounded number of files and bytes, and a privacy-sentinel scan whose blocking findings stop packaging before any archive is written. """ check_ancestors(vendor_dir) findings: list[dict] = [] files = [] total_bytes = 0 exempt_by_crate: dict[str, list[dict]] = {} for path in sorted(vendor_dir.rglob("*")): relative = path.relative_to(vendor_dir).as_posix() reason = unsafe_member_name(f"vendor/{relative}", limits) if reason: raise CheckFailure(f"vendor entry is unsafe ({reason}): {relative}") try: info = path.lstat() except OSError as error: raise CheckFailure(f"vendor entry could not be inspected: {relative} ({error})") from error if stat.S_ISLNK(info.st_mode): raise CheckFailure(f"vendor entry is a symbolic link: {relative}") if stat.S_ISDIR(info.st_mode): continue if not stat.S_ISREG(info.st_mode): raise CheckFailure( f"vendor entry is not a regular file ({stat.filemode(info.st_mode)}): {relative}" ) if info.st_size > limits.max_file_bytes: raise CheckFailure(f"vendor file {relative} is {info.st_size} bytes, above the file limit") files.append( { "path": relative, "size": info.st_size, "sha256": anchored_sha256(path, max_bytes=limits.max_file_bytes, description="vendor file"), } ) total_bytes += info.st_size if len(files) > limits.max_vendor_files: raise CheckFailure(f"vendor tree has more than {limits.max_vendor_files} files") if total_bytes > limits.max_vendor_bytes: raise CheckFailure(f"vendor tree exceeds the {limits.max_vendor_bytes} byte limit") for identifier in scan_file(path, limits): severity = (policy or {}).get(identifier, "block") archived_path = f"vendor/{relative}" exemption = (exemptions or {}).get((archived_path, identifier)) reviewed = (fixture_exemptions or {}).get((archived_path, identifier)) finding = {"path": archived_path, "sentinel": identifier, "severity": severity} if reviewed: actual = files[-1]["sha256"] if actual != reviewed["sha256"]: raise CheckFailure( f"the reviewed fixture list records SHA-256 {reviewed['sha256'][:16]}… for " f"{archived_path} [{identifier}], but the vendored file hashes to {actual[:16]}…; " "review the crate again instead of widening the exemption" ) finding["severity"] = "exempt_vendored_public_fixture" finding["exemption"] = "reviewed public upstream fixture, exact bytes recorded" finding["upstream"] = reviewed.get("upstream") finding["sha256"] = actual finding["review"] = reviewed.get("review", "pending") elif exemption: if exemption["overrides"] != severity: raise CheckFailure( f"the exemption for {archived_path} [{identifier}] declares overrides=" f"{exemption['overrides']!r} but the policy severity is {severity!r}" ) finding["severity"] = "exempt" if severity == "review_required" else "exempt_blocking" finding["exemption"] = exemption["reason"] findings.append(finding) if not files: raise CheckFailure(f"{vendor_dir} contains no vendored crate sources") blocked = [item for item in findings if item["severity"] == "block"] if blocked: raise CheckFailure( "the vendored crate sources contain disallowed content; packaging stopped before writing any archive:\n" + "\n".join(f" {item['path']} [{item['sentinel']}]" for item in blocked[:20]) ) return { "file_count": len(files), "total_bytes": total_bytes, "aggregate_sha256": aggregate_digest(files), "files": files, "privacy_findings": findings, "blocking_exemption_count": len([item for item in findings if item["severity"] == "exempt_blocking"]), "pending_review_findings": len( [item for item in findings if item["severity"] == "exempt_vendored_public_fixture"] ), "public_fixture_exemptions": [ item for item in findings if item["severity"] == "exempt_vendored_public_fixture" ], }
LICENSE_FILE_MARKERS = ("license", "licence", "copying", "unlicense", "notice")LICENSE_FILE_SUFFIXES = (".rs", ".toml", ".json", ".lock", ".sha256")
def is_license_file(relative: str) -> bool: """Whether a vendored member looks like a license or notice text.
Only used to name the crate's own license files in the resolved inventory; it decides nothing on its own, because a vendored file whose name happens to miss every marker is reported as a crate without a named license file rather than silently treated as licensed. """ name = relative.rsplit("/", 1)[-1].lower() if name.endswith(LICENSE_FILE_SUFFIXES): return False return any(marker in name for marker in LICENSE_FILE_MARKERS)
def vendored_crate_directories(files: list[dict]) -> dict[str, list[str]]: """Group vendored member paths by their crate directory.
``cargo vendor --versioned-dirs`` writes one ``<crate-name>-<version>`` directory per crate, so the first path component is the crate directory and every member below it belongs to that crate. """ grouped: dict[str, list[str]] = {} for entry in files: relative = entry.get("path") or "" head, _, tail = relative.partition("/") if head and tail: grouped.setdefault(head, []).append(relative) return grouped
def vendor_generation_verified(vendor_record: dict) -> bool: """Whether a vendor record is backed by an observed ``cargo vendor`` run.
Three facts must hold together: the record's status says the tree was generated (not supplied), the recorded command is an actual ``cargo vendor`` invocation, and the file list is non-empty with an aggregate this function re-derives from that very list. A supplied tree fails the first fact and an empty or unhashed tree fails the third, so neither can resolve the inventory. """ if vendor_record.get("status") != "generated_cargo_vendor": return False command = vendor_record.get("command") if not isinstance(command, (list, tuple)) or len(command) < 2: return False if "vendor" not in [str(part) for part in command]: return False if not str(command[0]).endswith("cargo"): return False files = list(vendor_record.get("files") or []) if not files: return False if vendor_record.get("file_count") != len(files): return False aggregate = vendor_record.get("aggregate_sha256") return bool(aggregate) and aggregate == aggregate_digest(files)
def resolve_vendor_inventory(base_inventory: list[dict], vendor_record: dict) -> list[dict]: """Resolve the ``conditional_on_vendor`` inventory entry against a vendor tree.
Returns a resolved copy of the inventory; the static manifest keeps its ``conditional_on_vendor`` entry and is never edited here. Only this copy can flip that entry to ``collected``, and only for a vendor tree that an observed ``cargo vendor`` execution produced (see :func:`vendor_generation_verified`).
An operator-supplied ``--vendor-dir`` tree is never relabelled. Its record says the generation was not observed, so the entry keeps its pending status and source completeness stays incomplete until a matching generated run is provided.
``collected_paths`` is the whole vendored file index, copied from the same ``vendor_record['files']`` list that is archived, so the proof is the source-complete tree rather than one license file. Because the paths are derived from that list, a record edited after preparation cannot keep the entry resolved: the named members no longer exist in the archive and :func:`check_inventory` plus the independent checker refuse it. """ files = list(vendor_record.get("files") or []) resolved = vendor_generation_verified(vendor_record) proof: list[str] = [] license_paths: list[str] = [] crates: list[str] = [] without_license: list[str] = [] if resolved: proof = sorted(f"vendor/{entry['path']}" for entry in files) grouped = vendored_crate_directories(files) crates = sorted(grouped) license_paths = sorted( f"vendor/{relative}" for members in grouped.values() for relative in members if is_license_file(relative) ) without_license = sorted( name for name, members in grouped.items() if not any(is_license_file(item) for item in members) ) entries: list[dict] = [] for entry in base_inventory: entry = dict(entry) if entry.get("status") == "conditional_on_vendor" and resolved: entry.update( { "status": "collected", "collected_paths": proof, "collected_license_paths": license_paths, "collected_crates": crates, "vendor_provenance": { "status": vendor_record.get("status"), "command": list(vendor_record.get("command") or []), "directory": vendor_record.get("directory"), "file_count": vendor_record.get("file_count"), "total_bytes": vendor_record.get("total_bytes"), "aggregate_sha256": vendor_record.get("aggregate_sha256"), "source": "observed cargo vendor execution inside the pinned environment", }, "crates_without_license_files": without_license, "note": ( "the vendored crate sources were produced by an observed cargo vendor execution; " "collected_paths is the whole vendored file index and collected_license_paths names " "each crate's own license files. This is a mechanical, digest-bound statement, not a " "human license approval" ), } ) entries.append(entry) return entries
# ---------------------------------------------------------------------------# Generated distribution documents# ---------------------------------------------------------------------------
def render_readme(context: dict) -> bytes: release = context["release"] boundary = release["support_boundary"] mode = ( f"development package labeled `{context['development_label']}`; this archive is not a release candidate" if context["development"] else "release candidate" ) version = context["version"] lines = [ f"# {release['product_title']} {version} — Linux x86-64 preview", "", f"This archive is a **{mode}**.", "", f"{release['product_title']} is a hardware-in-the-loop development platform for the", "Commodore 64. It connects existing build tools, VICE, C64 Ultimate hardware, tests", "and evidence through one deterministic development core.", "", "## Contents", "", "- `bin/ultimate-tool` — the command-line interface (validate, build, test, runs, doctor, serve).", "- `bin/ultimate-tool-mcp` — the stdio adapter that wraps the same versioned service operations.", "- `examples/first-run/` — a manifest and verified prebuilt inputs for a deterministic simulated run.", "- `examples/existing-project/` — a starting manifest for an existing project layout.", "- `docs/` — the reviewed public documents that ship with this preview.", "- `LICENSE` — GNU General Public License version 3 or later, the license of this software.", "- `SOURCE-AND-LICENSES.md` — where the corresponding source and third-party notices live.", "- `source-manifest.json`, `release-manifest.json` — the hashes and provenance of these files.", "", "## Support boundary", "", f"Tested on: {boundary['tested_platform']}.", "", "Not tested and not claimed:", "", ] lines.extend(f"- {item}" for item in boundary["untested"]) lines += ["", "Documented limits:", ""] lines.extend(f"- {item}" for item in boundary["documented_limits"]) lines += [ "", "## Install and first result", "", "See `INSTALL.md`. There is no installer script: unpack the archive, verify the", "checksum with `sha256sum -c`, and run the binaries from the extracted directory.", "", "## Corresponding source", "", f"The complete corresponding source archive is `{context['source_archive']['name']}` with", f"SHA-256 `{context['source_archive']['sha256']}`. It is distributed with these binaries, not", "downloaded separately.", "", f"Source revision: `{context['source']['revision'] or 'unrecorded (no Git metadata)'}`" + (f" (tag `{context['source']['tag']}`)" if context["source"].get("tag") else "") + f"; worktree dirty: {context['source']['dirty']}.", "", "This archive records evidence of what was observed. It does not claim that a", "different compiler, architecture or operating system produces these exact bytes.", "", ] return ("\n".join(lines) + "\n").encode("utf-8")
def render_install(context: dict) -> bytes: release = context["release"] version = context["version"] archive = context["archive"]["name"] source_archive = context["source_archive"]["name"] binary_dir = archive[: -len(".tar.gz")] source_dir = source_archive[: -len(".tar.gz")] lines = [ f"# Installing {release['product_title']} {version}", "", "This preview ships Linux x86-64 (GNU) binaries. There is no install script, no", "`curl | sh`, and no network access during installation. Everything is a manual,", "inspectable step.", "", "Download the binary archive, the source archive and `SHA256SUMS` into one directory,", "then run every command below from that directory. Each step wraps a directory change", "in a subshell, so no step leaves you somewhere a later step does not expect.", "", "## 1. Verify the download", "", "`SHA256SUMS` is published next to the archives; it is intentionally not stored inside", "an archive, because an archive cannot contain its own checksum.", "", "```sh", "sha256sum -c SHA256SUMS", "```", "", f"`SHA256SUMS` covers `{archive}`, `{source_archive}` and `release-manifest.json`.", "Compare the values against `release-manifest.json`, which also records the build", "environment, the source revision and the verification results.", "", "## 2. Extract", "", "```sh", f"tar -xzf {archive}", f"( cd {binary_dir} && ./bin/ultimate-tool --version && ./bin/ultimate-tool-mcp --help )", "```", "", "Extraction writes only inside the new directory. Nothing is installed system-wide", "and nothing requires root.", "", "## 3. Run the first simulated result", "", "The packaged example needs Python 3 and nothing else. It does not use Docker, VICE", "or a C64 Ultimate, and it performs no network access:", "", "```sh", f"( cd {binary_dir}/examples/first-run && ../../bin/ultimate-tool validate )", f"( cd {binary_dir}/examples/first-run && ../../bin/ultimate-tool doctor --target fake )", f"( cd {binary_dir}/examples/first-run && ../../bin/ultimate-tool test smoke --target fake )", "```", "", "`test` returns a nonzero exit code for the deliberately failing demonstration cases;", "that is the documented behaviour, not an installation failure. The fake target", "orchestrates a deterministic simulated run and does not execute a PRG.", "", "## Runtime requirements", "", ] for item in release["runtime_requirements"]: lines.append(f"- `{item['name']}`: {item['requirement']} — needed for {item['needed_for']}.") lines += [ "- The binaries need a Linux x86-64 system with GNU libc at or above the tested", f" baseline ({release['support_boundary']['tested_platform']}).", "- Docker, VICE, a C64 Ultimate and the Rust-MOS toolchain are **not** required to run", " the packaged example. They are external prerequisites for other workflows.", "", "## Install to a prefix (optional)", "", "```sh", f"install -Dm755 {binary_dir}/bin/ultimate-tool \"$HOME/.local/bin/ultimate-tool\"", f"install -Dm755 {binary_dir}/bin/ultimate-tool-mcp \"$HOME/.local/bin/ultimate-tool-mcp\"", "```", "", "Keep the extracted directory if you want the examples, docs and manifests; the two", "binaries run from anywhere.", "", "## Uninstall", "", "```sh", "rm -f \"$HOME/.local/bin/ultimate-tool\" \"$HOME/.local/bin/ultimate-tool-mcp\"", f"rm -rf ./{binary_dir}", "```", "", "Uninstalling deletes only those paths. Project run evidence written under a project", "directory (`.ultimate-tool/`) is yours and is not touched by this archive.", "", "## Verifying the extracted source", "", "The source archive is the corresponding source for the binaries. It carries no", "per-file hash index of its own; the resolved index ships as `source-manifest.json`", "in the release directory and inside the binary archive that step 2 extracted. From", "the same download directory, extract the source archive and check the tree it", "unpacks against that index with the archive's own helper:", "", "```sh", f"tar -xzf {source_archive}", f"python3 -B {source_dir}/tools/verify_extracted_source.py \\", f" --source-root {source_dir} \\", f" --manifest {binary_dir}/source-manifest.json", "```", "", "The helper re-hashes every member the index records — first-party source, the", "sanitized distribution documents, the vendored crate tree and the runtime", "corresponding sources — requires the members the packager generates", "(`SOURCE-PROVENANCE.json`, and `vendor-sources.toml` when the archive is vendored),", "and refuses a tree with a file, directory or symlink the index does not list. It", "needs Python 3 and reads files only: no network access, no toolchain, no device. `-B`", "and the helper's own `sys.dont_write_bytecode` keep it from writing a `__pycache__`", "into the tree it is checking. It confirms the extracted bytes, so run", "`sha256sum -c SHA256SUMS` (step 1) first to verify the archive those bytes came from.", "", "Rebuilding the binaries from source requires the Rust toolchain version recorded in", "`source-manifest.json`, and rebuilding the C64 fixture PRGs additionally requires the", "pinned Rust-MOS image named in `SOURCE-AND-LICENSES.md`.", "", ] return ("\n".join(lines) + "\n").encode("utf-8")
def render_source_and_licenses(context: dict) -> bytes: release = context["release"] lines = [ "# Source and licenses", "", f"{release['product_title']} is licensed under {release['license']}. The full license", "text is in `LICENSE` at the root of this archive.", "", "## Corresponding source", "", f"- Source archive: `{context['source_archive']['name']}`", f"- SHA-256: `{context['source_archive']['sha256']}`", f"- Files: {context['source_manifest']['file_count']} allowlisted first-party files" + ( f" plus {context['source_manifest']['vendor']['file_count']} vendored crate files" if context["source_manifest"]["vendor"].get("file_count") else " and a vendored crate tree that was not included" ), "", "The source archive is always distributed with the binaries. It contains the complete", "corresponding source of the two shipped host binaries: all five workspace crates, the", "C64 SDK crate the fixtures link against, the embedded dashboard assets, the fixture", "sources with their own manifests and lockfiles, the RISC-V SDK sources, the demo", "helpers, the tests and the reviewed public documentation.", "", f"Source completeness: **{context['source_completeness']['status']}**.", "", ] pending = context["source_completeness"].get("pending_inventory", []) if pending: lines += [ "The following third-party components are linked into or bundled with the shipped", "artifacts but their source or notice obligations are not collected yet. Until they", "are, this release does not claim license or source completeness:", "", ] lines.extend(f"- {item}" for item in pending) lines.append("") lines += render_runtime_sources_section(context) lines += [ "## Third-party inventory", "", "| Component | Scope | License | Status |", "| --- | --- | --- | --- |", ] for entry in context["inventory"]: lines.append( f"| {entry['component']} | {entry['scope']} | {entry['license']} | {entry['status']} |" ) lines += [ "", "## Retail evidence and rejected material", "", "This archive contains no credential, device password, per-run token or private", "acceptance corpus. Raw run evidence directories (`.ultimate-tool/`), the Git history", "and internal agent instructions are never packaged.", "", ] findings = context["privacy"].get("findings", []) if findings: lines += [ "The following reviewed mentions of private identifiers remain in the source archive", "and are recorded here deliberately rather than silently removed:", "", ] lines.extend( f"- `{item['path']}` — {item['sentinel']} ({item['severity']})" for item in findings ) lines.append("") lines += [ "## External prerequisites (not distributed)", "", ] for entry in context["external_prerequisites"]: lines.append(f"- {entry['name']}: needed for {entry['needed_for']}. {entry['note']}") lines.append("") return ("\n".join(lines) + "\n").encode("utf-8")
# ---------------------------------------------------------------------------# Distribution documents: reviewed sanitized copies at their archive paths# ---------------------------------------------------------------------------## A few repository documents are withheld from every archive in their original# form, yet other shipped documents and the packaged fixtures link to them by# relative path and to their heading anchors. ``package_layout.distribution_docs``# records a fixed mapping from each such archive path to a reviewed sanitized# copy under ``tools/release-docs/``. The copy ships at the archive path the# withheld original occupies, so every relative link and anchor resolves inside# both archives, and the withheld original is never read into a release.
# Both archives place every sanitized copy under this fixed destination root, and# only inside it: a copy must never alias a build output, an evidence tree or a# run directory, which are excluded from every archive.DISTRIBUTION_DOC_DESTINATION_ROOT = "docs/"DISTRIBUTION_DOC_FORBIDDEN_FRAGMENTS = ("build/", "outputs/", ".ultimate-tool/", "runs/")
def _canonical_relative_path(value, label: str) -> str: """Return ``value`` when it is a canonical repo-relative path, else fail.""" if not isinstance(value, str) or not value: raise CheckFailure(f"{label} must be a non-empty string") if any(ord(character) < 32 or ord(character) == 127 for character in value): raise CheckFailure(f"{label} {value!r} must not contain control characters") if value.startswith("/") or "\\" in value or re.match(r"^[A-Za-z]:", value): raise CheckFailure(f"{label} {value!r} must be a repo-relative path, not absolute") if any(part in ("", ".", "..") for part in value.split("/")): raise CheckFailure( f"{label} {value!r} must be a canonical repo-relative path with no empty, '.' or '..' component" ) return value
def distribution_doc_root(manifest: dict) -> str: """The canonical relative directory that holds the reviewed copies.""" section = (manifest.get("package_layout") or {}).get("distribution_docs") or {} root = section.get("sanitized_root") if not isinstance(root, str) or not root: raise CheckFailure("package_layout.distribution_docs records no sanitized_root") return _canonical_relative_path(root, "distribution_docs sanitized_root").rstrip("/")
def validate_distribution_doc_entry( entry: dict, deny_patterns: list[re.Pattern], sanitized_root: str) -> tuple[str, str]: """Require one mapping to name a confined copy and a supported destination.
The copy must be a Markdown document inside the sanitized root, never the withheld original itself, and the archive path must be a Markdown document under the fixed destination root, never a build, evidence or run path. The withheld original must be denied, and the origin must be that same path, so a mapping can only ever replace a withheld document with a reviewed copy. """ archive_path = _canonical_relative_path( entry.get("archive_path"), "distribution_docs archive_path" ) copy = _canonical_relative_path(entry.get("copy"), "distribution_docs copy") if not copy.startswith(f"{sanitized_root}/"): raise CheckFailure(f"distribution_docs copy {copy!r} must live under {sanitized_root}/") if not copy.endswith(".md"): raise CheckFailure(f"distribution_docs copy {copy!r} must be a Markdown document") if copy == archive_path: raise CheckFailure( f"distribution_docs entry {archive_path!r} names the withheld original as its copy" ) if not archive_path.startswith(DISTRIBUTION_DOC_DESTINATION_ROOT) or not archive_path.endswith(".md"): raise CheckFailure( f"distribution_docs archive_path {archive_path!r} must be a Markdown document under " f"{DISTRIBUTION_DOC_DESTINATION_ROOT}" ) if any(fragment in archive_path for fragment in DISTRIBUTION_DOC_FORBIDDEN_FRAGMENTS): raise CheckFailure( f"distribution_docs archive_path {archive_path!r} targets a build, evidence or run path" ) origin = entry.get("origin", archive_path) if origin != archive_path: raise CheckFailure( f"distribution_docs origin {origin!r} must equal its archive_path {archive_path!r}, " "because a copy only ever replaces the withheld document at that exact path" ) if not any(pattern.match(archive_path) for pattern in deny_patterns): raise CheckFailure( f"distribution document {archive_path} is not denied by the release source " "manifest; the withheld original must be denied so it can never be archived" ) return archive_path, copy
def distribution_doc_mapping(manifest: dict) -> dict[str, dict]: """Return the reviewed distribution-document mapping keyed by archive path.""" section = (manifest.get("package_layout") or {}).get("distribution_docs") or {} mapping: dict[str, dict] = {} for entry in section.get("entries", []): for field in ("archive_path", "copy"): if not isinstance(entry.get(field), str) or not entry[field]: raise CheckFailure(f"distribution_docs entry is missing {field!r}: {entry!r}") archive_path = entry["archive_path"] if archive_path in mapping: raise CheckFailure(f"duplicate distribution document mapping for {archive_path}") mapping[archive_path] = entry return mapping
def distribution_doc_members( root: Path, manifest: dict, limits: Limits = DEFAULT_LIMITS) -> list[dict]: """Resolve every reviewed distribution copy to the bytes that ship.
Each returned record carries the archive path, the sanitized copy that supplies the bytes, the withheld repository original, the digest recorded for the copy in the release source manifest, and the bytes themselves.
The recorded digest is authoritative: a copy whose bytes changed without its recorded hash being updated fails preparation instead of shipping unreviewed text. Each archive path must also be denied by the manifest, so the mapping can never be satisfied by reading the withheld original. """ deny_patterns = [ compile_path_pattern(entry["path"]) for entry in manifest["source_paths"]["deny"] ] mapping = distribution_doc_mapping(manifest) if not mapping: # A manifest that maps no distribution documents — still the common case # for a repository-shaped package — declares no sanitized root and needs # none: there is nothing to place. return [] sanitized_root = distribution_doc_root(manifest) records: list[dict] = [] for key in sorted(mapping): entry = mapping[key] archive_path, copy = validate_distribution_doc_entry(entry, deny_patterns, sanitized_root) _check_regular_file(root / copy, copy) data = read_regular_bytes( root / copy, max_bytes=limits.max_file_bytes, description="distribution document copy" ) digest = sha256_bytes(data) recorded = entry.get("sha256") if recorded != digest: raise CheckFailure( f"the distribution document copy {copy} has SHA-256 {digest}, but the release " f"source manifest records {recorded!r}; re-review the copy and update its hash" ) records.append( { "archive_path": archive_path, "copy": copy, "origin": entry.get("origin", archive_path), "transformations": list(entry.get("transformations") or []), "size": len(data), "sha256": digest, "bytes": data, } ) return records
def scan_distribution_document_privacy( records: list[dict], policy: dict[str, str], exemptions: dict[tuple[str, str], dict] | None = None,) -> list[dict]: """Scan the exact bytes each distribution document ships, under the policy.
A distribution copy is a packaging input rather than an allowlisted source file, so it is not covered by the source-tree scan. The sentinel policy is applied here to the bytes that are written and to the archive path that carries them, and every finding is recorded and bounded exactly as it would be for an ordinary allowlisted file: a blocking hit stops the preparation. """ exemptions = exemptions or {} findings: list[dict] = [] for record in records: for identifier in scan_bytes(record["bytes"]): severity = policy.get(identifier, "block") exemption = exemptions.get((record["archive_path"], identifier)) finding = {"path": record["archive_path"], "sentinel": identifier, "severity": severity} if exemption: if exemption["overrides"] != severity: raise CheckFailure( f"the exemption for {record['archive_path']} [{identifier}] declares overrides=" f"{exemption['overrides']!r} but the policy severity is {severity!r}" ) finding["severity"] = "exempt" if severity == "review_required" else "exempt_blocking" finding["exemption"] = exemption["reason"] findings.append(finding) blocked = [item for item in findings if item["severity"] == "block"] if blocked: raise CheckFailure( "privacy sentinel policy blocked a distribution document. These bytes contain disallowed " "content; sanitize the review copy or record a reviewed severity change in " f"{SOURCE_MANIFEST_RELATIVE}.\n" + "\n".join(f" {item['path']} [{item['sentinel']}]" for item in blocked[:20]) ) return findings
def merge_privacy_findings(privacy: dict, findings: list[dict]) -> None: """Merge findings into a privacy record, counting each path/sentinel once.
The source archive records the distribution documents it scanned, and the binary archive re-scans the same bytes, so the same finding can arrive twice; it is counted once, by its ``(path, sentinel)`` pair, because the release manifest states what was found, not how many times it was found. """ privacy.setdefault("counts", {}) privacy.setdefault("findings", []) privacy.setdefault("exemptions", []) privacy.setdefault("blocking_exemptions", []) privacy.setdefault("review_required_findings", 0) privacy.setdefault("blocking_exemption_count", 0) seen = {(item["path"], item["sentinel"]) for item in privacy["findings"]} for finding in findings: key = (finding["path"], finding["sentinel"]) if key in seen: continue seen.add(key) privacy["findings"].append(finding) privacy["counts"][finding["sentinel"]] = privacy["counts"].get(finding["sentinel"], 0) + 1 if finding["severity"] == "review_required": privacy["review_required_findings"] += 1 if finding["severity"] == "exempt": privacy["exemptions"].append(finding) if finding["severity"] == "exempt_blocking": privacy["exemptions"].append(finding) privacy["blocking_exemptions"].append(finding) privacy["blocking_exemption_count"] += 1
# ---------------------------------------------------------------------------# Packaging steps# ---------------------------------------------------------------------------
def versioned_names(version: str, development_label: str | None) -> dict: base = f"{PRODUCT}-{version}-linux-x86_64-gnu" if development_label: base = f"{base}-development-{development_label}" return { "base": base, "archive": base + ".tar.gz", "source_archive": base + "-source.tar.gz", "prefix": base, "source_prefix": base + "-source", }
def sanitize_label(label: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", label.strip()).strip("-.") if not cleaned: raise UsageError("--development label must contain at least one letter or digit") return cleaned.lower()
def select_shipped_files( root: Path, relative_dir: str, manifest: dict, limits: Limits = DEFAULT_LIMITS) -> list[str]: """Select files to ship from a directory using the same rules as source.
Hidden entries are always refused rather than silently dropped, so a private file such as `.env` beside a reviewed example cannot ride into a release archive, and the same deny patterns and always-excluded directories apply. """ deny_patterns = [compile_path_pattern(entry["path"]) for entry in manifest["source_paths"]["deny"]] explicit_hidden_files = { entry["path"] for entry in manifest["source_paths"]["include"] if entry.get("hidden") is True and not entry["path"].endswith("/") } selected: list[str] = [] pending = [relative_dir.rstrip("/")] while pending: current = pending.pop() check_ancestors(root / current) if (root / current).is_symlink() or not (root / current).is_dir(): raise CheckFailure(f"shipped directory is not a regular directory: {current}") with os.scandir(root / current) as entries: ordered = sorted(entries, key=lambda entry: entry.name) for entry in ordered: relative = f"{current}/{entry.name}" if _is_always_excluded(relative) or any(pattern.match(relative) for pattern in deny_patterns): continue if entry.is_symlink(): raise CheckFailure(f"shipped entry is a symlink: {relative}") if _hidden_components(relative): exact_file = relative in explicit_hidden_files approved_ancestor = any( path.startswith(relative + "/") for path in explicit_hidden_files ) if not exact_file and not approved_ancestor: raise CheckFailure(f"{relative} is a hidden entry in a shipped directory; it is not explicitly approved") if entry.is_dir(follow_symlinks=False): if exact_file: raise CheckFailure(f"approved hidden file is a directory: {relative}") pending.append(relative) continue if not exact_file: raise CheckFailure(f"approved hidden ancestor is not a directory: {relative}") elif entry.is_dir(follow_symlinks=False): pending.append(relative) continue _check_regular_file(root / relative, relative) selected.append(relative) if len(selected) > limits.max_members: raise CheckFailure("shipped file count exceeds the archive member bound") return sorted(selected)
def scan_files_for_privacy( root: Path, relatives: list[str], policy: dict[str, str], exemptions: dict[tuple[str, str], dict] | None = None, limits: Limits = DEFAULT_LIMITS,) -> list[dict]: """Scan files that will be shipped in the binary archive.""" exemptions = exemptions or {} findings: list[dict] = [] for relative in relatives: for identifier in scan_file(root / relative, limits): severity = policy.get(identifier, "block") exemption = exemptions.get((relative, identifier)) finding = {"path": relative, "sentinel": identifier, "severity": severity} if exemption: if exemption["overrides"] != severity: raise CheckFailure( f"the exemption for {relative} [{identifier}] declares overrides=" f"{exemption['overrides']!r} but the policy severity is {severity!r}" ) finding["severity"] = "exempt" if severity == "review_required" else "exempt_blocking" finding["exemption"] = exemption["reason"] findings.append(finding) blocked = [item for item in findings if item["severity"] == "block"] if blocked: raise CheckFailure( "shipped example or documentation files contain disallowed content:\n" + "\n".join(f" {item['path']} [{item['sentinel']}]" for item in blocked[:20]) ) return findings
# ---------------------------------------------------------------------------# The packaged first-run example: reviewed prebuilt inputs vs. a local build# ---------------------------------------------------------------------------## The shipped demo carries a verified prebuilt snapshot. ``snapshot.json``# records every packaged file at its true path: the approved prebuilt inputs# live under ``outputs/build/``, and ``tools/materialize_demo.py`` copies them# into the example's inner ``build/`` working tree at run time. Only the# ``outputs/`` tree is release material; the ``build/`` tree is a local product# of the helper and must never be archived as if it were a prebuilt input.## ``tools/release_source_manifest.json`` states this explicitly, and the# resolver below cross-checks that statement against the example's own# snapshot so the manifest cannot drift from the record it approves. Every# archive is then verified against the resolved payload, so a stray local output# can never ride into a release, and a snapshot input can never go missing.
EXAMPLE_SNAPSHOT_KIND = "ultimate-tool-prebuilt-snapshot"#: The canonical first-party snapshot validator, loaded by this fixed path.#:#: The packager never re-implements the snapshot schema: it binds the same#: module the packaged materializer imports, so the release gate and the#: run-time materializer enforce one contract (schema 1, every required key,#: exact entry arity). The path is fixed and independent of the manifest and the#: snapshot document, and the module is imported as data-validating code only —#: no producer command recorded in a snapshot is ever executed.EXAMPLE_SNAPSHOT_VALIDATOR_RELATIVE = "examples/first-run/tools/prepare_demo_snapshot.py"_SNAPSHOT_VALIDATORS: dict[str, object] = {}
def _example_relative_path(value: str, what: str) -> str: """Refuse an example-relative path that is absolute or traverses upward.""" if not isinstance(value, str) or not value: raise CheckFailure(f"{what} must be a non-empty path") reason = unsafe_member_name(value) if reason: raise CheckFailure(f"{what} is unsafe ({reason}): {value!r}") return value
def load_snapshot_validator(source_root: Path): """Import the canonical snapshot validator from its fixed first-party path.
The module is first-party code already trusted by the operator, loaded by a constant repository-relative path rather than by a path the manifest or the snapshot could steer. Only its validation functions are called; nothing in a snapshot (including a recorded producer command) is executed. """ path = source_root / EXAMPLE_SNAPSHOT_VALIDATOR_RELATIVE key = str(path) cached = _SNAPSHOT_VALIDATORS.get(key) if cached is not None: return cached check_ancestors(path) if not path.is_file() or path.is_symlink(): raise CheckFailure( f"the canonical snapshot validator is missing from {EXAMPLE_SNAPSHOT_VALIDATOR_RELATIVE}; " "the packaged example cannot be checked against the contract the materializer uses" ) spec = importlib.util.spec_from_file_location("_ultimate_tool_snapshot_validator", path) if spec is None or spec.loader is None: raise CheckFailure(f"cannot load the canonical snapshot validator from {path}") module = importlib.util.module_from_spec(spec) try: spec.loader.exec_module(module) except Exception as error: # pragma: no cover - import-time failure of trusted code raise CheckFailure(f"the canonical snapshot validator at {path} could not be loaded: {error}") from error _SNAPSHOT_VALIDATORS[key] = module return module
def read_canonical_snapshot(source_root: Path, example: str, snapshot_relative: str) -> list[dict]: """Validate one example's snapshot with the canonical validator.
Returns the normalised ``file_entries`` list. A document that is not schema 1 with every required key — a future schema, a missing ``schema``, an unknown key, or a malformed entry — is refused here, so the packager gate fails for exactly the documents the packaged materializer would refuse later. """ validator = load_snapshot_validator(source_root) snapshot_path = source_root / example / snapshot_relative try: document = validator.read_snapshot(snapshot_path) except Exception as error: raise CheckFailure( f"{snapshot_path} is not a valid {EXAMPLE_SNAPSHOT_KIND} snapshot: {error}" ) from error entries = document.get("file_entries") if not isinstance(entries, list) or not entries: raise CheckFailure(f"{snapshot_path} declares no file_entries") return entries
def resolve_example_snapshot(manifest: dict, source_root: Path, limits: Limits = DEFAULT_LIMITS) -> dict: """Resolve the reviewed prebuilt-input payload for every shipped example.
The manifest declares, per example, the snapshot record, the ``outputs/`` tree that holds the approved immutable prebuilt inputs, the inner ``build/`` tree the materializer writes, and the exact outputs it approves. This resolver reads the example's own ``snapshot.json`` and refuses to proceed unless that record and the manifest payload agree file for file, so the reviewed list cannot silently drift from the bytes it approves.
Every returned path is repository-relative and carries the exact size and SHA-256 the snapshot records. An absent section resolves to no examples, so repository-shaped fixtures without a packaged demo keep working. """ section = manifest.get("example_snapshot_layout") if not section: return {"examples": []} if not isinstance(section, dict) or not isinstance(section.get("examples"), list): raise CheckFailure("example_snapshot_layout must be an object with an examples list") resolved: list[dict] = [] seen_examples: set[str] = set() for entry in section["examples"]: if not isinstance(entry, dict): raise CheckFailure("example_snapshot_layout entries must be objects") example = entry.get("example") if not isinstance(example, str) or not example: raise CheckFailure("example_snapshot_layout entry has no example path") example = _example_relative_path(example, "example_snapshot_layout example") if example in seen_examples: raise CheckFailure(f"example_snapshot_layout names {example} twice") seen_examples.add(example) snapshot_relative = entry.get("snapshot") outputs_root = entry.get("outputs_root") sources_root = entry.get("sources_root") materialized_root = entry.get("materialized_root") for value, what in ( (snapshot_relative, f"{example} snapshot"), (outputs_root, f"{example} outputs_root"), (sources_root, f"{example} sources_root"), (materialized_root, f"{example} materialized_root"), ): _example_relative_path(value, what) if outputs_root == materialized_root: raise CheckFailure( f"{example} cannot use {outputs_root!r} as both its outputs root and its " "materialized destination: the approved inputs and the local product are distinct trees" ) # The declared roots must be the canonical tree names, so the manifest # payload cannot drift from the contract the materializer enforces. validator = load_snapshot_validator(source_root) if outputs_root != validator.OUTPUTS_DIRNAME or sources_root != validator.SOURCES_DIRNAME: raise CheckFailure( f"{example} declares outputs/sources roots {outputs_root!r}/{sources_root!r}, but the " f"canonical snapshot layout is {validator.OUTPUTS_DIRNAME!r}/{validator.SOURCES_DIRNAME!r}" ) declared = read_canonical_snapshot(source_root, example, snapshot_relative) snapshot_path = source_root / example / snapshot_relative outputs = [item for item in declared if item["output"]] inputs = [item for item in declared if not item["output"]] if not outputs: raise CheckFailure(f"{snapshot_path} declares no build outputs") if not inputs: raise CheckFailure(f"{snapshot_path} declares no source inputs") for item in inputs: if not item["path"].startswith(sources_root + "/"): raise CheckFailure( f"{snapshot_path} declares the input {item['path']!r} outside {sources_root}/" )
manifest_outputs = entry.get("outputs") if not isinstance(manifest_outputs, list) or not manifest_outputs: raise CheckFailure(f"{example} must list its approved outputs explicitly") approved: dict[str, dict] = {} for item in manifest_outputs: if not isinstance(item, dict): raise CheckFailure(f"{example} approved outputs entries must be objects") path = _example_relative_path(item.get("path"), f"{example} approved output") if path in approved: raise CheckFailure(f"{example} lists the approved output {path!r} twice") approved[path] = {"path": path, "sha256": item.get("sha256"), "size": item.get("size")} recorded = {item["path"]: item for item in outputs} if set(approved) != set(recorded): details = [] missing = sorted(set(recorded) - set(approved)) extra = sorted(set(approved) - set(recorded)) if missing: details.append("not listed in the manifest: " + ", ".join(missing)) if extra: details.append("listed in the manifest but absent from the snapshot: " + ", ".join(extra)) raise CheckFailure( f"{example} manifest payload does not match {snapshot_relative}: " + "; ".join(details) ) for path, item in sorted(approved.items()): want = recorded[path] if item["sha256"] != want["sha256"] or item["size"] != want["size"]: raise CheckFailure( f"{example} manifest payload for {path} records size/sha256 " f"{item['size']}/{item['sha256']}, but the snapshot records " f"{want['size']}/{want['sha256']}" ) prefix = example + "/" resolved.append( { "example": example, "snapshot": prefix + snapshot_relative, "outputs_root": prefix + outputs_root, "sources_root": prefix + sources_root, "materialized_root": prefix + materialized_root, "outputs": [ {"path": prefix + path, "sha256": approved[path]["sha256"], "size": approved[path]["size"]} for path in sorted(approved) ], "inputs": [ {"path": prefix + item["path"], "sha256": item["sha256"], "size": item["size"]} for item in sorted(inputs, key=lambda item: item["path"]) ], } ) return {"examples": resolved}
def check_example_snapshot_selection( source_root: Path, manifest: dict, selected: list[str], limits: Limits = DEFAULT_LIMITS,) -> dict | None: """Verify one archive's member set against the reviewed example snapshot.
``selected`` holds the repository-relative paths chosen for a single archive (the source archive or the binary archive). Every declared prebuilt output must appear at its true ``outputs/`` path and every declared input at its ``sources/`` path, each matching the exact size and SHA-256 the snapshot records. Any undeclared file under the example's ``outputs/`` tree, and any member under its inner ``build/`` tree, is refused: the packaged example carries exactly the approved inputs and never a local materialization.
Returns the resolved record for the manifest, or ``None`` when no example is declared. """ policy = resolve_example_snapshot(manifest, source_root, limits) if not policy["examples"]: return None selected_set = set(selected) problems: list[str] = [] records: list[dict] = [] for example in policy["examples"]: approved = {item["path"] for item in example["outputs"]} declared_inputs = {item["path"] for item in example["inputs"]} outputs_prefix = example["outputs_root"] + "/" sources_prefix = example["sources_root"] + "/" materialized = example["materialized_root"] record = { "example": example["example"], "snapshot": example["snapshot"], "outputs_root": example["outputs_root"], "sources_root": example["sources_root"], "materialized_root": materialized, "outputs": [], "inputs": [], "verified": True, } for item in example["outputs"] + example["inputs"]: relative = item["path"] if relative not in selected_set: problems.append( f"{relative} is declared by {example['snapshot']} but is not in the archive" ) continue path = source_root / relative try: _check_regular_file(path, relative) info = path.stat() except CheckFailure as error: problems.append(str(error)) continue actual_sha = anchored_sha256(path, max_bytes=limits.max_file_bytes, description="packaged example input") if (info.st_size, actual_sha) != (item["size"], item["sha256"]): problems.append( f"{relative} is {info.st_size} bytes with sha256 {actual_sha}, but " f"{example['snapshot']} records {item['size']} bytes with sha256 {item['sha256']}" ) continue bucket = record["outputs"] if relative in approved else record["inputs"] bucket.append({"path": relative, "sha256": item["sha256"], "size": item["size"]}) # The materializer refuses any file under sources/ or outputs/ that the # snapshot does not declare, so the archive must carry exactly the # declared membership across both trees. Helpers outside those trees # (tools/, the manifest, the snapshot record itself) are unaffected. for relative in sorted(selected_set): if relative.startswith(outputs_prefix) and relative not in approved: problems.append( f"{relative} is an undeclared file under {example['outputs_root']}; the packaged " "example carries exactly the reviewed snapshot inputs" ) if relative.startswith(sources_prefix) and relative not in declared_inputs: problems.append( f"{relative} is an undeclared file under {example['sources_root']}; the packaged " "example carries exactly the reviewed snapshot sources" ) if relative == materialized or relative.startswith(materialized + "/"): problems.append( f"{relative} is the local materialized destination {materialized}/, which is a " "build product of the example helper and is never archived" ) records.append(record) if problems: raise CheckFailure( "the packaged example does not match the reviewed snapshot payload:\n" + "\n".join(f" {item}" for item in problems) ) return {"examples": records}
def write_registered_evidence(destination: Path, data: bytes, limits: Limits = DEFAULT_LIMITS) -> str: """Write the portable copy of the rebuild evidence, exclusively and atomically.""" if len(data) > limits.max_evidence_bytes: raise CheckFailure(f"the rebuild evidence is {len(data)} bytes, above the evidence limit") if destination.exists() or destination.is_symlink(): raise CheckFailure(f"refusing to overwrite the existing file {destination.name}") workdir = destination.parent with tempfile.NamedTemporaryFile(dir=workdir, prefix=".pack-evidence-", delete=False) as handle: staged_path = Path(handle.name) try: staged_path.write_bytes(data) try: os.link(staged_path, destination) except FileExistsError as error: raise CheckFailure(f"refusing to overwrite the existing file {destination.name}") from error finally: staged_path.unlink(missing_ok=True) return sha256_bytes(data)
def register_evidence( release_dir: Path, evidence_bytes: bytes, limits: Limits = DEFAULT_LIMITS) -> dict: """Register the rebuild evidence inside a release directory.
Returns the ``source_completeness`` fields to merge, so the packager and any fixture that writes a release directory by hand use one implementation. A manifest that claims ``clean_rebuild: verified`` without these fields and the registered file is rejected by ``tools/check_release.py``. """ digest = write_registered_evidence(release_dir / REGISTERED_EVIDENCE_NAME, evidence_bytes, limits) return { "clean_rebuild": "verified", "evidence_file": REGISTERED_EVIDENCE_NAME, "evidence_file_sha256": digest, "evidence_source_sha256": sha256_bytes(evidence_bytes), "clean_rebuild_evidence": REGISTERED_EVIDENCE_NAME, }
def verify_prepared_source( archive_path: Path, source_manifest: dict, limits: Limits = DEFAULT_LIMITS) -> dict: """Narrowly re-verify the archive produced by ``--prepare-source``.
``--pack`` continues only from a prepared source archive that matches the prepared manifest member for member: presence, size and SHA-256 for every first-party file and every vendored crate file, plus the recomputed aggregate. An empty or unmatched manifest refuses the pack.
The comparison is always against the metadata *recorded in the prepared manifest*, never against the current source tree: a source tree that changed after ``--prepare-source`` must not make the packager accept bytes that do not match the manifest the archive was written with. A mismatch therefore means the archive and the recorded inventory disagree — the archive was not written from the recorded bytes — not that the working tree moved on. """ files = source_manifest.get("files") or [] if not files: raise CheckFailure( "the prepared source manifest lists no files; run --prepare-source before packing" ) vendor = source_manifest.get("vendor") or {} distribution_docs = source_manifest.get("distribution_docs") or [] prefix = (source_manifest.get("archive") or {}).get("prefix") if not prefix: raise CheckFailure("the prepared source manifest does not record its archive prefix") audit = audit_tar(archive_path, limits) problems: list[str] = [] checked = 0 if (source_manifest.get("archive") or {}).get("name") not in (None, archive_path.name): raise CheckFailure( f"source-manifest.json describes {source_manifest['archive']['name']}, " f"not the prepared archive {archive_path.name}" ) with open_bounded_tar(archive_path, limits) as tar: members = {member.name: member for member in tar.getmembers() if member.isreg()}
def verify(relative: str, entry: dict, label: str) -> None: nonlocal checked member = members.get(f"{prefix}/{relative}") if member is None: problems.append(f"{label} {relative} is missing from the source archive") return if member.size != entry["size"]: problems.append(f"{label} {relative} has size {member.size}, expected {entry['size']}") return handle = tar.extractfile(member) digest = hashlib.sha256() while True: chunk = handle.read(1 << 20) if not chunk: break digest.update(chunk) if digest.hexdigest() != entry["sha256"]: problems.append(f"{label} {relative} does not match the prepared hash") checked += 1
for entry in files: verify(entry["path"], entry, "source file") for entry in distribution_docs: verify(entry["archive_path"], entry, "distribution document") for entry in vendor.get("files") or []: verify(f"vendor/{entry['path']}", entry, "vendored file")
if files and aggregate_digest(files) != source_manifest.get("aggregate_sha256"): problems.append("the prepared source aggregate does not match its file list") if vendor.get("files") and aggregate_digest(vendor["files"]) != vendor.get("aggregate_sha256"): problems.append("the prepared vendor aggregate does not match its file list") if problems: raise CheckFailure( f"the prepared source archive {archive_path.name} does not match " f"source-manifest.json:\n" + "\n".join(f" {item}" for item in problems[:20]) ) return {"verified_members": checked, "archive": audit}
def mandatory_artifact_names(release_manifest: dict) -> list[str]: """The artifact names a release manifest must carry, with their states.""" binaries = {entry["name"] for entry in release_manifest["artifacts"] if entry["kind"] == "host-binary"} kinds = {entry["kind"] for entry in release_manifest["artifacts"]} missing = [] for required in BINARY_NAMES: if required not in binaries: missing.append(f"binary {required}") for required in ("source-archive", "binary-archive"): if required not in kinds: missing.append(f"artifact kind {required}") if not release_manifest.get("source_archive", {}).get("name"): missing.append("source archive name") if not release_manifest.get("archive", {}).get("name"): missing.append("binary archive name") return missing
def validate_rebuild_document( evidence: dict, *, expected_archive: dict, expected_aggregate: str, expected_binaries: dict, expected_environment: dict, commands_required: list[list[str]] | None = None,) -> list[str]: """The single, shared contract for a clean-rebuild record.
Returns the problems found. The caller supplies the *facts* it has verified itself — the prepared or shipped archive, the recomputed source aggregate, the measured binaries and the expected environment — so the packager and ``tools/check_release.py`` enforce exactly the same rules against their own independent measurements rather than trusting each other's text. """ problems: list[str] = [] required = [ "schema", "record_kind", "source_archive", "source_aggregate_sha256", "environment", "commands", "outputs", ] missing = [key for key in required if key not in evidence] if missing: return [f"the rebuild record is missing {', '.join(missing)}"] if evidence.get("schema") != SCHEMA or evidence.get("record_kind") != "release_rebuild_evidence": problems.append( f"the rebuild record must be schema {SCHEMA} release_rebuild_evidence; got " f"schema {evidence.get('schema')!r} record_kind {evidence.get('record_kind')!r}" ) archive = evidence.get("source_archive") or {} if archive.get("name") != expected_archive.get("name"): problems.append(f"the rebuild record names source archive {archive.get('name')!r}") if not expected_archive.get("sha256") or archive.get("sha256") != expected_archive.get("sha256"): problems.append("the rebuild record is not bound to the source archive hash") if evidence.get("source_aggregate_sha256") != expected_aggregate: problems.append("the rebuild record is not bound to the source file aggregate") environment = evidence.get("environment") or {} if environment.get("base_image_digest") != expected_environment.get("base_image_digest"): problems.append("the rebuild record was not produced in the pinned base image") if version_token(environment.get("rustc"), "rustc") != expected_environment.get("rust_version"): problems.append(f"the rebuild record does not record rustc {expected_environment.get('rust_version')}") if environment.get("target_triple") != expected_environment.get("target_triple"): problems.append("the rebuild record records a different target triple") commands = evidence.get("commands") or [] if not commands: problems.append("the rebuild record lists no executed commands") for command in commands: if "argv" not in command or "exit_code" not in command: problems.append("a rebuild command entry is missing argv or exit_code") continue if command["exit_code"] != 0: problems.append(f"the rebuild record records exit code {command['exit_code']} for {' '.join(command['argv'])}") if list(command["argv"]) != list(BUILD_COMMAND) and command["argv"][:2] != ["cargo", "build"]: problems.append(f"the rebuild record lists an unexpected command: {' '.join(command['argv'])}") for required_command in commands_required or [BUILD_COMMAND]: if not any(list(command.get("argv") or []) == list(required_command) for command in commands): problems.append(f"the rebuild record does not list the documented command {' '.join(required_command)}") outputs = evidence.get("outputs") or [] if not outputs: problems.append("the rebuild record lists no rebuilt outputs") names = [item.get("name") for item in outputs] duplicates = sorted({name for name in names if names.count(name) > 1 and name}) if duplicates: problems.append("the rebuild record lists duplicate outputs: " + ", ".join(duplicates)) for item in outputs: if not item.get("name") or item.get("sha256") is None or item.get("size") is None: problems.append("a rebuilt output entry is missing name, size or sha256") listed = {item.get("name"): item for item in outputs if item.get("name")} for name, measured in sorted(expected_binaries.items()): shipped = listed.get(name) if not shipped: problems.append(f"the rebuild record lists no output for {name}") continue if shipped.get("sha256") != measured["sha256"]: problems.append(f"the rebuilt {name} does not match the measured hash") if shipped.get("size") != measured["size"]: problems.append(f"the rebuilt {name} does not match the measured size") return problems
def validate_rebuild_evidence( path: Path, *, source_archive: dict, binaries: dict, manifest: dict, limits: Limits = DEFAULT_LIMITS,) -> dict: """Validate rebuild evidence before source completeness may be claimed.
``verified_by_clean_rebuild`` is never taken on the operator's word: the evidence must be a bounded regular file that binds the *prepared* source archive hash, the source aggregate, the pinned environment, the executed commands with their exit codes, and the hashes of the shipped outputs. Anything that does not match fails closed. """ data = read_regular_bytes(path, max_bytes=limits.max_evidence_bytes, description="rebuild evidence") try: evidence = json.loads(data) except json.JSONDecodeError as error: raise CheckFailure(f"{path} is not valid JSON: {error}") from error problems = validate_rebuild_document( evidence, expected_archive=source_archive, expected_aggregate=source_archive.get("aggregate_sha256"), expected_binaries=binaries, expected_environment=manifest["build_environment"], ) if problems: raise CheckFailure( "the clean-rebuild evidence does not satisfy the shared rebuild contract:\n" + "\n".join(f" {item}" for item in problems) ) return {"status": "verified", "evidence_path": str(path), "evidence_sha256": sha256_bytes(data)}
def check_inventory(inventory: list[dict], archived_paths: set[str]) -> list[str]: """Every ``collected`` inventory entry must name material actually shipped.
``archived_paths`` contains the first-party files and the vendored crate files as they appear inside the source archive, so an entry that claims its source and notices are collected must point at real members rather than at a prose statement. """ problems: list[str] = [] for entry in inventory: if entry.get("status") != "collected": continue paths = entry.get("collected_paths") if not paths: problems.append( f"inventory entry {entry.get('component')!r} claims status 'collected' without collected_paths" ) continue for relative in paths: if relative not in archived_paths: problems.append( f"inventory entry {entry.get('component')!r} names {relative}, which is not in the source archive" ) return problems
# ---------------------------------------------------------------------------# Optional runtime corresponding-source bundle# ---------------------------------------------------------------------------## The shipped C64 PRGs link third-party runtime code. tools/collect_runtime_sources.py# collects that runtime's actual corresponding source and notices into a# byte-reproducible archive, and tools/runtime_source_bundle.py verifies it and# binds it to a digest record. The two optional options accepted by build_parser# wire that verified output into a package. None of this approves the coverage# mapping on a human's behalf: the bundled mapping is the explicit, per-component# factory requirement, the entries keep review ``pending_human_license_approval``,# and release readiness stays false.
def runtime_bundle_module(): """Import the bundler lazily; it imports this module, so an eager import would cycle.""" tool_dir = str(Path(__file__).resolve().parent) if tool_dir not in sys.path: sys.path.insert(0, tool_dir) import runtime_source_bundle # noqa: PLC0415 - deferred to break the import cycle
return runtime_source_bundle
def _require_explicit_runtime_coverage(coverage: dict) -> None: """Refuse a coverage mapping that is a generic claim rather than pinned requirements.
The trusted factory mapping must name, per component, the required source prefixes, the collected artifacts or the notice files. A component with none of those is an unaccountable claim, so the hook refuses it instead of treating the report's own "collected" labels as evidence. """ for spec in coverage.get("components", []): if not ( spec.get("required_source_prefixes") or spec.get("required_collected_artifacts") or spec.get("required_notices") ): raise CheckFailure( f"the trusted runtime coverage mapping declares no explicit requirement for " f"{spec.get('key')!r}; a generic claim is not accepted" )
def load_runtime_bundle(args, context: dict) -> dict | None: """Bind the verified runtime source bundle from the two optional options.
Returns ``None`` when neither option is given, so a package built without them behaves exactly as before. The archive is an allowlisted, externally produced operator artifact: it is read only through the shared bounded, no-follow reader, and the collector report's record kind, schema, configured digest, size, member count and tree digest must all match the bytes on disk, so an arbitrary or private copy is refused rather than trusted. """ report_arg = getattr(args, "runtime_sources_report", None) dir_arg = getattr(args, "runtime_sources_dir", None) if report_arg is None and dir_arg is None: return None if report_arg is None or dir_arg is None: raise UsageError( "--runtime-sources-report and --runtime-sources-dir must be given together, or neither" ) rsb = runtime_bundle_module() limits = context["limits"] report_path = report_arg.absolute() collected_dir = dir_arg.absolute() if not collected_dir.is_dir(): raise CheckFailure(f"--runtime-sources-dir {collected_dir} is not a directory") try: report, report_sha256 = rsb.load_report(report_path) archive_path = collected_dir / rsb.archive_filename(report) audit = rsb.audit_archive(archive_path, limits=limits) rsb.bind_report(report, audit) coverage = rsb.validate_coverage(rsb.REQUIRED_COVERAGE) _require_explicit_runtime_coverage(coverage) state = rsb.evaluate_coverage(coverage, audit, report, collected_dir, limits) notice_mapping, notice_problems = rsb.evaluate_notice_mapping(audit, report) notice_payloads = rsb.extract_members(archive_path, audit["notice_members"], limits=limits) except rsb.BundleError as error: raise CheckFailure(f"runtime source bundle: {error}") from error if notice_problems: state["complete"] = False state["problems"].extend(f"notice binding: {problem}" for problem in notice_problems) findings = rsb.license_findings(coverage, report) blocking = [item for item in findings if item["kind"] in ("license_mismatch", "no_obligation")] if blocking: state["complete"] = False state["problems"].extend( f"license coverage: {item['component']}: {item['finding']} " f"(declared {item['declared_license']!r}, expected {item['expected_license']!r})" for item in blocking ) record = rsb.build_bundle_record( report=report, report_name=rsb.display_path(report_path), report_sha256=report_sha256, coverage=coverage, coverage_name=None, coverage_sha256=None, coverage_state=state, findings=findings, notice_mapping=notice_mapping, audit=audit, created_at=context["created_at"], limits=limits, ) record_bytes = json.dumps(record, indent=2, sort_keys=True).encode("utf-8") + b"\n" fragment = rsb.render_fragment(record) patch = rsb.build_inventory_patch(context["manifest"], record, report_name=rsb.display_path(report_path)) return { "rsb": rsb, "record": record, "record_bytes": record_bytes, "fragment": fragment, "audit": audit, "archive_path": archive_path, "notice_payloads": notice_payloads, "state": state, "findings": findings, "patch": patch, "coverage": coverage, "notice_mapping": notice_mapping, "report_name": rsb.display_path(report_path), "report_sha256": report_sha256, }
def runtime_bundle_source_members(bundle: dict, names: dict) -> list[tuple[str, Path | bytes, int]]: """The ``runtime-sources/`` members that ship inside the host source archive.""" rsb = bundle["rsb"] prefix = names["source_prefix"] members: list[tuple[str, Path | bytes, int]] = [ ( f"{prefix}/{rsb.STAGED_SOURCE_DIR}/{bundle['audit']['name']}", bundle["archive_path"], ARCHIVE_MODE_FILE, ), ( f"{prefix}/{rsb.STAGED_SOURCE_DIR}/{rsb.BUNDLE_RECORD_NAME}", bundle["record_bytes"], ARCHIVE_MODE_FILE, ), ( f"{prefix}/{rsb.STAGED_SOURCE_DIR}/{rsb.FRAGMENT_NAME}", bundle["fragment"], ARCHIVE_MODE_FILE, ), ] for member, payload in sorted(bundle["notice_payloads"].items()): members.append((f"{prefix}/{rsb.staged_source_notice_path(member)}", payload, ARCHIVE_MODE_FILE)) return members
def runtime_bundle_manifest_entry(bundle: dict) -> dict: """The portable, digest-bound runtime bundle record written into the manifests.
Only repo-relative member paths, sizes and SHA-256 values are recorded: the collector report path, its ``output_root`` and any probe output are never copied, so the bundle stays portable. """ rsb = bundle["rsb"] audit = bundle["audit"] source_root = rsb.STAGED_SOURCE_DIR source_members = [ { "path": f"{source_root}/{audit['name']}", "sha256": audit["sha256"], "size": audit["size_bytes"], "role": "runtime corresponding-source archive", }, { "path": f"{source_root}/{rsb.BUNDLE_RECORD_NAME}", "sha256": sha256_bytes(bundle["record_bytes"]), "size": len(bundle["record_bytes"]), "role": "digest-bound bundle record", }, { "path": f"{source_root}/{rsb.FRAGMENT_NAME}", "sha256": sha256_bytes(bundle["fragment"]), "size": len(bundle["fragment"]), "role": "SOURCE-AND-LICENSES runtime fragment", }, ] binary_notices = [] for member, payload in sorted(bundle["notice_payloads"].items()): source_members.append( { "path": rsb.staged_source_notice_path(member), "sha256": sha256_bytes(payload), "size": len(payload), "role": "runtime notice (source archive copy)", } ) binary_notices.append( { "path": rsb.staged_binary_notice_path(member), "sha256": sha256_bytes(payload), "size": len(payload), "source_member": rsb.staged_source_notice_path(member), "member": member, } ) state = bundle["state"] return { "present": True, "record_kind": rsb.RECORD_KIND, "role": "verified third-party runtime corresponding source and notices bound into this package", "collector_report": { "name": bundle["report_name"], "sha256": bundle["report_sha256"], "record_kind": (bundle["record"].get("provenance", {}).get("collector_report") or {}).get("record_kind"), }, "archive": { "name": audit["name"], "sha256": audit["sha256"], "size_bytes": audit["size_bytes"], "members": audit["members"], "source_members": audit["source_members"], "tree_digest": audit["tree_digest"], "internal_manifest": audit["internal_manifest"], "internal_manifest_sha256": audit["internal_manifest_sha256"], "notice_members": [ {"member": member, "sha256": audit["member_hashes"][member], "size": len(payload)} for member, payload in sorted(bundle["notice_payloads"].items()) ], }, "source_root": source_root, "binary_notice_root": rsb.STAGED_BINARY_NOTICE_DIR, "bundle_record": f"{source_root}/{rsb.BUNDLE_RECORD_NAME}", "fragment": f"{source_root}/{rsb.FRAGMENT_NAME}", "source_members": source_members, "binary_notices": binary_notices, "notices": bundle["record"].get("notices", []), "notice_mapping": bundle.get("notice_mapping", []), "requirements": bundle["record"].get("coverage", {}).get("requirements", []), "coverage": { "reviewed": state["reviewed"], "complete": state["complete"], "problems": state["problems"], "components": state["components"], "maintainer": bundle["coverage"].get("maintainer"), "legal_review": bundle["coverage"].get("legal_review"), "source": "tools/runtime_source_bundle.py REQUIRED_COVERAGE (explicit per-component factory requirement)", }, "license_findings": bundle["findings"], "source_complete": bool(state["complete"]), "human_review": False, "release_ready": False, "review": "pending_human_license_approval", "runtime_inputs": [ { "name": bundle["report_name"], "sha256": bundle["report_sha256"], "role": "collector report", }, { "name": audit["name"], "sha256": audit["sha256"], "size_bytes": audit["size_bytes"], "members": audit["members"], "tree_digest": audit["tree_digest"], "role": "verified corresponding-source archive", }, ], "inventory_operations": bundle["patch"].get("operations", []), "note": ( "source completeness here is mechanical and bound to the archive digest; human license " "approval and the clean extracted-source rebuild are still outstanding, so release_ready " "stays false" ), }
def resolve_runtime_inventory(base_inventory: list[dict], bundle: dict) -> list[dict]: """Apply the bundler's inventory patch to the static inventory.
The static manifest keeps its ``pending_external`` entries. Only this resolved copy flips an entry to ``collected``, and only when the mapped coverage is satisfied and reviewed; when it is not, the entry keeps its pending status and source completeness stays incomplete. """ patch = bundle["patch"] replacements = patch.get("replacement_entries", {}) resolved = [dict(replacements.get(entry.get("component"), entry)) for entry in base_inventory] present = {entry.get("component") for entry in resolved} for name in patch.get("added_entries", []): if name not in present and name in replacements: resolved.append(dict(replacements[name])) return resolved
def render_runtime_sources_section(context: dict) -> list[str]: """The SOURCE-AND-LICENSES.md section for the runtime bundle, if it is attached.""" entry = (context.get("source_manifest") or {}).get("runtime_bundle") or {} if not entry.get("present"): return [] archive = entry["archive"] coverage = entry["coverage"] lines = [ "## Third-party runtime corresponding source and notices", "", "The shipped C64 PRGs link third-party runtime code. Its actual corresponding source and", "the license and notice text ship with this release, bound to the digests below:", "", f"- Archive: `{entry['source_root']}/{archive['name']}` (inside the source archive)", f"- SHA-256: `{archive['sha256']}`", f"- Bytes: {archive['size_bytes']}", f"- Members: {archive['members']} ({archive['source_members']} source files plus " f"`{archive['internal_manifest']}`)", f"- Tree digest: `{archive['tree_digest']}`", f"- Binding record: `{entry['bundle_record']}`", f"- Binary notices: `{entry['binary_notice_root']}/` beside the binaries", "", f"Coverage status: **{'satisfied' if coverage['complete'] else 'incomplete'}** " f"(mapping reviewed: {'yes' if coverage['reviewed'] else 'no'}). This is a mechanical, " "digest-bound statement, not a human license approval: legal review is " f"{coverage.get('legal_review') or 'pending'} and the clean extracted-source rebuild is " "still outstanding.", "", ] if coverage["components"]: lines += ["| Component | Scope | Status |", "| --- | --- | --- |"] lines.extend( f"| {item['component']} | {item['scope']} | {item['status']} |" for item in coverage["components"] ) lines.append("") return lines
def prepare_source(args, context: dict) -> dict: manifest = context["manifest"] source_root = context["source_root"] output = context["output"] names = context["names"] limits = context["limits"] files, selection_report = collect_source_files(source_root, manifest, limits) example_snapshot = check_example_snapshot_selection( source_root, manifest, [entry["path"] for entry in files], limits ) privacy = scan_source_tree(source_root, files, context["policy"], context["exemptions"], limits) context["privacy"] = privacy distribution_docs = distribution_doc_members(source_root, manifest, limits) # The distribution copies are not in the source file list, so the sentinel # policy is applied to the exact bytes each one ships, recorded against the # archive path that carries them, and merged into the same privacy record. distribution_findings = scan_distribution_document_privacy( distribution_docs, context["policy"], context["exemptions"] ) merge_privacy_findings(privacy, distribution_findings) vendor_record = {"status": "not_included", "command": None, "file_count": 0, "total_bytes": 0} vendor_members: list[tuple[str, Path, int]] = [] if args.no_vendor: print("warning: --no-vendor was given; the source archive is explicitly incomplete", file=sys.stderr) elif args.vendor_dir: provided = args.vendor_dir.absolute() vendor_record = collect_vendor(provided, manifest, context["policy"], context["exemptions"], limits, context["vendor_fixture_exemptions"]) vendor_record["status"] = "provided" vendor_record["command"] = None vendor_record["directory"] = provided.name vendor_record["notes"] = "operator-supplied vendor directory; not observed cargo vendor output" vendor_members = [ (f"{names['source_prefix']}/vendor/{item['path']}", provided / item["path"], ARCHIVE_MODE_FILE) for item in vendor_record["files"] ] else: if args.dry_run: vendor_record = {"status": "would_run_cargo_vendor", "file_count": 0, "total_bytes": 0} else: vendor_dir = output / "vendor" command, snippet = run_cargo_vendor(source_root, vendor_dir, args.vendor_timeout) # cargo echoes the directory it was handed, which here is the host # output path; the archived configuration must name the directory the # extracted archive actually ships. snippet, config_provenance = normalize_generated_vendor_config(snippet, vendor_dir) vendor_record = collect_vendor(vendor_dir, manifest, context["policy"], context["exemptions"], limits, context["vendor_fixture_exemptions"]) vendor_record.update( { "status": "generated_cargo_vendor", "command": command.split(), "directory": "vendor", "config_snippet": snippet.strip(), "config_provenance": config_provenance, "config_snippet_source": ( "observed cargo vendor output, with the vendored source directory rewritten to the " "archive-relative 'vendor'" ), } ) vendor_members = [ (f"{names['source_prefix']}/vendor/{item['path']}", vendor_dir / item["path"], ARCHIVE_MODE_FILE) for item in vendor_record["files"] ]
vendor_findings = vendor_record.get("privacy_findings") or [] if vendor_findings: privacy["findings"].extend(vendor_findings) privacy["review_required_findings"] += len( [item for item in vendor_findings if item["severity"] == "review_required"] ) privacy["blocking_exemption_count"] += vendor_record.get("blocking_exemption_count", 0) privacy["vendored_fixture_pending_review"] = vendor_record.get("pending_review_findings", 0) privacy["exemptions"] = [item for item in privacy["findings"] if item["severity"].startswith("exempt")]
bundle = context.get("runtime_bundle") runtime_entry = runtime_bundle_manifest_entry(bundle) if bundle is not None else None inventory = manifest["inventory"] if runtime_entry is not None: # Only the resolved copy is flipped; the static manifest is never edited. inventory = resolve_runtime_inventory(inventory, bundle) # The same rule for the vendored crate sources: the static manifest keeps its # conditional_on_vendor entry and only an observed cargo vendor run resolves # the copy. A supplied tree leaves the entry pending. inventory = resolve_vendor_inventory(inventory, vendor_record) archived_paths = {entry["path"] for entry in files} archived_paths |= {entry["archive_path"] for entry in distribution_docs} archived_paths |= {f"vendor/{entry['path']}" for entry in vendor_record.get("files") or []} if runtime_entry is not None: archived_paths |= {item["path"] for item in runtime_entry["source_members"]} inventory_problems = check_inventory(inventory, archived_paths) if inventory_problems: raise CheckFailure("the third-party inventory does not match the archive:\n" + "\n".join(f" {item}" for item in inventory_problems)) pending_inventory = [ f"{entry['component']} ({entry['status']}; retrieval: {entry.get('retrieval', 'not recorded')})" for entry in inventory if entry["status"] in ("pending_external", "conditional_on_vendor") ] if vendor_record["status"] != "generated_cargo_vendor" and vendor_record["status"] != "provided": completeness = "incomplete_missing_vendored_crate_sources" elif any(entry["status"] in ("pending_external", "conditional_on_vendor") for entry in inventory): # A supplied vendor tree that never resolved the conditional entry keeps # that entry pending, so the notices state stays incomplete. completeness = "incomplete_pending_third_party_notices" else: completeness = "complete_pending_clean_rebuild_verification"
provenance = { "schema": SCHEMA, "record_kind": "release_source_provenance", "product": PRODUCT, "version": context["version"], "development": context["development"], "development_label": context["development_label"], "source": context["source"], "manifest": { "path": SOURCE_MANIFEST_RELATIVE, "sha256": sha256_file(context["manifest_path"]), }, "files": { "count": len(files), "total_bytes": sum(item["size"] for item in files), "aggregate_sha256": aggregate_digest(files), }, "vendor": {key: value for key, value in vendor_record.items() if key not in ("files", "privacy_findings")}, "distribution_docs": [ {key: value for key, value in entry.items() if key != "bytes"} for entry in distribution_docs ], "privacy": { "findings": privacy["findings"], "policy": privacy["policy"], "blocking_findings": privacy["blocking_findings"], }, "source_completeness": { "status": completeness, "clean_rebuild": "requested" if args.rebuild_status == "verified" else "not_verified", }, "runtime_bundle": ( { "present": True, "archive_name": runtime_entry["archive"]["name"], "archive_sha256": runtime_entry["archive"]["sha256"], "members": runtime_entry["archive"]["members"], "source_complete": runtime_entry["source_complete"], "human_review": runtime_entry["human_review"], "release_ready": runtime_entry["release_ready"], } if runtime_entry is not None else {"present": False} ), "created_at": context["created_at"], "generator": context["generator"], }
members: list[tuple[str, Path | bytes, int]] = [] for entry in files: path = source_root / entry["path"] mode = ARCHIVE_MODE_EXECUTABLE if entry["executable"] else ARCHIVE_MODE_FILE members.append((f"{names['source_prefix']}/{entry['path']}", path, mode)) # Each reviewed sanitized copy travels at its own source path (it is an # allowlisted first-party file above) and is aliased here to the archive path # of the withheld original it replaces, so the documentation's relative links # and anchors resolve inside the source archive too. for entry in distribution_docs: members.append( (f"{names['source_prefix']}/{entry['archive_path']}", entry["bytes"], ARCHIVE_MODE_FILE) ) members.extend(vendor_members) generated_files = ["SOURCE-PROVENANCE.json"] + (["vendor-sources.toml"] if vendor_members else []) if runtime_entry is not None: members.extend(runtime_bundle_source_members(bundle, names)) generated_files += [item["path"] for item in runtime_entry["source_members"]] if vendor_members: snippet = vendor_record.get("config_snippet") if snippet: vendor_record.setdefault("config_snippet_source", "cargo vendor") else: snippet = ( "[source.crates-io]\n" 'replace-with = "vendored-sources"\n\n' "[source.vendored-sources]\n" 'directory = "vendor"\n' ) vendor_record["config_snippet_source"] = ( "standard crates.io replacement written for an operator-supplied vendor directory" ) # Last gate before the bytes are written, for every origin of the snippet: # the shipped configuration must name the archive-relative directory. assert_portable_vendor_config(snippet) members.append( (f"{names['source_prefix']}/vendor-sources.toml", snippet.encode("utf-8"), ARCHIVE_MODE_FILE) ) members.append( ( f"{names['source_prefix']}/SOURCE-PROVENANCE.json", json.dumps(provenance, indent=2, sort_keys=True).encode("utf-8") + b"\n", ARCHIVE_MODE_FILE, ) )
source_manifest = { "schema": SCHEMA, "record_kind": "release_source_manifest_resolved", "role": "resolved inventory of the corresponding source archive; tools/check_release.py re-verifies it", "created_at": context["created_at"], "generator": context["generator"], "product": PRODUCT, "version": context["version"], "development": context["development"], "development_label": context["development_label"], "source": context["source"], "manifest": {"path": SOURCE_MANIFEST_RELATIVE, "sha256": sha256_file(context["manifest_path"])}, "selection": { "include_entries": selection_report["include_entries"], "files": selection_report, "rules": manifest["source_paths"]["rules"], }, "files": files, "file_count": len(files), "total_bytes": sum(item["size"] for item in files), "aggregate_sha256": aggregate_digest(files), "example_snapshot": example_snapshot, "vendor": vendor_record, "distribution_docs": [ {key: value for key, value in entry.items() if key != "bytes"} for entry in distribution_docs ], "privacy": privacy, "inventory": inventory, "runtime_bundle": runtime_entry if runtime_entry is not None else {"present": False}, "external_prerequisites": manifest["external_prerequisites"], "source_completeness": { "status": completeness, "clean_rebuild": "requested" if args.rebuild_status == "verified" else "not_verified", "clean_rebuild_evidence": str(args.rebuild_evidence) if args.rebuild_evidence else None, "pending_inventory": pending_inventory, "generated_by": "tools/package_release.py --prepare-source", }, "archive": { "name": names["source_archive"], "prefix": names["source_prefix"], "sha256": None, "recorded_in": "the release directory and SHA256SUMS, never inside the archive itself", "members": len(members), "generated_files": generated_files, }, }
# The size and SHA-256 recorded for each member are the contract the archive # must satisfy. ``write_archive`` re-verifies the bytes it streams from every # input against this map, so a file that changes between hashing and archive # writing fails the preparation instead of shipping unmatched content. expected_members: dict[str, tuple[int, str]] = {} for entry in files: expected_members[f"{names['source_prefix']}/{entry['path']}"] = (entry["size"], entry["sha256"]) for entry in distribution_docs: expected_members[f"{names['source_prefix']}/{entry['archive_path']}"] = ( entry["size"], entry["sha256"], ) for item in vendor_record.get("files") or []: expected_members[f"{names['source_prefix']}/vendor/{item['path']}"] = (item["size"], item["sha256"]) if runtime_entry is not None: for item in runtime_entry["source_members"]: expected_members[f"{names['source_prefix']}/{item['path']}"] = (item["size"], item["sha256"])
if args.dry_run: print(f"dry run: would write {names['source_archive']} with {len(members)} members") else: archive_info = write_archive(output / names["source_archive"], members, limits, expected_members) source_manifest["archive"].update( {key: archive_info[key] for key in ("sha256", "size", "members", "tar_safety", "determinism")} ) source_manifest["archive"]["uncompressed_bytes"] = archive_info["uncompressed_bytes"] write_text(output / "source-manifest.json", json.dumps(source_manifest, indent=2, sort_keys=True) + "\n") return source_manifest
def write_text(path: Path, text: str) -> None: with open(path, "w", encoding="utf-8") as handle: handle.write(text)
def pack(args, context: dict, source_manifest: dict) -> dict: manifest = context["manifest"] names = context["names"] output = context["output"] limits = context["limits"] binaries = verify_binaries(context["binary_dir"], manifest) runtime_entry = source_manifest.get("runtime_bundle") or {"present": False} distribution_docs = distribution_doc_members(context["source_root"], manifest, limits)
# Files that ship inside the binary archive (examples) go through the same # selection and privacy rules as the source tree, so a hidden private file # beside a reviewed example can never be distributed. shipped_members: list[tuple[str, Path | bytes, int]] = [] shipped_findings: list[dict] = [] shipped_relatives: list[str] = [] for relative in manifest["package_layout"]["shipped_examples"]: selected = select_shipped_files(context["source_root"], relative, manifest, limits) shipped_relatives.extend(selected) shipped_findings.extend( scan_files_for_privacy( context["source_root"], selected, context["policy"], context["exemptions"], limits ) ) for shipped_relative in selected: shipped_members.append( (f"{names['prefix']}/{shipped_relative}", context["source_root"] / shipped_relative, ARCHIVE_MODE_FILE) ) # The binary archive ships the same reviewed prebuilt snapshot as the source # archive, so it is checked against the declared payload too: the approved # inputs travel at their true paths and the local materialized ``build/`` # tree is never packaged. example_snapshot = check_example_snapshot_selection( context["source_root"], manifest, shipped_relatives, limits ) if shipped_findings: context["privacy"]["findings"].extend(shipped_findings) context["privacy"]["review_required_findings"] += len( [item for item in shipped_findings if item["severity"] == "review_required"] ) context["privacy"]["exemptions"] = [ item for item in context["privacy"]["findings"] if item["severity"].startswith("exempt") ] # The binary archive re-runs the sentinel policy over the exact bytes it # writes for every sanitized distribution document, so the copies are bound # by the same policy in this archive, not only in the source archive that # recorded them. Findings already recorded are counted once. merge_privacy_findings( context["privacy"], scan_distribution_document_privacy( distribution_docs, context["policy"], context["exemptions"] ), ) environment = validate_build_environment(context["build_env"], manifest, context["development"]) context["binaries"] = binaries context["environment"] = environment
source_archive_path = output / names["source_archive"] if not args.dry_run and not source_archive_path.exists(): raise CheckFailure( f"{names['source_archive']} is missing from {output}; run --prepare-source (or pass both flags) " "before packing, because the binaries and their corresponding source ship together" ) if args.dry_run: source_archive_info = { "name": names["source_archive"], "sha256": None, "size": None, "members": None, "tar_safety": "not_checked_in_dry_run", } else: source_archive_info = verify_prepared_source(source_archive_path, source_manifest, limits) source_archive_info = {**source_archive_info["archive"], "verified_members": source_archive_info["verified_members"]} if source_manifest.get("archive", {}).get("sha256") and source_archive_info.get("sha256"): if source_manifest["archive"]["sha256"] != source_archive_info["sha256"]: raise CheckFailure("source-manifest.json does not describe the source archive it was packed with") if not source_manifest.get("archive", {}).get("sha256") and not args.dry_run: source_manifest["archive"]["sha256"] = source_archive_info["sha256"]
completeness = source_manifest["source_completeness"] if args.rebuild_status == "verified": rebuild = validate_rebuild_evidence( args.rebuild_evidence.absolute(), source_archive={ "name": names["source_archive"], "sha256": source_archive_info.get("sha256"), "aggregate_sha256": source_manifest.get("aggregate_sha256"), }, binaries=binaries, manifest=manifest, limits=limits, ) completeness["clean_rebuild"] = "verified" evidence_bytes = read_regular_bytes( args.rebuild_evidence.absolute(), max_bytes=limits.max_evidence_bytes, description="rebuild evidence" ) # The registered copy is a portable file inside the release directory, so # the checker can re-read and re-hash it. No absolute path is recorded. completeness.update(register_evidence(output, evidence_bytes, limits)) completeness["clean_rebuild_evidence"] = REGISTERED_EVIDENCE_NAME if not completeness.get("pending_inventory") and completeness["status"] == ( "complete_pending_clean_rebuild_verification" ): completeness["status"] = "verified_by_clean_rebuild" if not args.dry_run: write_text( output / "source-manifest.json", json.dumps(source_manifest, indent=2, sort_keys=True) + "\n", ) context["source_manifest"] = source_manifest context["source_completeness"] = completeness context["release"] = manifest["release"] context["runtime_bundle"] = runtime_entry
context["archive"] = {"name": names["archive"], "sha256": None} context["source_archive"] = { "name": names["source_archive"], "sha256": source_archive_info.get("sha256"), "size": source_archive_info.get("size"), } context["inventory"] = source_manifest.get("inventory") or manifest["inventory"] context["external_prerequisites"] = manifest["external_prerequisites"]
generated_readme = render_readme(context) generated_install = render_install(context) generated_source = render_source_and_licenses(context)
checks: list[dict] = []
def record(name: str, state: str, detail: str) -> None: checks.append({"check": name, "state": state, "detail": detail})
record( "binary_elf_target", "verified", "; ".join( f"{name}: {record_['class']} {record_['machine']} {record_['type']} " f"linkage={record_['linkage']} interpreter={record_['interpreter'] or 'none'}" for name, record_ in sorted(binaries.items()) ), ) baseline = manifest["build_environment"]["glibc"] record( "glibc_baseline", "verified", "; ".join( f"{name}: requires {record_['glibc_max_required'] or 'no versioned symbols'} " f"within baseline {baseline}" for name, record_ in sorted(binaries.items()) ), ) environment_state = "verified" if environment["status"] == "verified" else "not_verified" record( "build_environment_observed", environment_state, environment["status"] + ("" if not environment["problems"] else ": " + "; ".join(environment["problems"])[:600]), ) record( "pinned_base_image", "verified" if environment["status"] == "verified" else "pending", f"expected {manifest['build_environment']['base_image']}@" f"{manifest['build_environment']['base_image_digest']}", ) record("archive_sha256", "verified", "archives are hashed from their bytes; SHA256SUMS is written last") record("tar_member_safety", "verified", "no absolute path, traversal, symlink or non-regular member") record( "privacy_sentinels", "verified" if not context["privacy"]["review_required_findings"] else "pending", f"{context['privacy']['files_scanned']} files scanned, 0 blocking findings, " f"{context['privacy']['review_required_findings']} review-required findings, " f"{context['privacy'].get('blocking_exemption_count', 0)} reviewed exemption(s) over blocking content", ) record("source_manifest_paths", "verified", f"{source_manifest['file_count']} allowlisted files hashed") record( "distribution_documents", "verified", "; ".join( f"{entry['archive_path']} from {entry['copy']} ({entry['sha256'][:12]}…)" for entry in distribution_docs ) or "no sanitized distribution documents are mapped", ) vendor_status = source_manifest["vendor"]["status"] record( "vendor_sources_present", "verified" if vendor_status in ("generated_cargo_vendor", "provided") else "pending", f"vendor tree status: {vendor_status}", ) tag = context["source"].get("tag") record( "annotated_version_tag", "verified" if tag and context["source"]["tree_state"] == "clean_tagged" else "pending", f"tag {tag!r} on commit {context['source'].get('revision')}" if tag else "no annotated version tag given", ) record( "clean_worktree", "verified" if context["source"]["dirty"] is False else "pending", "clean" if context["source"]["dirty"] is False else f"dirty: {len(context['source']['dirty_paths'])} path(s)", ) reserved = manifest["release"]["reserved_preview_version"] record( "reserved_preview_version", "verified" if context["version"] == reserved else "pending", f"workspace version {context['version']} (reserved preview version {reserved})", ) pending_inventory = completeness.get("pending_inventory", []) record( "third_party_notices_collected", "pending" if pending_inventory else "verified", "; ".join(item.split(" (")[0] for item in pending_inventory) or "all inventory entries collected", ) record( "clean_extracted_source_rebuild", "verified" if completeness.get("clean_rebuild") == "verified" else "not_verified", "requires the prepared toolchain environment and an extracted-source rebuild; " "this scaffolding does not perform it", ) if runtime_entry.get("present"): record( "runtime_source_bundle", "verified" if runtime_entry.get("source_complete") else "not_verified", f"{runtime_entry['archive']['name']} sha256 {runtime_entry['archive']['sha256'][:16]}…, " f"{runtime_entry['archive']['members']} members, " f"coverage complete={runtime_entry['coverage']['complete']}, human_review=false", ) else: record( "runtime_source_bundle", "not_applicable", "no runtime corresponding-source bundle was attached; the third-party runtime inventory stays pending", )
blockers: list[str] = [] if context["development"]: blockers.append(f"development package labeled {context['development_label']!r}") if context["source"]["dirty"] is False and not tag: blockers.append("the source tree is untagged") if context["source"]["dirty"]: blockers.append("the source tree is dirty") if environment["status"] != "verified": blockers.append("the build environment is not verified as the pinned environment") if pending_inventory: blockers.append("third-party source/notice inventory is incomplete") if context["privacy"]["review_required_findings"]: blockers.append("privacy findings need a recorded human review") if shipped_findings: blockers.append( "shipped example or documentation files carry " f"{len([item for item in shipped_findings if item['severity'] != 'exempt'])} recorded " "privacy finding(s) that a reviewer must accept" ) if context["privacy"].get("vendored_fixture_pending_review"): blockers.append( "public upstream fixture exemptions await human review: " + "; ".join( f"{item['path']} [{item['sentinel']}]" for item in context["privacy"].get("findings", []) if item["severity"] == "exempt_vendored_public_fixture" )[:400] ) if context["privacy"].get("blocking_exemption_count"): blockers.append( "a reviewed exemption covers blocking sentinel content: " + "; ".join(f"{item['path']} [{item['sentinel']}]" for item in context["privacy"]["blocking_exemptions"]) ) if context["version"] != reserved: blockers.append(f"workspace version {context['version']} is not the reserved preview version") if completeness.get("clean_rebuild") != "verified": blockers.append("no clean extracted-source rebuild is recorded") if runtime_entry.get("present"): blockers.append( "the runtime corresponding-source bundle is mechanically complete but the human license " "review is still outstanding" )
release_manifest = { "schema_version": SCHEMA, "record_kind": "release_manifest", "product": PRODUCT, "product_title": manifest["release"]["product_title"], "version": context["version"], "release_type": manifest["release"]["release_type"], "development": context["development"], "development_label": context["development_label"], "created_at": context["created_at"], "target": { "platform": "linux", "arch": "x86_64", "abi": "gnu", "target_triple": manifest["build_environment"]["target_triple"], "os": f"{manifest['build_environment']['os']} {manifest['build_environment']['os_version_id']}", "glibc_baseline": baseline, "glibc_max_required": { name: record_["glibc_max_required"] for name, record_ in sorted(binaries.items()) }, }, "baseline": { "os": manifest["build_environment"]["os"], "version_id": manifest["build_environment"]["os_version_id"], "glibc": baseline, "target_triple": manifest["build_environment"]["target_triple"], "base_image": manifest["build_environment"]["base_image"], "base_image_digest": manifest["build_environment"]["base_image_digest"], "base_image_pinned": manifest["build_environment"]["base_image_digest_status"] == "pinned", "verified_on": environment["status"], }, "source": { "revision": context["source"]["revision"], "in_git_worktree": context["source"]["in_git_worktree"], "dirty": context["source"]["dirty"], "dirty_paths": context["source"]["dirty_paths"], "tree_state": context["source"]["tree_state"], "tag": context["source"]["tag"], "tag_type": context["source"]["tag_type"], "tag_object": context["source"]["tag_object"], "tag_commit": context["source"]["tag_commit"], "source_archive": names["source_archive"], "source_manifest": "source-manifest.json", }, "build": { "command": manifest["release"]["build_command"], "environment_status": environment["status"], "environment_problems": environment["problems"], "environment": environment["environment"], "expected_environment": manifest["build_environment"], "compiler_reproducibility": "not_claimed", }, "release_policy": { "reserved_preview_version": manifest["release"]["reserved_preview_version"], "release_type": manifest["release"]["release_type"], "license": manifest["release"]["license"], "license_file": manifest["release"]["license_file"], "required_checks": manifest["checks"]["required_binary_checks"], "release_ready_requires": manifest["checks"]["release_ready_requires"], "supported_versions": [manifest["release"]["reserved_preview_version"]], }, "support_boundary": manifest["release"]["support_boundary"], "artifacts": [ { "name": name, "kind": "host-binary", "path": f"bin/{name}", "sha256": record_["sha256"], "size": record_["size"], "measured_from": record_.get("path"), "elf": { key: value for key, value in record_.items() if key not in ("path", "sha256", "size", "problems") }, } for name, record_ in sorted(binaries.items()) ] + [ { "name": source_archive_info["name"], "kind": "source-archive", "path": source_archive_info["name"], "sha256": source_archive_info.get("sha256"), "size": source_archive_info.get("size"), "always_distributed_with_binaries": True, }, { "name": names["archive"], "kind": "binary-archive", "path": names["archive"], "sha256": None, "self_reference": "the hash of an archive cannot be recorded inside itself; see SHA256SUMS", }, ], "archive": { "name": names["archive"], "prefix": names["prefix"], "sha256": None, "size": None, "members": None, "tar_safety": None, "determinism": "sorted members, mtime 0, uid/gid 0, gzip mtime 0", "recorded_in": "SHA256SUMS in the release directory, never inside the archive itself", }, "source_archive": { "name": source_archive_info["name"], "prefix": names["source_prefix"], "sha256": source_archive_info.get("sha256"), "size": source_archive_info.get("size"), "members": source_archive_info.get("members"), "tar_safety": source_archive_info.get("tar_safety"), }, "source_completeness": completeness, "inventory": manifest["inventory"], "example_snapshot": example_snapshot, "privacy": context["privacy"], "limits": asdict(limits), "verification": checks, "release_ready": not blockers, "publication": { "status": "not_publishable" if blockers else "preflight_required", "blocking": blockers, "preflight": { "status": "not_run", "command": "python3 tools/check_release.py --release-dir <release-directory> --mode release", "note": "release_ready records the packager's own check results; publication also requires the " "independent preflight to pass", }, "publisher_interface": { "consumer": "tools/publish_tangled_artifacts.py (separate explicit publication command)", "fields": [ "version", "source.tag", "source.tag_object", "source.tag_commit", "source.revision", "support_boundary", "artifacts[].name", "artifacts[].kind", "artifacts[].path", "artifacts[].sha256", "artifacts[].size", "source_archive.name", "source_archive.sha256", "verification[]", "release_ready", ], "note": "the publisher must reject any manifest whose release_ready is false or whose " "verification list contains a pending or not_verified state", }, }, "runtime_requirements": manifest["release"]["runtime_requirements"], "generator": context["generator"], }
missing_names = mandatory_artifact_names(release_manifest) if missing_names: raise CheckFailure( "the release manifest does not name the mandatory artifacts: " + ", ".join(missing_names) ) checks.append( { "check": "mandatory_artifacts", "state": "verified", "detail": f"both host binaries, the source archive, the binary archive and both manifests are named; " f"the source archive carries {source_archive_info.get('verified_members')} verified members", } )
members = [] for name, record_ in sorted(binaries.items()): mode = ARCHIVE_MODE_EXECUTABLE if os.access(context["binary_dir"] / name, os.X_OK) else ARCHIVE_MODE_FILE members.append((f"{names['prefix']}/bin/{name}", context["binary_dir"] / name, mode)) members.append((f"{names['prefix']}/LICENSE", context["source_root"] / manifest["release"]["license_file"], ARCHIVE_MODE_FILE)) members.append((f"{names['prefix']}/README.md", generated_readme, ARCHIVE_MODE_FILE)) members.append((f"{names['prefix']}/INSTALL.md", generated_install, ARCHIVE_MODE_FILE)) members.append((f"{names['prefix']}/SOURCE-AND-LICENSES.md", generated_source, ARCHIVE_MODE_FILE)) distribution_by_path = {entry["archive_path"]: entry for entry in distribution_docs} for relative in manifest["package_layout"]["shipped_docs"]: entry = distribution_by_path.get(relative) if entry is not None: # A withheld repository document ships only as its reviewed sanitized # copy, placed at the archive path the original occupies. members.append((f"{names['prefix']}/{relative}", entry["bytes"], ARCHIVE_MODE_FILE)) else: members.append( (f"{names['prefix']}/{relative}", context["source_root"] / relative, ARCHIVE_MODE_FILE) ) members.extend(shipped_members) if runtime_entry.get("present") and not args.dry_run: # The binary archive carries the notice text beside the binaries; the bytes # are read back out of the already-verified source archive and re-hashed, so # pack never needs the collected directory and never ships the report. for item in runtime_entry.get("binary_notices", []): member_name = f"{names['source_prefix']}/{item['source_member']}" payload = read_archive_bytes(source_archive_path, member_name, limits) if sha256_bytes(payload) != item["sha256"]: raise CheckFailure( f"the runtime notice {item['source_member']} does not match the digest recorded for it" ) members.append((f"{names['prefix']}/{item['path']}", payload, ARCHIVE_MODE_FILE))
in_archive_manifest = json.loads(json.dumps(release_manifest)) in_archive_manifest["archive"]["sha256"] = None in_archive_manifest["archive"]["self_reference"] = ( "not recorded inside the archive that contains it; use SHA256SUMS in the release directory" ) in_archive_source_manifest = json.loads(json.dumps(source_manifest)) in_archive_source_manifest["archive"]["sha256"] = source_archive_info.get("sha256") in_archive_source_manifest["archive"]["self_reference"] = ( "this file describes the source archive, not the binary archive that carries this copy" ) members.append( (f"{names['prefix']}/release-manifest.json", json.dumps(in_archive_manifest, indent=2, sort_keys=True).encode("utf-8") + b"\n", ARCHIVE_MODE_FILE) ) members.append( (f"{names['prefix']}/source-manifest.json", json.dumps(in_archive_source_manifest, indent=2, sort_keys=True).encode("utf-8") + b"\n", ARCHIVE_MODE_FILE) )
generated_blobs = { "README.md": generated_readme, "INSTALL.md": generated_install, "SOURCE-AND-LICENSES.md": generated_source, } generated_findings = [] for name, blob in generated_blobs.items(): for identifier in scan_bytes(blob): generated_findings.append({"path": f"generated:{name}", "sentinel": identifier}) if generated_findings: raise CheckFailure( "a generated distribution document matched a privacy sentinel: " + ", ".join(f"{item['path']} [{item['sentinel']}]" for item in generated_findings) )
if args.dry_run: print(f"dry run: would write {names['archive']} with {len(members)} members") release_manifest["archive"]["members"] = len(members) return release_manifest
archive_info = write_archive(output / names["archive"], members, limits) release_manifest["archive"].update( {key: archive_info[key] for key in ("sha256", "size", "members", "tar_safety")} ) release_manifest["archive"]["uncompressed_bytes"] = archive_info["uncompressed_bytes"] release_manifest["archive"]["member_kinds"] = archive_info["member_kinds"] for artifact in release_manifest["artifacts"]: if artifact["kind"] == "binary-archive": artifact["sha256"] = archive_info["sha256"] artifact.pop("self_reference", None) write_text(output / "release-manifest.json", json.dumps(release_manifest, indent=2, sort_keys=True) + "\n") write_sha256sums( output, [ names["archive"], names["source_archive"], "release-manifest.json", "source-manifest.json", REGISTERED_EVIDENCE_NAME, ], ) return release_manifest
def write_sha256sums(output: Path, names: list[str]) -> None: lines = [] for name in names: path = output / name if not path.exists(): continue lines.append(f"{sha256_file(path)} {name}") write_text(output / "SHA256SUMS", "\n".join(lines) + "\n")
# ---------------------------------------------------------------------------# Entry point# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--source-root", type=Path, default=None, help="repository root (default: this repository)") parser.add_argument("--output", type=Path, required=True, help="new or empty directory for the release files") parser.add_argument("--prepare-source", action="store_true", help="write the corresponding source archive") parser.add_argument("--pack", action="store_true", help="verify the built binaries and write the release") parser.add_argument("--binary-dir", type=Path, default=None, help="directory holding the built binaries") parser.add_argument("--build-env", type=Path, default=None, help="observed build-environment descriptor") parser.add_argument("--source-manifest", type=Path, default=None, help="release source manifest") parser.add_argument("--tag", default=None, help="annotated version tag to bind the package to") parser.add_argument("--development", default=None, metavar="LABEL", help="development label; mandatory for a dirty or untagged tree") parser.add_argument("--vendor-dir", type=Path, default=None, help="reuse an existing cargo vendor directory") parser.add_argument( "--runtime-sources-report", type=Path, default=None, metavar="PATH", help="optional collector report for the verified runtime corresponding-source bundle", ) parser.add_argument( "--runtime-sources-dir", type=Path, default=None, metavar="PATH", help="optional directory holding the verified runtime corresponding-source archive", ) parser.add_argument("--no-vendor", action="store_true", help="omit vendored crate sources and record the archive as incomplete") parser.add_argument("--vendor-timeout", type=int, default=900, help="seconds allowed for cargo vendor") parser.add_argument("--rebuild-status", choices=("verified", "not_verified"), default="not_verified") parser.add_argument("--rebuild-evidence", type=Path, default=None) parser.add_argument( "--review-vendor-fixtures", type=Path, default=None, metavar="DIR", help="review a vendored tree: report every sentinel hit with exact file hashes to --output", ) parser.add_argument("--dry-run", action="store_true", help="validate and report without writing anything") parser.add_argument( "--max-members", type=int, default=None, metavar="N", help="test-only bound: refuse to write an archive with more than N members", ) parser.add_argument("--json", action="store_true", help="print a machine-readable summary") return parser
def run(argv: list[str]) -> int: parser = build_parser() args = parser.parse_args(argv) if args.review_vendor_fixtures is None and not args.prepare_source and not args.pack: raise UsageError("choose a mode: --prepare-source, --pack, or --review-vendor-fixtures DIR") if args.tag and args.development: raise UsageError("--tag and --development are mutually exclusive: a tagged tree is not a development package") source_root = (args.source_root or Path(__file__).resolve().parent.parent).resolve() if not (source_root / "Cargo.toml").exists(): raise UsageError(f"{source_root} does not look like the repository root (no Cargo.toml)") manifest_path = (args.source_manifest or (source_root / SOURCE_MANIFEST_RELATIVE)).resolve() manifest = load_source_manifest(manifest_path) if args.vendor_dir and args.no_vendor: raise UsageError("--vendor-dir and --no-vendor are mutually exclusive")
limits = Limits() if args.max_members is not None: if args.max_members < 1: raise UsageError("--max-members must be a positive integer") limits = Limits(max_members=args.max_members) if args.review_vendor_fixtures: review = review_vendor_fixtures(args.review_vendor_fixtures.absolute(), limits) payload = json.dumps(review, indent=2, sort_keys=True) + "\n" target = args.output.absolute() if not args.dry_run: if target.exists() or target.is_symlink(): raise CheckFailure(f"refusing to overwrite the existing file {target.name}") target.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=target.parent, prefix=".pack-review-", delete=False) as handle: staged = Path(handle.name) try: staged.write_text(payload, encoding="utf-8") os.link(staged, target) finally: staged.unlink(missing_ok=True) summary = { "mode": "review-vendor-fixtures", "vendor_dir": str(args.review_vendor_fixtures), "entries": len(review["entries"]), "crates": len({entry["upstream"] for entry in review["entries"]}), "output": str(target), "document_sha256": sha256_bytes(payload.encode("utf-8")), } print(json.dumps(summary, indent=2, sort_keys=True) if args.json else f"{summary['entries']} vendored sentinel hits in {summary['crates']} crates; " f"review document SHA-256 {summary['document_sha256']}") return EXIT_OK
version_info = read_workspace_version(source_root) if version_info["license"] and version_info["license"] != manifest["release"]["license"]: raise CheckFailure( f"the workspace license {version_info['license']!r} does not match the release manifest " f"license {manifest['release']['license']!r}" ) development_label = sanitize_label(args.development) if args.development else None development = development_label is not None
if args.rebuild_status == "verified" and not args.rebuild_evidence: raise UsageError("--rebuild-status verified requires --rebuild-evidence PATH") if args.rebuild_evidence and not args.pack: raise UsageError( "--rebuild-evidence needs --pack as well, because the evidence must bind the packaged outputs" )
output = args.output.absolute() if not args.dry_run: prepare_output_directory(output, versioned_names(version_info["version"], development_label)["source_archive"])
check_ancestors(source_root) binary_dir = (args.binary_dir or (source_root / "target" / "release")).absolute() check_ancestors(binary_dir) if args.pack and not binary_dir.is_dir(): raise UsageError(f"--binary-dir {binary_dir} does not exist; build the binaries first") build_env_path = (args.build_env or (binary_dir / manifest["build_environment"]["descriptor_name"])).absolute() check_ancestors(build_env_path) build_env = None if args.pack and build_env_path.exists(): build_env = _load_json(build_env_path, limits, "build-environment descriptor") if args.vendor_dir: check_ancestors(args.vendor_dir.absolute())
state = resolve_source_state(source_root, args.tag, development, exclude_prefixes=[output])
context = { "manifest": manifest, "manifest_path": manifest_path, "source_root": source_root, "output": output, "binary_dir": binary_dir, "build_env": build_env, "build_env_path": build_env_path, "version": version_info["version"], "version_info": version_info, "source": state, "development": development, "development_label": development_label, "created_at": utc_now(), "names": versioned_names(version_info["version"], development_label), "policy": sentinel_policy(manifest), "exemptions": review_exemptions(manifest), "vendor_fixture_exemptions": vendored_fixture_exemptions(manifest, source_root), "limits": limits, "generator": {"tool": "tools/package_release.py", "sha256": sha256_file(Path(__file__).resolve())}, }
context["runtime_bundle"] = load_runtime_bundle(args, context) source_manifest = None if args.prepare_source: source_manifest = prepare_source(args, context) if args.pack: if source_manifest is None: path = output / "source-manifest.json" if not path.exists(): raise CheckFailure( f"{path} is missing; run --prepare-source first so the binaries ship with their source" ) source_manifest = _load_json(path) context["privacy"] = source_manifest["privacy"] if context["runtime_bundle"] is not None and not ( source_manifest.get("runtime_bundle") or {} ).get("present"): raise CheckFailure( "the runtime-sources options were given with --pack alone, but the prepared source " "manifest records no runtime bundle; run --prepare-source with the same options first, " "so the source archive carries the runtime-sources tree the notices come from" ) release_manifest = pack(args, context, source_manifest) else: release_manifest = None
summary = { "modes": {"prepare_source": args.prepare_source, "pack": args.pack}, "dry_run": args.dry_run, "output": str(output), "version": version_info["version"], "development": development, "development_label": development_label, "archive": context["names"]["archive"], "source_archive": context["names"]["source_archive"], "source_completeness": (source_manifest or {}).get("source_completeness", {}).get("status"), "runtime_bundle": ( "absent" if not ((source_manifest or {}).get("runtime_bundle") or {}).get("present") else ( "source_complete" if (source_manifest or {})["runtime_bundle"].get("source_complete") else "incomplete_coverage" ) ), "release_ready": (release_manifest or {}).get("release_ready", False), "blocking": (release_manifest or {}).get("publication", {}).get("blocking", []), } if args.json: print(json.dumps(summary, indent=2, sort_keys=True)) else: print(f"version: {summary['version']}" + (" (development)" if development else "")) print(f"output: {summary['output']}") if args.prepare_source: print(f"source archive: {summary['source_archive']}") print(f"source completeness: {summary['source_completeness']}") print(f"runtime sources: {summary['runtime_bundle']}") if args.pack: print(f"binary archive: {summary['archive']}") print(f"release ready: {summary['release_ready']}") if summary["blocking"]: print("blocking:") for item in summary["blocking"]: print(f" - {item}") return EXIT_OK
def main(argv: list[str] | None = None) -> int: try: return run(list(sys.argv[1:] if argv is None else argv)) except UsageError as error: print(f"error: {error}", file=sys.stderr) return EXIT_USAGE except CheckFailure as error: print(f"check failed: {error}", file=sys.stderr) return EXIT_FAILED except Exception as error: # pragma: no cover - internal error path import traceback
traceback.print_exc() print(f"internal error: {error}", file=sys.stderr) return EXIT_INTERNAL
if __name__ == "__main__": raise SystemExit(main())