🔒 Repository is read-only – file editing is disabled.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916391739183919392039213922392339243925392639273928392939303931393239333934393539363937393839393940394139423943394439453946394739483949395039513952395339543955395639573958395939603961396239633964396539663967396839693970397139723973397439753976397739783979398039813982398339843985398639873988398939903991399239933994399539963997399839994000400140024003400440054006400740084009401040114012401340144015401640174018401940204021402240234024402540264027402840294030403140324033403440354036403740384039404040414042404340444045404640474048404940504051405240534054405540564057405840594060406140624063406440654066406740684069407040714072407340744075407640774078407940804081408240834084408540864087408840894090409140924093409440954096409740984099410041014102410341044105410641074108410941104111411241134114411541164117411841194120412141224123412441254126412741284129413041314132413341344135413641374138413941404141414241434144414541464147414841494150415141524153415441554156415741584159416041614162416341644165416641674168416941704171417241734174417541764177417841794180418141824183418441854186418741884189419041914192419341944195419641974198419942004201420242034204420542064207420842094210421142124213421442154216421742184219422042214222422342244225422642274228422942304231423242334234423542364237423842394240424142424243424442454246424742484249425042514252425342544255425642574258425942604261426242634264426542664267426842694270427142724273427442754276427742784279428042814282428342844285428642874288428942904291429242934294429542964297429842994300430143024303430443054306430743084309431043114312431343144315431643174318431943204321432243234324432543264327432843294330433143324333433443354336433743384339434043414342434343444345434643474348434943504351435243534354435543564357435843594360436143624363436443654366436743684369437043714372437343744375437643774378437943804381438243834384438543864387438843894390439143924393439443954396439743984399440044014402440344044405440644074408440944104411441244134414441544164417441844194420442144224423442444254426442744284429443044314432443344344435443644374438443944404441444244434444444544464447444844494450445144524453445444554456445744584459446044614462446344644465446644674468446944704471447244734474447544764477447844794480448144824483448444854486448744884489449044914492449344944495449644974498449945004501450245034504450545064507450845094510451145124513451445154516451745184519452045214522452345244525452645274528452945304531453245334534453545364537453845394540454145424543454445454546454745484549455045514552455345544555455645574558455945604561456245634564456545664567456845694570457145724573457445754576457745784579458045814582458345844585458645874588458945904591459245934594459545964597459845994600460146024603460446054606460746084609461046114612461346144615461646174618461946204621462246234624462546264627462846294630463146324633463446354636463746384639464046414642464346444645464646474648464946504651465246534654465546564657465846594660466146624663466446654666466746684669467046714672467346744675467646774678467946804681468246834684468546864687468846894690469146924693469446954696469746984699470047014702470347044705470647074708470947104711471247134714471547164717471847194720472147224723472447254726472747284729473047314732473347344735473647374738473947404741474247434744474547464747474847494750475147524753475447554756475747584759476047614762476347644765476647674768476947704771477247734774477547764777477847794780478147824783478447854786478747884789479047914792479347944795479647974798479948004801480248034804480548064807480848094810481148124813481448154816481748184819482048214822482348244825482648274828482948304831483248334834483548364837483848394840484148424843484448454846484748484849485048514852485348544855485648574858485948604861486248634864486548664867486848694870487148724873487448754876487748784879488048814882488348844885488648874888488948904891489248934894489548964897489848994900490149024903490449054906490749084909491049114912491349144915491649174918491949204921492249234924492549264927492849294930493149324933493449354936493749384939494049414942494349444945494649474948494949504951495249534954495549564957495849594960496149624963496449654966496749684969497049714972497349744975497649774978497949804981498249834984498549864987498849894990499149924993499449954996499749984999500050015002500350045005500650075008500950105011501250135014501550165017501850195020502150225023502450255026502750285029503050315032503350345035503650375038503950405041504250435044504550465047504850495050505150525053505450555056505750585059506050615062506350645065506650675068506950705071507250735074507550765077507850795080508150825083508450855086508750885089509050915092509350945095509650975098509951005101510251035104510551065107510851095110511151125113511451155116511751185119512051215122512351245125512651275128512951305131513251335134513551365137513851395140514151425143514451455146514751485149515051515152515351545155515651575158515951605161516251635164516551665167516851695170517151725173517451755176517751785179518051815182518351845185518651875188518951905191519251935194519551965197519851995200520152025203520452055206520752085209521052115212521352145215521652175218521952205221522252235224522552265227522852295230523152325233523452355236523752385239524052415242524352445245524652475248524952505251525252535254525552565257525852595260526152625263526452655266526752685269527052715272527352745275527652775278527952805281528252835284528552865287528852895290529152925293529452955296529752985299530053015302530353045305530653075308530953105311531253135314531553165317531853195320532153225323532453255326532753285329533053315332533353345335533653375338533953405341534253435344534553465347534853495350535153525353535453555356535753585359536053615362536353645365536653675368536953705371537253735374537553765377537853795380538153825383538453855386538753885389539053915392539353945395539653975398539954005401540254035404540554065407540854095410541154125413541454155416541754185419542054215422542354245425542654275428542954305431543254335434543554365437543854395440544154425443544454455446544754485449545054515452545354545455545654575458545954605461546254635464546554665467546854695470547154725473547454755476547754785479548054815482548354845485548654875488548954905491549254935494549554965497549854995500550155025503550455055506550755085509551055115512551355145515551655175518551955205521552255235524552555265527552855295530553155325533553455355536553755385539554055415542554355445545554655475548554955505551555255535554555555565557555855595560556155625563556455655566556755685569557055715572557355745575557655775578557955805581558255835584558555865587558855895590559155925593559455955596559755985599560056015602560356045605560656075608560956105611561256135614561556165617561856195620562156225623562456255626562756285629563056315632563356345635563656375638563956405641564256435644564556465647564856495650565156525653565456555656565756585659566056615662566356645665566656675668566956705671567256735674567556765677567856795680568156825683568456855686568756885689569056915692569356945695569656975698569957005701570257035704570557065707570857095710571157125713571457155716571757185719572057215722572357245725572657275728572957305731573257335734573557365737573857395740574157425743574457455746574757485749575057515752575357545755575657575758575957605761576257635764576557665767576857695770577157725773577457755776577757785779578057815782578357845785578657875788578957905791579257935794579557965797579857995800580158025803580458055806580758085809581058115812581358145815581658175818581958205821582258235824582558265827582858295830583158325833583458355836583758385839584058415842584358445845584658475848584958505851585258535854585558565857585858595860586158625863586458655866586758685869587058715872587358745875587658775878587958805881588258835884588558865887588858895890589158925893589458955896589758985899590059015902590359045905590659075908590959105911591259135914591559165917591859195920592159225923592459255926592759285929593059315932593359345935
#!/usr/bin/env python3
"""
╔══════════════════════════════════════════════════════════════════════════════╗
║ PAG - Pagan Linux Package Manager v3.3.21 ║
║ Produkcyjny menedżer pakietów – atomowy, bezpieczny, i18n ║
╚══════════════════════════════════════════════════════════════════════════════╝
KLUCZOWE CECHY:
- Atomowa instalacja przez staging (tmpdir → rename) – brak pół-instalacji
- Bezpieczne usuwanie – sprawdza czy plik nie jest współdzielony
- SQLite dla bazy plików – miliony plików bez problemu
- GPG: weryfikacja repo.json + podpisy pakietów + pinning fingerprintu
- Hooki: pre/post-install, pre/post-remove (piaskownica env, timeout, audit)
- Głęboka weryfikacja SHA256 per-plik
- Pełny rollback – cofa fizyczne pliki
- Blokada flock – tylko jedna instancja
- Transakcje z migawkami + rejestr wykonanych hooków
- Cache HTTP (ETag/If-Modified-Since)
- Wielojęzyczność (i18n) – PL, EN
FORMAT PAKIETU (.pag):
├── data.tar.xz – pliki + sums.json (SHA256 per plik)
├── metadata.json – nazwa, wersja, zależności
└── hooks/ – pre-install, post-install, pre-remove, post-remove
MODEL ZAUFANIA / BEZPIECZEŃSTWO:
- Repozytorium MUSI być zaufane: podpisy GPG zweryfikowane; fingerprint
klucza przypiętego do repo (TOFU przy pierwszym użyciu, potem pinning).
- Hooki uruchamiają dowolny plik z pakietu jako ROOT (jak apt/pacman).
Ograniczamy je (czyste env, timeout, PAG_NO_HOOKS=1, log do
/var/log/pag/audit.log) i rejestrujemy w transakcji, ale ostatecznie
instalujesz kod, któremu ufasz.
- self-update: weryfikacja podpisu + SHA256 + składnia, atomowa podmiana.
"""
import os, sys, json, copy, shutil, hashlib, tarfile, tempfile, subprocess, time, fcntl, sqlite3, locale, re, difflib
# Fix TLS trust inside the Pagan chroot: point Python at the CA bundle that
# pag ships, otherwise urlopen() fails with "unable to get local issuer
# certificate" (no default capath/cafile is resolved in the chroot).
for _cafile in (
"/etc/ssl/certs/ca-certificates.crt",
"/etc/ssl/cert.pem",
):
if os.path.isfile(_cafile):
os.environ["SSL_CERT_FILE"] = _cafile
break
from pathlib import Path
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple, Set
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.request import urlopen, Request
import threading, itertools
import uuid # serialNumber SBOM (CycloneDX)
# Wersja klienta – do porównania z repo.json["pag_version"] (self-update)
PAG_VERSION = "3.4.0"
from urllib.error import URLError, HTTPError
# =============================================================================
# ProgressBar — minimalistyczny pasek postępu (bez zewnętrznych zależności)
# =============================================================================
class ProgressBar:
"""Czysty Python progress bar — działa z TTY i bez."""
def __init__(self, total: int, desc: str = "", unit: str = "", width: int = 30):
self.total = max(total, 1)
self.desc = desc
self.unit = unit
self.width = width
self.n = 0
self.start = time.time()
self.tty = sys.stderr.isatty()
self._last_line_len = 0
def update(self, n: Optional[int] = None, suffix: str = ""):
if n is not None:
self.n = n
else:
self.n += 1
pct = self.n / self.total * 100
elapsed = time.time() - self.start
speed = self.n / elapsed if elapsed > 0 else 0
if self.n >= self.total:
eta_str = "done"
elif speed > 0:
eta = (self.total - self.n) / speed
eta_str = f"{eta:.0f}s" if eta < 60 else f"{eta/60:.1f}m"
else:
eta_str = "?..."
bar_len = int(self.width * pct / 100)
bar = "█" * bar_len + "░" * (self.width - bar_len)
line = f" {self.desc} [{bar}] {self.n}/{self.total} ({pct:.0f}%) ETA {eta_str}{suffix}"
if self.tty:
# Overwrite current line
clear = " " * max(0, self._last_line_len - len(line))
print(f"\r{line}{clear}", end="", file=sys.stderr, flush=True)
self._last_line_len = len(line)
else:
# Print milestone lines only (every 10% or when done)
if self.n == 1 or self.n >= self.total or self.n % max(1, self.total // 10) == 0:
print(line, file=sys.stderr)
def close(self):
if self.tty:
print(file=sys.stderr)
self._last_line_len = 0
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
class DownloadBar:
"""Pasek postępu pobierania — na podstawie Content-Length."""
def __init__(self, filename: str, total_bytes: int):
self.filename = filename
self.total = total_bytes
self.downloaded = 0
self.start = time.time()
self.tty = sys.stderr.isatty()
self._last_len = 0
def update(self, chunk_size: int):
self.downloaded += chunk_size
if self.total <= 0:
return
pct = self.downloaded / self.total * 100
elapsed = time.time() - self.start
speed = self.downloaded / elapsed if elapsed > 0 else 0
if speed > 0:
eta = (self.total - self.downloaded) / speed
eta_str = f"{eta:.0f}s" if eta < 60 else f"{eta/60:.1f}m"
else:
eta_str = "?..."
bar_len = 25
filled = int(bar_len * pct / 100)
bar = "█" * filled + "░" * (bar_len - filled)
sz = self._fmt_size(self.total)
spd = self._fmt_size(int(speed))
line = f" ↓ {self.filename} [{bar}] {pct:.0f}% {sz} {spd}/s ETA {eta_str}"
if self.tty:
clear = " " * max(0, self._last_len - len(line))
print(f"\r{line}{clear}", end="", file=sys.stderr, flush=True)
self._last_len = len(line)
def close(self):
if self.tty and self.total > 0:
print(file=sys.stderr)
@staticmethod
def _fmt_size(n: int) -> str:
for unit in ("B", "KB", "MB", "GB"):
if n < 1024:
return f"{n:.1f} {unit}"
n /= 1024
return f"{n:.1f} TB"
# =============================================================================
# GPG – BEZPIECZNE WYWOŁYWANIE (odporne na brak binarki gpg)
# =============================================================================
GPG_BINARY = shutil.which("gpg2") or shutil.which("gpg") or "gpg"
GPG_HOME = "/etc/pag/gpg" # izolowany keyring (działa z keyboxd GPG 2.4+)
def _ssl_ok() -> bool:
"""Czy Python ma działający moduł ssl (HTTPS w urllib).
Bez niego urllib nie zna schematu https i każda operacja sieciowa kończy
się „unknown url type: https” – myląco, bo wygląda jak błąd adresu.
"""
try:
import ssl # noqa: F401
return True
except Exception:
return False
def _gpg_run(*args, timeout: int = 30, **kwargs) -> subprocess.CompletedProcess:
"""
Bezpieczne wywołanie GPG – przechwytuje FileNotFoundError,
gdyby gpg/gpg2 nie było zainstalowane w minimalnym środowisku.
Wymusza LC_ALL=C aby komunikaty GPG były zawsze po angielsku
(niezależnie od locale systemu) – kluczowe dla parsowania stderr.
"""
env = kwargs.pop("env", None) or os.environ.copy()
env["LC_ALL"] = "C"
env["GNUPGHOME"] = GPG_HOME
try:
return subprocess.run([GPG_BINARY, *args], timeout=timeout, env=env, **kwargs)
except FileNotFoundError:
# GPG nie jest dostępne – zwróć błąd z komunikatem
# (szanuj text=True – inaczej caller dostaje bytes i może wybuchnąć TypeError)
_text = bool(kwargs.get("text") or kwargs.get("universal_newlines"))
_msg = f"GPG binary not found ({GPG_BINARY})"
return subprocess.CompletedProcess(
[GPG_BINARY, *args], 127,
stdout=("" if _text else b""),
stderr=(_msg if _text else _msg.encode()),
)
except subprocess.TimeoutExpired:
return subprocess.CompletedProcess(
[GPG_BINARY, *args], 124,
stdout=b"", stderr=b"GPG operation timed out"
)
def _load_trust_db() -> dict:
"""Mapa repo_url → fingerprint klucza podpisującego (baza zaufania)."""
try:
with open(TRUST_DB) as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {}
def _save_trust_db(db: dict):
os.makedirs(os.path.dirname(TRUST_DB), exist_ok=True)
with open(TRUST_DB, "w") as f:
json.dump(db, f, indent=2)
def _gpg_verify_fp(sig_path: str, data_path: str, timeout: int = 30):
"""Weryfikuje podpis i odczytuje fingerprint podpisującego.
Używa --status-fd=1 i linii VALIDSIG <fingerprint>. Zwraca (ok, fingerprint).
FAIL-CLOSED: OK tylko gdy GPG zwrócił 0 ORAZ w statusie maszynowym jest
linia VALIDSIG. Nie zakładamy sukcesu na podstawie samego kodu wyjścia –
gdyby format --status-fd się zmienił, wolimy odrzucić podpis niż przyjąć
niezaufany pakiet (wywołujący traktuje fp=None jako brak potwierdzenia).
"""
env = os.environ.copy()
res = _gpg_run("--verify", "--status-fd", "1", sig_path, data_path,
capture_output=True, text=True, timeout=timeout, env=env)
out = res.stdout or ""
m = re.search(r"\[GNUPG:\]\s+VALIDSIG\s+([0-9A-Fa-f]{16,})", out)
if not m:
m = re.search(r"\bVALIDSIG\s+([0-9A-Fa-f]{16,})", out)
if res.returncode != 0 or not m:
return False, None
return True, m.group(1).upper()
def _verify_local_pkg_signature(pkg_path: str) -> bool:
"""Weryfikuje lokalny .pag tym samym kluczem co zaufane repo."""
sig_path = next((pkg_path + ext for ext in (".asc", ".sig")
if os.path.isfile(pkg_path + ext)), "")
if not sig_path:
print(f" ❌ Brak podpisu dla pakietu: {pkg_path}")
print(" Wymagany jest plik .pag.asc/.pag.sig z tego samego źródła.")
return False
ok, fingerprint = _gpg_verify_fp(sig_path, pkg_path)
if not ok or not fingerprint:
print(f" ❌ Nieprawidłowy podpis pakietu: {pkg_path}")
return False
expected = _repo_pinned_fp(DEFAULT_REPOS[0])
if not expected:
print(" ❌ Brak przypiętego klucza repo dla lokalnego pakietu.")
print(" Uruchom najpierw: sudo pag sync")
return False
if fingerprint != expected.upper():
print(" ❌ [SECURITY ERROR] Pakiet podpisano innym kluczem repo.")
print(f" Oczekiwany: {expected}")
print(f" Otrzymany: {fingerprint}")
return False
print(f" 🔐 Podpis pakietu OK ({fingerprint})")
return True
# =============================================================================
# i18n – WIELOJĘZYCZNOŚĆ
# =============================================================================
LANG = os.environ.get("LANG", "en_US.UTF-8")[:2] # pl, en, de...
COLOR = os.environ.get("NO_COLOR", "") == "" and sys.stdout.isatty()
def _c(code: str, text: str) -> str:
"""Dodaje kody ANSI jeśli kolor jest włączony."""
if not COLOR:
return text
colors = {
"green": "\033[32m", "red": "\033[31m", "yellow": "\033[33m",
"cyan": "\033[36m", "bold": "\033[1m", "dim": "\033[2m",
"reset": "\033[0m",
}
return f"{colors.get(code,'')}{text}{colors['reset']}"
T = {
"en": {
"root_required": "pag requires root privileges (sudo).",
"db_locked": "Another pag instance is running.",
"db_lock_hint": "If no other pag process is running, wait a moment and retry.",
"no_index": "Cannot fetch repository indexes. Run 'pag update'.",
"cache_ro": "Repo cache is read-only ({cache}) – using local index (may be outdated).\n Refresh as root: sudo pag sync",
"all_installed": "All packages are already installed.",
"to_install": "To install: {} packages ({:.2f} MB)",
"new": "NEW",
"continue_q": "Continue? [Y/n] ",
"no_tty": "No TTY / stdin closed (EOF) – cancelling.",
"cancelled": "Cancelled.",
"not_found": "not found in repos",
"pkg_not_found": "Package not found: {} (not in any repo)",
"not_found_hint": "Check the spelling or run 'pag search <query>'.",
"downloading": "Downloading",
"download_fail": "download failed",
"gpg_fail": "GPG verification failed",
"sha256_mismatch": "SHA256 mismatch",
"install_failed": "installation failed",
"installed": "Installed {} packages.",
"rollback_restored": "Restored previous state from snapshot.",
"rollback_files": "Rolled back {} files.",
"no_history": "No transaction history.",
"pinned_list": "Pinned packages ({}):",
"no_pinned": "No pinned packages.",
"pinned_to": "pinned to",
"unpinned": "unpinned.",
"not_pinned": "was not pinned.",
"repo_added": "Added repository: {}",
"repo_exists": "Repository already exists: {}",
"updated_done": "Index refresh complete. {} packages cached.",
"indexes_refreshed": "Indexes refreshed.",
"updates_available": "⚠ {} packages have updates – run: pag update",
"upgrading": "Upgrading: {} packages",
"all_up_to_date": "All packages are up to date.",
"removing": "Removing",
"orphans_found": "Orphaned dependencies ({}): {}",
"flatpak_missing": "Flatpak is not installed.",
"flatpak_adding": "Adding Flathub remote...",
"flatpak_searching": "Searching Flathub for '{}'...",
"flatpak_found": "Found {} results:",
"flatpak_not_found": "not found on Flathub",
"flatpak_install_prompt": "Install {}? [Y/n] ",
"flatpak_installing": "Installing {}...",
"flatpak_installed": "Flatpak {} installed.",
"flatpak_removed": "Flatpak {} removed.",
"flatpak_not_installed": "Flatpak {} is not installed.",
"flatpak_info_id": "ID",
"flatpak_info_version": "Version",
"flatpak_info_branch": "Branch",
"flatpak_info_origin": "Origin",
"flatpak_info_size": "Installed size",
"flatpak_info_desc": "Description",
"flatpak_updated": "Flatpaks updated.",
"flatpak_usage": "Usage: pag flatpak <search|install|remove|list|update|info> [args]",
"flatpak_scope": "Installation",
"flatpak_menu_hint": "Installed and verified, but this session may not show it in the menu: Flatpak exports are missing from XDG_DATA_DIRS.",
"flatpak_menu_fix": "Add /var/lib/flatpak/exports/share to XDG_DATA_DIRS (and .../exports/bin to PATH) in your session, then re-login – PaganDE does this in session/pagande-session.",
"key_imported": "Key imported successfully.",
"key_removed": "Key removed: {}",
"no_keys": "No trusted GPG keys.",
"gpg_missing": "GNUPG MISSING – install gnupg and retry",
"ssl_broken": "Python has no working ssl module (HTTPS impossible) – fix the python/openssl packages (e.g. sudo pag install -f python).",
"key_add_failed": "Key import failed (gpg error) – key was NOT added.",
"verify_ok": "All {} files intact.",
"verify_errors": "{} problems found:",
"cache_cleared": "{} files ({:.2f} MB) cleared from cache.",
"deployments_list": "Deployments ({}):",
"no_deployments": "No deployments.",
"active_deployment": "ACTIVE",
"deploy_rollback_ok": "Switched to deployment: {}",
"deploy_rollback_fail": "No previous deployment.",
"deploy_cleanup_ok": "Removed {} old deployments.",
"deploy_cleanup_none": "No deployments to clean (minimum {}).",
"why_explicit": "explicitly installed",
"why_dependency": "dependency of",
"why_not_installed": "not installed",
"autoremove_ok": "Removed {} orphaned packages.",
"autoremove_none": "No orphaned packages.",
"downloaded": "Downloaded {} to cache ({:.2f} MB).",
"provides_mapped": "{} → {} (provides)",
"stats_title": "PAG Statistics",
"stats_packages": "Installed packages",
"stats_files": "Tracked files",
"stats_size": "Total size",
"stats_cache": "Cache size",
"stats_history": "Transactions",
"stats_last_update": "Last update",
# Komunikaty bezpieczeństwa (baza EN; PL w tabeli "pl" jako sec_*_pl)
"sec_downgrade": "Downgrade blocked: {pkg} {new} < {old}",
"sec_suid": "SUID stripped from {path}",
"sec_https": "HTTPS required for repos",
"sec_badname": "Invalid package name: {name}",
"sec_toobig": "Package too large: {size_mb}MB > {max_mb}MB",
"sec_conflict": "File conflict: {path} owned by {owner}",
"sec_audit": "{pkg} installed by {user}",
"sec_locked": "Another pag process is running",
# Konfiguracja (/etc) – zachowanie zmian użytkownika
"conf_pacnew": "Modified config kept; new version saved as {path}",
"conf_pacsave": "Modified config kept as {path}",
# Zależności wirtualne (provides)
"provides_conflict": "Multiple providers for '{name}' ({providers}) – using '{chosen}'",
# Base baked into the ISO image (pag base / base adopt)
"base_title": "Base from the ISO image",
"base_no_manifest": "This system carries no base manifest (/usr/share/pag/base.json) – nothing to adopt.",
"base_iso_version": "Image version",
"base_total": "Base packages (from image)",
"base_tracked": "Managed by pag",
"base_adoptable": "Adoptable from repo",
"base_unmatched": "No counterpart in repo",
"base_adopt_hint": "Adopt them (image files get replaced by repo packages): sudo pag base adopt",
"base_unmatched_hint": "These stay at the image version (no such package in repo):",
"base_more": "more",
"base_nothing_selected": "None of the given names is on the adopt list.",
"base_all_managed": "Every base package is already managed by pag.",
"base_adopt_plan": "Adopting {} base packages from the repo (~{:.0f} MB):",
"base_dry_run": "Dry run – nothing was installed.",
"base_unmanaged_hint": "{} packages from the ISO image are not managed by pag yet – 'sudo pag base adopt' replaces them with repo packages (only then does 'pag upgrade' update them).",
"base_overwrite": "{} comes from the ISO image – its files will be replaced (backup kept for 'pag rollback').",
"base_reboot_hint": "Base adopted from the repo – reboot before using the new libraries/toolchain.",
"stats_base": "Base (image) managed",
},
"pl": {
"root_required": "pag wymaga uprawnień root (sudo).",
"db_locked": "Inna instancja pag jest uruchomiona.",
"db_lock_hint": "Jeśli żaden inny proces pag nie działa, poczekaj chwilę i spróbuj ponownie.",
"no_index": "Nie można pobrać indeksów repozytoriów. Uruchom 'pag update'.",
"cache_ro": "Cache repozytoriów jest tylko-do-odczytu ({cache}) – używam lokalnego indeksu (może być nieaktualny).\n Odśwież jako root: sudo pag sync",
"all_installed": "Wszystkie pakiety są już zainstalowane.",
"to_install": "Do zainstalowania: {} pakietów ({:.2f} MB)",
"new": "NOWY",
"continue_q": "Kontynuować? [T/n] ",
"no_tty": "Brak terminala (EOF) – anuluję.",
"cancelled": "Anulowano.",
"not_found": "brak w repozytoriach",
"pkg_not_found": "Nie znaleziono pakietu: {} (brak w repozytoriach)",
"not_found_hint": "Sprawdź pisownię lub uruchom 'pag search <fraza>'.",
"downloading": "Pobieranie",
"download_fail": "błąd pobierania",
"gpg_fail": "błąd weryfikacji GPG",
"sha256_mismatch": "niezgodność SHA256",
"install_failed": "błąd instalacji",
"installed": "Zainstalowano {} pakietów.",
"rollback_restored": "Przywrócono poprzedni stan z migawki.",
"rollback_files": "Wycofano {} plików.",
"no_history": "Brak historii transakcji.",
"pinned_list": "Przypięte pakiety ({}):",
"no_pinned": "Brak przypiętych pakietów.",
"pinned_to": "przypięty do",
"unpinned": "odpięty.",
"not_pinned": "nie był przypięty.",
"repo_added": "Dodano repozytorium: {}",
"repo_exists": "Repozytorium już istnieje: {}",
"updated_done": "Odświeżanie zakończone. {} pakietów w cache.",
"indexes_refreshed": "Indeksy odświeżone.",
"updates_available": "⚠ jest {} pakietów do zaktualizowania – wpisz: pag update",
"upgrading": "Aktualizacje: {} pakietów",
"all_up_to_date": "Wszystkie pakiety są aktualne.",
"removing": "Usuwanie",
"orphans_found": "Osierocone zależności ({}): {}",
"flatpak_missing": "Flatpak nie jest zainstalowany.",
"flatpak_adding": "Dodaję zdalne repozytorium Flathub...",
"flatpak_searching": "Szukam '{}' we Flathub...",
"flatpak_found": "Znaleziono {} wyników:",
"flatpak_not_found": "nie znaleziono we Flathub",
"flatpak_install_prompt": "Zainstalować {}? [T/n] ",
"flatpak_installing": "Instalowanie {}...",
"flatpak_installed": "Flatpak {} zainstalowany.",
"flatpak_removed": "Flatpak {} usunięty.",
"flatpak_not_installed": "Flatpak {} nie jest zainstalowany.",
"flatpak_info_id": "ID",
"flatpak_info_version": "Wersja",
"flatpak_info_branch": "Gałąź",
"flatpak_info_origin": "Źródło",
"flatpak_info_size": "Rozmiar",
"flatpak_info_desc": "Opis",
"flatpak_updated": "Flapaki zaktualizowane.",
"flatpak_usage": "Użycie: pag flatpak <search|install|remove|list|update|info> [args]",
"flatpak_scope": "Instalacja",
"flatpak_menu_hint": "Zainstalowano i zweryfikowano, ale sesja może jej nie pokazać w menu: brak eksportów Flatpaka w XDG_DATA_DIRS.",
"flatpak_menu_fix": "Dodaj /var/lib/flatpak/exports/share do XDG_DATA_DIRS (a .../exports/bin do PATH) w sesji i zaloguj się ponownie – PaganDE robi to w session/pagande-session.",
"key_imported": "Klucz zaimportowany pomyślnie.",
"key_removed": "Klucz usunięty: {}",
"no_keys": "Brak zaufanych kluczy GPG.",
"gpg_missing": "BRAK GNUPG – zainstaluj gnupg i spróbuj ponownie",
"ssl_broken": "Python bez działającego modułu ssl (HTTPS niemożliwe) – napraw pakiety python/openssl (np. sudo pag install -f python).",
"key_add_failed": "Import klucza nie powiódł się (błąd gpg) – klucz NIE został dodany.",
"verify_ok": "Wszystkie {} plików sprawne.",
"verify_errors": "Znaleziono {} problemów:",
"cache_cleared": "{} plików ({:.2f} MB) usuniętych z cache.",
"deployments_list": "Deploymenty ({}):",
"no_deployments": "Brak deploymentów.",
"active_deployment": "AKTYWNY",
"deploy_rollback_ok": "Przełączono na deployment: {}",
"deploy_rollback_fail": "Brak poprzedniego deploymentu.",
"deploy_cleanup_ok": "Usunięto {} starych deploymentów.",
"deploy_cleanup_none": "Nie ma deploymentów do wyczyszczenia (minimum {}).",
"why_explicit": "zainstalowany jawnie",
"why_dependency": "zależność od",
"why_not_installed": "niezainstalowany",
"autoremove_ok": "Usunięto {} osieroconych pakietów.",
"autoremove_none": "Brak osieroconych pakietów.",
"downloaded": "Pobrano {} do cache ({:.2f} MB).",
"sec_downgrade": "Downgrade blocked: {pkg} {new} < {old}",
"sec_suid": "SUID stripped from {path}",
"sec_https": "HTTPS required for repos",
"sec_badname": "Invalid package name: {name}",
"sec_toobig": "Package too large: {size_mb}MB > {max_mb}MB",
"sec_conflict": "File conflict: {path} owned by {owner}",
"sec_audit": "{pkg} installed by {user}",
"sec_locked": "Another pag process is running",
"sec_downgrade_pl": "Blokada downgrade: {pkg} {new} < {old}",
"sec_suid_pl": "SUID usuniety z {path}",
"sec_https_pl": "Repozytorium wymaga HTTPS",
"sec_badname_pl": "Nieprawidlowa nazwa pakietu: {name}",
"sec_toobig_pl": "Paczka za duza: {size_mb}MB > {max_mb}MB",
"sec_conflict_pl": "Konflikt plikow: {path} nalezy do {owner}",
"sec_audit_pl": "{pkg} zainstalowany przez {user}",
"sec_locked_pl": "Inny proces pag juz dziala",
"provides_mapped": "{} → {} (provides)",
"stats_title": "Statystyki PAG",
"stats_packages": "Zainstalowane pakiety",
"stats_files": "Śledzone pliki",
"stats_size": "Całkowity rozmiar",
"stats_cache": "Rozmiar cache",
"stats_history": "Transakcje",
"stats_last_update": "Ostatnia aktualizacja",
"conf_pacnew": "Zmieniony plik konfiguracyjny zachowany; nowa wersja: {path}",
"conf_pacsave": "Zmieniony plik konfiguracyjny zachowany jako {path}",
"provides_conflict": "Wielu dostawców dla '{name}' ({providers}) – używam '{chosen}'",
# Baza wypalona w obrazie ISO (pag base / base adopt)
"base_title": "Baza z obrazu ISO",
"base_no_manifest": "Ten system nie ma manifestu bazy (/usr/share/pag/base.json) – nie ma czego adoptować.",
"base_iso_version": "Wersja obrazu",
"base_total": "Pakiety bazy (z obrazu)",
"base_tracked": "Zarządzane przez pag",
"base_adoptable": "Do adoptowania z repo",
"base_unmatched": "Bez odpowiednika w repo",
"base_adopt_hint": "Adoptuj (pliki z obrazu zostaną zastąpione pakietami z repo): sudo pag base adopt",
"base_unmatched_hint": "Te zostaną w wersji z obrazu (brak takiego pakietu w repo):",
"base_more": "więcej",
"base_nothing_selected": "Żadna z podanych nazw nie jest na liście do adoptowania.",
"base_all_managed": "Wszystkie pakiety bazy są już zarządzane przez pag.",
"base_adopt_plan": "Adoptowanie {} pakietów bazy z repo (~{:.0f} MB):",
"base_dry_run": "Próba na sucho – nic nie zainstalowano.",
"base_unmanaged_hint": "{} pakietów z obrazu ISO nie jest jeszcze zarządzanych przez pag – 'sudo pag base adopt' zamieni je na pakiety z repo (dopiero wtedy 'pag upgrade' je aktualizuje).",
"base_overwrite": "{} pochodzi z obrazu ISO – pliki zostaną zastąpione (kopia zapasowa dla 'pag rollback').",
"base_reboot_hint": "Baza zaadoptowana z repo – zrestartuj system, zanim zaczniesz używać nowych bibliotek/narzędzi.",
"stats_base": "Baza (obraz) zarządzana",
},
}
# ── Tłumaczenia z PLIKÓW (nadpisują/rozszerzają wbudowane PL/EN) ─────────────
# Kolejność: PAG_LANG_DIR (env) → /etc/pag/lang → /usr/share/pag/lang →
# ./pag-lang obok binarki (dev). Brak plików NIE jest błędem – zostaje
# wbudowany słownik T (fallback), więc pag zawsze działa.
# Przykład pliku (pag-lang/pl.json): {"app_title": "...", "usage": "..."}.
def _load_lang_files() -> None:
# PAG_LANG_NO_FILES=1 → pomiń pliki (używane przy eksporcie --lang-extract,
# żeby wyeksportować CZYSTE wbudowane słowniki, bez starych nadpisań).
if os.environ.get("PAG_LANG_NO_FILES") == "1":
return
dirs = []
_env = os.environ.get("PAG_LANG_DIR")
if _env:
dirs.append(_env)
# __file__ bywa niedostępne, gdy moduł jest exec/frozen (np. harness
# instalacyjny) – liczymy wtedy od ścieżki programu, zamiast wywalać
# NameError przy samym imporcie.
_self = globals().get("__file__") or sys.argv[0] or "pag"
dirs += ["/etc/pag/lang", "/usr/share/pag/lang",
os.path.join(os.path.dirname(os.path.abspath(_self)), "pag-lang")]
for _d in dirs:
for _code in list(T.keys()) + ["pl", "en", "de"]:
_p = os.path.join(_d, f"{_code}.json")
try:
with open(_p, "r", encoding="utf-8") as _fh:
_data = json.load(_fh)
if isinstance(_data, dict):
# Merge (nie replace): klucze nieobecne w pliku zachowują
# wbudowane tłumaczenie. Puste wartości pomijamy, żeby
# niekompletny/uszkodzony plik nie wyczyścił komunikatu.
_clean = {str(k): str(v) for k, v in _data.items()
if str(v).strip()}
T.setdefault(_code, {}).update(_clean)
except (OSError, ValueError):
continue
_load_lang_files()
def _(key: str, *args, **kwargs) -> str:
"""Tłumaczy klucz i formatuje argumenty.
Dla LANG=pl preferuje wariant „<key>_pl” (np. komunikaty bezpieczeństwa
mają krótkie wersje PL obok bazy EN), potem zwykły klucz, potem EN/klicz.
"""
if LANG == "pl":
_pl = T.get("pl", {})
msg = _pl.get(key + "_pl") or _pl.get(key) or T["en"].get(key, key)
else:
msg = T.get(LANG, T["en"]).get(key, T["en"].get(key, key))
if args or kwargs:
return msg.format(*args, **kwargs)
return msg
def _ask_confirm() -> bool:
"""Pytanie potwierdzające (T/n). PAG_YES=1 → zawsze tak.
EOF/brak terminala (stdin zamknięty, np. ssh bez TTY, cron, subprocess
panelu webowego) → NIE – anuluj, nie wykonuj operacji bez potwierdzenia
(inaczej input() rzuca EOFError i pag pada tracebackiem).
Enter → tak (domyślne Y/n).
"""
if os.environ.get("PAG_YES", "") == "1":
print(_("continue_q") + " t (--yes)")
return True
try:
ans = input(_("continue_q")).strip().lower()
except (EOFError, KeyboardInterrupt):
print(f"\n ⚠ {_('no_tty')}")
return False
return not ans or ans in ("t", "y")
# =============================================================================
# ŚCIEŻKI
# =============================================================================
PAG_ROOT = os.environ.get("PAG_ROOT", "/")
PAG_DB = "/var/lib/pag"
PAG_CACHE = "/var/cache/pag"
PAG_CONF = "/etc/pag"
REPO_CACHE = "/var/cache/pag/repos"
REPOS_CONF = "/etc/pag/repos.conf"
REPOS_DIR = PAG_CONF + "/repos" # drop-in: /etc/pag/repos/<nazwa>.conf
INSTALLED_DB = "/var/lib/pag/installed.json"
FILES_DB_SQL = "/var/lib/pag/files.db" # SQLite!
WORLD_FILE = "/var/lib/pag/world"
PINNED_FILE = "/var/lib/pag/pinned.json"
HISTORY_FILE = "/var/lib/pag/history.json"
LOCK_FILE = "/var/lib/pag/pag.lock"
STAGING_DIR = "/.pag_staging" # na tej samej partycji co / (unikamy EXDEV)
PKG_EXT = ".pag"
REPO_CACHE_TTL = 3600
MAX_PKG_SIZE = 2 * 1024 * 1024 * 1024 # 2 GB – maksymalny rozmiar paczki
BASE_MANIFEST = "/usr/share/pag/base.json" # baza z obrazu ISO (pag base/adopt)
# ── UPRZYWILEJOWANE PLIKI ────────────────────────────────────────────────────
# Domyślnie pag zdejmuje bit SUID i wymusza właściciela root:root. Niektóre
# programy MUSZĄ jednak zachować setuid/setgid albo grupę inną niż root
# (/usr/bin/sudo, /usr/bin/passwd, /usr/bin/unix_chkpwd,
# /usr/lib/dbus-daemon-launch-helper). Bez tego tracą możliwość działania
# po aktualizacji przez pag (np. użytkownik nie zmieni hasła).
#
# Bezpieczeństwo: zawartość pakietu jest uwierzytelniona (podpis GPG repo +
# przypięty fingerprint + SHA256 całego pakietu), więc metadane w nim są
# zaufane tak samo jak same pliki. Zachowujemy uprawnienia WYŁĄCZNIE dla
# ścieżek z tej listy (albo z pola `privileged:` receptury, które pagbuild
# zapisuje w metadata.json).
PRIVILEGED_PATHS = frozenset({
"/usr/bin/cdda2wav", "/usr/bin/cdrecord",
"/usr/bin/chage", "/usr/bin/expiry",
"/usr/bin/fcrondyn", "/usr/bin/fcronsighup", "/usr/bin/fcrontab",
"/usr/bin/firejail", "/usr/bin/fping",
"/usr/bin/fusermount", "/usr/bin/fusermount3",
"/usr/bin/gpasswd", "/usr/bin/ksu", "/usr/bin/lockfile",
"/usr/bin/mount", "/usr/bin/mount.nfs",
"/usr/bin/newgidmap", "/usr/bin/newuidmap",
"/usr/bin/passwd", "/usr/bin/ping", "/usr/bin/ping6",
"/usr/bin/procmail", "/usr/bin/readcd", "/usr/bin/rscsi",
"/usr/bin/screen", "/usr/bin/screen-5.0.2",
"/usr/bin/sg", "/usr/bin/su", "/usr/bin/sudo", "/usr/bin/traceroute",
"/usr/bin/umount", "/usr/bin/unix_chkpwd", "/usr/bin/wall",
"/usr/lib/dbus-daemon-launch-helper",
"/usr/lib/httpd/suexec",
"/usr/lib/utempter/utempter",
"/usr/lib/xscreensaver/xscreensaver-auth",
"/usr/libexec/libgtop_server2",
"/usr/libexec/qemu-bridge-helper",
"/usr/libexec/ssh-keysign",
"/usr/libexec/xf86-video-intel-backlight-helper",
"/opt/google/chrome/chrome-sandbox",
"/opt/gitkraken/chrome-sandbox",
"/opt/VirtualBox/VBoxHeadless", "/opt/VirtualBox/VBoxNetAdpCtl",
"/opt/VirtualBox/VBoxNetDHCP", "/opt/VirtualBox/VBoxVolInfo",
"/opt/VirtualBox/VirtualBoxVM",
})
ALLOWED_PKG_RE = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9._+@-]*$')
# Bezpieczeństwo / audyt
AUDIT_LOG = "/var/log/pag/audit.log" # dziennik operacji krytycznych (hooki, self-update)
TRUST_DB = "/etc/pag/trusted.json" # mapa repo_url → fingerprint klucza podpisującego
HOOK_API_VERSION = "1" # wersjonowane API hooków (env PKG_HOOK_API)
# =============================================================================
# IMMUTABLE OS – DEPLOYMENTY
# =============================================================================
# Model: zamiast mutować /, każda operacja tworzy NOWY deployment.
# /var, /etc, /home są współdzielone między deploymentami.
#
# STRUKTURA:
# /.deployments/
# active → 20260723T120000 (symlink do aktywnego)
# 20260723T120000/
# usr/ bin/ lib/ lib64/ ... (pełny system)
# var → /var (symlink do współdzielonego)
# etc → /etc
# home → /home
# ...
#
# Jak to działa:
# 1. pag install → kopiuje active → nowy deployment + nakłada zmiany → switch symlinka
# 2. pag remove → kopiuje active → nowy deployment - usuwa pliki → switch symlinka
# 3. pag deploy-rollback → przełącza active symlink na poprzedni deployment
# 4. Przy starcie systemu: initrd montuje /.deployments/active jako /
# =============================================================================
DEPLOYMENTS_DIR = "/.deployments"
ACTIVE_LINK = "/.deployments/active"
DEPLOYMENTS_DB = "/var/lib/pag/deployments.json"
# Ścieżki współdzielone – NIE wchodzą do deploymentu (są symlinkami do /...)
SHARED_PATHS = {
"/var", "/etc", "/home", "/root", "/tmp", "/run",
"/dev", "/proc", "/sys", "/mnt", "/media", "/srv",
# /boot współdzielone: jądro + initramfs są wspólne dla wszystkich
# deploymentów (inaczej każdy trzyma własną ~50-100 MB kopię). GRUB
# wskazuje na /.deployments/<id>/boot → symlink do /boot.
"/boot",
"/.deployments", "/.pag_staging",
}
def _is_shared_path(rel: str) -> bool:
"""Sprawdza czy ścieżka należy do katalogów współdzielonych (poza deploymentem)."""
for sp in SHARED_PATHS:
if rel == sp or rel.startswith(sp + "/"):
return True
return False
def _is_config_path(rel: str) -> bool:
"""Czy ścieżka to plik konfiguracyjny (/etc/...)?
Dla takich plików stosujemy .pacnew/.pacsave (zachowanie zmian użytkownika)
zamiast bezwarunkowego nadpisania/usunięcia."""
p = rel.lstrip("/")
return p == "etc" or p.startswith("etc/")
# Wrażliwa konfiguracja auth/systemowa – NIGDY nie nadpisujemy jej po cichu.
# Nadpisanie /etc/pam.d/system-* przez pakiet (np. shadow) potrafi zablokować
# logowanie/sudo („account validation failure”); dla tych ścieżek zawsze robimy
# .pacnew, gdy treść na dysku różni się od tej z pakietu.
_SENSITIVE_CONFIG_RE = re.compile(
r'^(pam\.d/|security/|sudoers$|sudoers\.d/|shadow$|gshadow$|passwd$|group$|'
r'login\.defs$|nsswitch\.conf$|default/useradd$)')
def _is_sensitive_config(rel: str) -> bool:
"""Czy rel to wrażliwy plik /etc (PAM, sudoers, shadow, login.defs…)?"""
p = rel.lstrip("/")
if not p.startswith("etc/"):
return False
return bool(_SENSITIVE_CONFIG_RE.match(p[4:]))
def _get_deployment_root() -> str:
"""Zwraca ścieżkę do aktywnego deploymentu, lub PAG_ROOT jeśli tryb niemutowalny wyłączony."""
if os.environ.get("PAG_IMMUTABLE", "") in ("0", "no", "false", ""):
return PAG_ROOT
if os.path.islink(ACTIVE_LINK):
return os.readlink(ACTIVE_LINK)
if os.path.isdir(ACTIVE_LINK):
return ACTIVE_LINK
# Brak deploymentów – użyj /
return PAG_ROOT
def _load_deployments() -> List[dict]:
"""Wczytuje historię deploymentów."""
if not os.path.exists(DEPLOYMENTS_DB):
return []
try:
return json.load(open(DEPLOYMENTS_DB))
except Exception:
return []
def _save_deployments(deployments: List[dict]):
os.makedirs(os.path.dirname(DEPLOYMENTS_DB), exist_ok=True)
json.dump(deployments, open(DEPLOYMENTS_DB, "w"), indent=2)
def _create_deployment(pkg_names: List[str], action: str) -> Tuple[str, str]:
"""
Tworzy nowy deployment przez skopiowanie aktywnego (CoW) i zwraca jego ścieżkę.
Zwraca (deployment_dir, deployment_id).
"""
deploy_id = datetime.now().strftime("%Y%m%dT%H%M%S")
deploy_dir = os.path.join(DEPLOYMENTS_DIR, deploy_id)
os.makedirs(DEPLOYMENTS_DIR, exist_ok=True)
active = _get_deployment_root()
if os.path.isdir(active) and active != PAG_ROOT:
# Trójstopniowa strategia kopiowania deploymentu:
# 1. reflink (CoW – btrfs, xfs) → 0 MB kopiowane
# 2. hardlink (linki twarde) → 0 MB kopiowane, tylko inody
# 3. zwykłe cp (ostateczność) → pełna kopia
print(f" ⚡ Kopiowanie aktywnego deploymentu...")
copied = False
for method, cmd, label in [
("reflink", ["cp", "--reflink=auto", "-a", active + "/.", deploy_dir + "/"], "CoW (reflink)"),
("hardlink", ["cp", "-al", active + "/.", deploy_dir + "/"], "hardlinki"),
("copy", ["cp", "-a", active + "/.", deploy_dir + "/"], "pełna kopia"),
]:
try:
subprocess.run(cmd, check=True, timeout=600, capture_output=True)
print(f" ✅ Deployment: {deploy_id} ({label})")
copied = True
break
except subprocess.CalledProcessError:
if method == "copy":
raise # ostatnia deska – niech leci wyjątek
continue
if not copied:
raise RuntimeError("Nie udało się skopiować deploymentu żadną metodą")
else:
# Pierwszy deployment – tylko katalogi szkieletowe
# /boot celowo POMINIĘTE – jest współdzielone (SHARED_PATHS); poniższa
# pętla utworzy w deploymencie symlink boot → /boot.
for d in ["/usr", "/lib", "/lib64", "/bin", "/sbin", "/opt"]:
if os.path.isdir(d):
dest = os.path.join(deploy_dir, d.lstrip("/"))
os.makedirs(dest, exist_ok=True)
print(f" ✅ Pierwszy deployment: {deploy_id}")
# Utwórz symlinki do współdzielonych katalogów
for sp in SHARED_PATHS:
link_dst = os.path.join(deploy_dir, sp.lstrip("/"))
if not os.path.lexists(link_dst) and os.path.isdir(sp):
os.symlink(sp, link_dst)
# Zapisz w bazie deploymentów
deployments = _load_deployments()
deployments.append({
"id": deploy_id,
"action": action,
"packages": pkg_names,
"timestamp": datetime.now().isoformat(),
"active": True,
})
# Oznacz poprzednie jako nieaktywne
for d in deployments[:-1]:
d["active"] = False
_save_deployments(deployments)
return deploy_dir, deploy_id
def _switch_deployment(deploy_dir: str) -> bool:
"""Atomowo przełącza aktywny deployment przez podmianę symlinka."""
tmp_link = ACTIVE_LINK + ".new"
if os.path.lexists(tmp_link):
os.remove(tmp_link)
os.symlink(deploy_dir, tmp_link)
os.rename(tmp_link, ACTIVE_LINK) # atomowe na tym samym FS
return True
DEFAULT_REPOS = [
"https://repo.paganlinux.eu/stable/",
]
# =============================================================================
# INICJALIZACJA
# =============================================================================
def ensure_dirs():
for d in [PAG_DB, PAG_CACHE, PAG_CONF, REPO_CACHE, REPOS_DIR, STAGING_DIR, DEPLOYMENTS_DIR]:
os.makedirs(d, exist_ok=True)
for f, default in [
(REPOS_CONF, "\n".join(DEFAULT_REPOS) + "\n"),
(INSTALLED_DB, "{}"),
(PINNED_FILE, "{}"),
(HISTORY_FILE, "[]"),
]:
if not os.path.exists(f):
with open(f, "w") as fh: fh.write(default)
if not os.path.exists(WORLD_FILE):
Path(WORLD_FILE).touch()
if not os.path.exists(GPG_HOME):
os.makedirs(GPG_HOME, exist_ok=True)
os.chmod(GPG_HOME, 0o700)
_gpg_run("--list-keys", capture_output=True)
# Inicjalizuj SQLite
_db_init()
# Wyczyść staging po poprzednim przerwanym buildzie/instalacji
if os.path.isdir(STAGING_DIR):
for entry in os.listdir(STAGING_DIR):
if entry == "backups":
continue # backupy starych wersji – potrzebne do `pag rollback`
path = os.path.join(STAGING_DIR, entry)
try:
if os.path.isfile(path) or os.path.islink(path):
os.unlink(path)
elif os.path.isdir(path):
shutil.rmtree(path, ignore_errors=True)
except OSError:
pass
# =============================================================================
# SQLITE – BAZA PLIKÓW (poprawne zarządzanie połączeniami)
# =============================================================================
from contextlib import contextmanager
@contextmanager
def _db_session(readonly: Optional[bool] = None):
"""Context manager – gwarantuje zamknięcie połączenia.
Gdy katalog bazy nie jest zapisywalny (np. komenda read-only uruchomiona
jako zwykły user), otwieramy połączenie w trybie read-only. Inaczej
`PRAGMA journal_mode=WAL` próbuje pisać i kończy się błędem
„attempt to write a readonly database” zamiast zwrócić wynik.
"""
if readonly is None:
readonly = not os.access(os.path.dirname(FILES_DB_SQL) or ".", os.W_OK)
if readonly:
# Katalog bazy nie jest zapisywalny (np. komenda read-only uruchomiona
# jako zwykły user). Baza jest w trybie WAL, więc SQLite przy `mode=ro`
# próbuje utworzyć plik `-shm` w katalogu – bez prawa zapisu kończy się
# to błędem „attempt to write a readonly database” nawet przy SELECT.
# Dlatego próbujemy zwykłego `mode=ro`, a gdy zawiedzie, otwieramy z
# `immutable=1` (SQLite pomija wtedy WAL/SHM i czyta ostatni
# checkpoint). Przy równoległym zapisie roota odczyt może być chwilowo
# nieaktualny – dla komend diagnostycznych (files/list/verify) to OK.
conn = sqlite3.connect(f"file:{FILES_DB_SQL}?mode=ro", uri=True, timeout=15)
try:
conn.execute("SELECT 1 FROM sqlite_master LIMIT 1")
except sqlite3.OperationalError:
conn.close()
conn = sqlite3.connect(
f"file:{FILES_DB_SQL}?mode=ro&immutable=1", uri=True, timeout=15)
else:
conn = sqlite3.connect(FILES_DB_SQL, timeout=15)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA busy_timeout=15000")
conn.row_factory = sqlite3.Row
try:
yield conn
if not readonly:
conn.commit()
except Exception:
if not readonly:
conn.rollback()
raise
finally:
conn.close()
def _db_init():
"""Tworzy tabele SQLite jeśli nie istnieją."""
with _db_session() as db:
db.execute("""
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
path TEXT NOT NULL,
package TEXT NOT NULL,
sha256 TEXT,
size INTEGER,
is_symlink INTEGER DEFAULT 0,
symlink_target TEXT,
UNIQUE(path, package)
)
""")
db.execute("CREATE INDEX IF NOT EXISTS idx_files_path ON files(path)")
db.execute("CREATE INDEX IF NOT EXISTS idx_files_pkg ON files(package)")
db.execute("""
CREATE TABLE IF NOT EXISTS file_checksums (
path TEXT PRIMARY KEY,
sha256 TEXT NOT NULL,
installed_at TEXT
)
""")
db.commit()
def _db_record_files(pkg_name: str, files: List[dict]):
"""Zapisuje pliki do SQLite (obsługuje symlinki)."""
with _db_session() as db:
# Jawna transakcja – atomowość obu zapisów i szybsze wykrycie blokady
try:
db.execute("BEGIN IMMEDIATE")
except sqlite3.OperationalError:
pass # transakcja już otwarta (implicit)
db.executemany(
"INSERT OR REPLACE INTO files (path, package, sha256, size, is_symlink, symlink_target) "
"VALUES (?,?,?,?,?,?)",
[(f["path"], pkg_name, f.get("sha256",""), f.get("size",0),
f.get("is_symlink", 0), f.get("symlink_target", ""))
for f in files]
)
db.executemany(
"INSERT OR REPLACE INTO file_checksums (path, sha256, installed_at) VALUES (?,?,?)",
[(f["path"], f.get("sha256",""), datetime.now().isoformat())
for f in files if f.get("sha256")]
)
def _db_get_package_files(pkg_name: str) -> List[str]:
with _db_session() as db:
return [r["path"] for r in db.execute(
"SELECT DISTINCT path FROM files WHERE package=?", (pkg_name,)
)]
def _db_get_file_owners(filepath: str) -> List[str]:
"""Zwraca listę pakietów będących właścicielami pliku."""
with _db_session() as db:
return [r["package"] for r in db.execute(
"SELECT package FROM files WHERE path=?", (filepath,)
)]
def _db_remove_package_files(pkg_name: str):
with _db_session() as db:
db.execute("DELETE FROM files WHERE package=?", (pkg_name,))
db.commit()
def _db_get_all_file_checksums() -> Dict[str, str]:
with _db_session() as db:
return {r["path"]: r["sha256"] for r in db.execute("SELECT path, sha256 FROM file_checksums")}
def _db_get_package_checksums(pkg_name: str) -> Dict[str, str]:
"""Sumy SHA256 zapisane dla plików należących do pakietu.
Używane do wykrycia, czy użytkownik zmodyfikował plik konfiguracyjny
(porównanie z sumą z chwili instalacji) – serce mechanizmu .pacnew/.pacsave.
"""
with _db_session() as db:
return {r["path"]: r["sha256"] for r in db.execute(
"SELECT fc.path AS path, fc.sha256 AS sha256 FROM file_checksums fc "
"JOIN files f ON f.path = fc.path WHERE f.package=?", (pkg_name,))}
def _db_count_files() -> int:
with _db_session() as db:
return db.execute("SELECT COUNT(*) FROM files").fetchone()[0]
# =============================================================================
# BLOKADA
# =============================================================================
class DatabaseLock:
"""Blokada plikowa (flock) – jądro zwalnia ją AUTOMATYCZNIE, gdy proces
ginie (kill -9, twardy reset). Stary PID-file miał race condition: po
śmierci pag PID mógł zostać przydzielony obcemu procesowi (PID reuse)
i pag odmawiał działania na zawsze („baza zablokowana”).
"""
def __init__(self):
self._f = None
def __enter__(self):
os.makedirs(os.path.dirname(LOCK_FILE), exist_ok=True)
self._f = open(LOCK_FILE, "w")
try:
# LOCK_NB: rzuca wyjątek zamiast czekać w nieskończoność
fcntl.flock(self._f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
print(f"❌ {_('db_locked')}", file=sys.stderr)
print(f" {_('db_lock_hint', LOCK_FILE)}", file=sys.stderr)
sys.exit(1)
self._f.write(str(os.getpid()))
self._f.flush()
return self
def __exit__(self, *args):
if self._f:
try:
fcntl.flock(self._f, fcntl.LOCK_UN)
except OSError:
pass
self._f.close()
self._f = None
# Uwaga: NIE usuwamy pliku blokady. Stały plik + flock na inode to jedyny
# bezpieczny wzorzec – os.remove(), gdy inny proces trzyma blokadę na starym
# inode, otwiera wyścig (nowy proces blokowałby nowo utworzony inode).
class SelfUpdateLock:
"""Blokada podmiany binarki pag podczas self-update.
Niezależna od DatabaseLock (inny plik) – cmd_self_update może zostać
wywołane poza globalną blokadą (np. z zewnętrznego skryptu), a dwie
równoległe aktualizacje pisałyby do tego samego `dst.new`; `os.replace`
mógłby wtedy podmienić plik w trakcie wykonywania (uszkodzony pag).
"""
def __init__(self):
self._f = None
def __enter__(self):
lock_path = LOCK_FILE + ".self-update"
os.makedirs(os.path.dirname(lock_path), exist_ok=True)
self._f = open(lock_path, "w")
try:
fcntl.flock(self._f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
self._f.close()
self._f = None
print("❌ Inna aktualizacja pag jest w toku – spróbuj ponownie później.",
file=sys.stderr)
sys.exit(1)
self._f.write(str(os.getpid()))
self._f.flush()
return self
def __exit__(self, *args):
if self._f:
try:
fcntl.flock(self._f, fcntl.LOCK_UN)
except OSError:
pass
self._f.close()
self._f = None
# =============================================================================
# POMOCNICZE
# =============================================================================
_ALLOWED_PREFIXES = ("/usr/", "/etc/", "/var/", "/opt/",
"/boot/", "/lib/", # kernel: vmlinuz/System.map + moduły (usrmerge: lib→usr/lib)
# Pliki wewnętrzne paczki .pkg.tar.xz
"metadata.json", "data.tar.xz", "hooks/",
"sums.json")
def _check_path_safety(name: str) -> bool:
# Normalizuj – usuń leading ./
if name.startswith("./"):
name = name[2:]
if name in (".", ""):
return True
# Porównuj z prefiksami BEZ wiodącego '/', by zarówno "/usr/bin/ls", jak i
# wewnętrzne pliki pakietu ("hooks/pre-install", "data.tar.xz") przechodziły.
norm = name.lstrip("/")
for prefix in _ALLOWED_PREFIXES:
p = prefix.lstrip("/").rstrip("/")
if norm == p or norm.startswith(p + "/"):
return True
return False
def _validate_pkg_name(name):
return bool(ALLOWED_PKG_RE.match(name))
def _audit(msg):
from datetime import datetime, timezone
os.makedirs(os.path.dirname(AUDIT_LOG), exist_ok=True)
with open(AUDIT_LOG, "a") as f:
f.write(datetime.now(timezone.utc).isoformat() + " " + msg + "\n")
def _strip_suid(path):
try:
st = os.stat(path)
if st.st_mode & 0o4000:
os.chmod(path, st.st_mode & ~0o4000)
print(f" {_("sec_suid", path=path)}")
except OSError:
pass
def _check_downgrade(pkg_name, new_ver, installed_db):
if pkg_name in installed_db:
old = installed_db[pkg_name].get("version", "0")
if new_ver < old:
print(f" {_("sec_downgrade", pkg=pkg_name, new=new_ver, old=old)}")
return False
return True
def _safe_extractall(tar: tarfile.TarFile, dest: str, *, preserve_perms: bool = True,
privileged=frozenset()):
"""
Bezpieczne rozpakowanie archiwum tar z ochroną przed Directory Traversal.
Działa na Python < 3.12 (gdzie parametr 'filter' w extractall nie istnieje)
oraz na Python 3.12+. W przeciwieństwie do filtra 'data' z Pythona 3.12,
zachowuje bity uprawnień POSIX (SUID, SGID, sticky) – preserve_perms=True.
Ochrona oparta jest na FINALNEJ ścieżce (os.path.realpath), nie tylko na
prostym sprawdzaniu stringa:
- Blokuje ścieżki absolutne i z '..' (path traversal)
- Blokuje symlinki/hardlinki, których cel wychodzi poza dest
- Blokuje zapis "przez" złośliwy symlink, który został wcześniej
rozpakowany (np. katalog → /etc, potem zapis katalog/plik)
- Zachowuje oryginalne uprawnienia plików
"""
dest_real = os.path.realpath(dest)
os.makedirs(dest_real, exist_ok=True)
def _target_within(path: str) -> bool:
try:
return os.path.commonpath([dest_real, os.path.realpath(path)]) == dest_real
except ValueError:
# różne napędy / ścieżki nie da się wspólnie porównać → odrzuć
return False
for member in tar.getmembers():
name = member.name
# --- Ochrona przed Directory Traversal (szybkie string-checki) ---
if name.startswith('/'):
continue
if '..' in name.split('/'):
continue
# Zablokuj bajt NUL i backslash (bugi/obejścia tarfile na niektórych platformach)
if '\x00' in name or '\\' in name:
continue
if not _check_path_safety(name):
print(f" BLOCKED: {name}")
continue
target = os.path.join(dest, name)
# --- Ochrona na podstawie finalnej ścieżki ---
# Jeśli którykolwiek komponent nadrzędny jest (złośliwym) symlinkiem
# wskazującym poza dest, realpath to wykryje – zablokuj zapis.
if not _target_within(target):
print(f" BLOCKED (escape): {name}")
continue
# --- Ochrona dla symlinków i hardlinków ---
if member.issym() or member.islnk():
link = member.linkname
# Szybkie odrzucenie linków absolutnych / z '..'
if link.startswith('/') or '..' in link.split('/'):
continue
# Sprawdź, gdzie realnie prowadzi cel linku (względem katalogu linku)
link_target = os.path.join(os.path.dirname(target), link)
if not _target_within(link_target):
print(f" BLOCKED (link escape): {name} -> {link}")
continue
# Rozpakuj z zachowaniem metadanych. Python 3.12+ wymaga jawnego
# `filter=` (inaczej DeprecationWarning, a w 3.14+ błąd).
# UWAGA: 'fully_trusted' CELOWO pomija wbudowane filtry bezpieczeństwa
# Pythona – to nie przeoczenie. Nasza walidacja powyżej (path traversal,
# NUL/backslash, escape przez symlink, linki wychodzące poza dest) jest
# równoważna lub ostrzejsza, a 'fully_trusted' pozwala zachować bity
# SUID/SGID/sticky, które filtr 'data' by usunął (np. /usr/bin/sudo).
# SUID jest i tak zdejmowany przez _strip_suid() tuż po rozpakowaniu.
try:
if hasattr(tarfile, 'data_filter'):
# Python 3.12+
tar.extract(member, dest, set_attrs=preserve_perms, numeric_owner=False,
filter='fully_trusted')
else:
tar.extract(member, dest, set_attrs=preserve_perms, numeric_owner=False)
except Exception as e:
print(f" ⚠ Nie rozpakowano {name}: {e}")
continue
# Pliki z listy uprzywilejowanej zachowują SUID/SGID; reszta jest
# czyszczona (ochrona przed SUID niespodziewanym w pakiecie).
if "/" + os.path.relpath(target, dest_real) not in privileged:
_strip_suid(target)
def _sha256_file(path: str) -> str:
"""SHA256 pliku. Na Python ≥3.11 używa hashlib.file_digest (pętla w C);
starsze wersje mają fallback z pętlą chunków."""
with open(path, "rb") as f:
if hasattr(hashlib, "file_digest"):
return hashlib.file_digest(f, "sha256").hexdigest()
h = hashlib.sha256()
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def _split_version(v: str):
"""Rozdziela wersję na (release_parts, prerelease_parts).
Przykład: '1.2.0-rc1' → ([1,2,0], ['rc','1']).
"""
v = v.strip().lower().lstrip("v")
# build metadata po '+' jest ignorowane przy porównywaniu (semver)
v = v.split("+", 1)[0]
# prerelease po '-' lub '_' (np. 1.2.0-rc1, 1.2.0_rc1)
if "-" in v:
rel, pre = v.split("-", 1)
elif "_" in v:
rel, pre = v.split("_", 1)
else:
rel, pre = v, ""
nums = []
for part in rel.split("."):
m = re.match(r"(\d+)", part)
nums.append(int(m.group(1)) if m else 0)
pre_parts = [p for p in pre.split(".") if p]
return nums, pre_parts
def _cmp_pre(a, b):
"""Porównuje ciągi identyfikatorów prerelease (reguły semver)."""
for i in range(max(len(a), len(b))):
if i >= len(a):
return -1 # krótszy prerelease jest niższy
if i >= len(b):
return 1
ia, ib = a[i], b[i]
if ia == ib:
continue
na, nb = ia.isdigit(), ib.isdigit()
if na and nb:
return 1 if int(ia) > int(ib) else -1
if na != nb:
return -1 if na else 1 # identyfikator liczbowy < alfanumeryczny
return 1 if ia > ib else -1
return 0
def _cmp_version(a: str, b: str) -> int:
"""Porównuje dwie wersje; zwraca -1/0/1. Obsługuje prerelease (rc1, beta...)."""
a_rel, a_pre = _split_version(a)
b_rel, b_pre = _split_version(b)
# Porównaj część release (brakujące komponenty traktuj jako 0)
for i in range(max(len(a_rel), len(b_rel))):
xa = a_rel[i] if i < len(a_rel) else 0
xb = b_rel[i] if i < len(b_rel) else 0
if xa != xb:
return 1 if xa > xb else -1
# Część release równa → decyduje prerelease.
# Wersja finalna (bez prerelease) jest ZAWSZE nowsza od prerelease.
if not a_pre and not b_pre:
return 0
if not a_pre:
return 1
if not b_pre:
return -1
return _cmp_pre(a_pre, b_pre)
def _version_newer(a: str, b: str) -> bool:
"""True gdy wersja a jest nowsza od b (z poprawną obsługą prerelease)."""
try:
return _cmp_version(a, b) > 0
except Exception:
return a != b
def load_json(path):
try:
with open(path) as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {}
def save_json(path, data):
with open(path, "w") as f:
json.dump(data, f, indent=2)
class PackageInfo:
__slots__ = ("name","version","release","description","dependencies",
"size_bytes","sha256","gpg_fp","repo_url","filename","provides","license",
"provides_so","requires_so","privileged")
def __init__(self, d, repo=""):
self.name = d.get("name","?")
self.version = d.get("version","0")
self.release = d.get("release", 1)
self.description = d.get("description","")
self.dependencies = d.get("dependencies", d.get("depends", []))
self.size_bytes = d.get("size",0)
self.sha256 = d.get("sha256","")
self.gpg_fp = d.get("gpg_fingerprint","")
self.repo_url = repo
self.filename = d.get("filename", f"{self.name}-{self.version}{PKG_EXT}")
self.provides = d.get("provides", []) or []
self.license = d.get("license", []) or []
self.provides_so = d.get("provides_so", []) or []
self.requires_so = d.get("requires_so", []) or []
# Ścieżki, którym wolno zachować setuid/setgid i grupę (z receptury).
self.privileged = d.get("privileged", []) or []
# =============================================================================
# REPOZYTORIA (cache, ETag, GPG)
# =============================================================================
def _parse_repos_config():
"""Parsuje repozytoria z /etc/pag/repos.conf oraz /etc/pag/repos/*.conf.
Format linii: <url> [fingerprint]
Opcjonalny `fingerprint` (40 znaków hex) pozwala przypiąć klucz
podpisujący repo do konkretnego adresu – wtedy TOFU (auto-zaufanie przy
pierwszym użyciu) nie jest potrzebne, a zmiana klucza = błąd bezpieczeństwa.
Drop-iny (np. stable.conf) są czytane alfabetycznie – pozwalają na
wygodne dodawanie repo bez dotykania głównego repos.conf
(np. `echo 'https://repo.paganlinux.eu/stable' > /etc/pag/repos/stable.conf`).
"""
entries = []
def _read_lines(path):
if not os.path.exists(path):
return
for line in open(path):
line = line.strip()
if not line or line.startswith("#"):
continue
parts = line.split()
url = parts[0].rstrip("/")
fp = parts[1].lower() if len(parts) > 1 else ""
entries.append({"url": url, "fingerprint": fp or None})
# 1) Legacy: pojedynczy plik /etc/pag/repos.conf
_read_lines(REPOS_CONF)
# 2) Drop-in: /etc/pag/repos/<nazwa>.conf (sortowane, stabilna kolejność)
if os.path.isdir(REPOS_DIR):
for drop in sorted(os.listdir(REPOS_DIR)):
if drop.endswith(".conf"):
_read_lines(os.path.join(REPOS_DIR, drop))
# Dedupe po URL (zachowaj pierwszy wpis – może mieć fingerprint)
seen, unique = set(), []
for e in entries:
if e["url"] not in seen:
seen.add(e["url"])
unique.append(e)
if not unique:
for url in DEFAULT_REPOS:
unique.append({"url": url, "fingerprint": None})
return unique
def get_repos():
return [e["url"] for e in _parse_repos_config()]
def _repo_pinned_fp(repo_url):
"""Zwraca przypięty fingerprint klucza dla repo (z konfiguracji lub trust DB)."""
by_url = {e["url"]: e["fingerprint"] for e in _parse_repos_config()}
if by_url.get(repo_url):
return by_url[repo_url]
db = _load_trust_db()
fp = db.get(repo_url)
return fp.lower() if fp else None
def _repo_cache_path(url):
return os.path.join(REPO_CACHE, url.replace("://","_").replace("/","_").replace(".","_") + ".json")
def _repo_etag_path(url): return _repo_cache_path(url) + ".etag"
def _repo_ts_path(url): return _repo_cache_path(url) + ".ts"
# Informacja (raz na uruchomienie), gdy cache repozytoriów jest tylko-do-odczytu –
# np. komendy read-only (`pag info`, `pag search`…) jako zwykły user: nie ma sensu
# ani prawa odświeżać /var/cache/pag/repos, więc używamy lokalnej kopii indeksu.
_cache_ro_notice_done = False
def _cache_ro_notice():
global _cache_ro_notice_done
if _cache_ro_notice_done:
return
_cache_ro_notice_done = True
print(f" ⚠ {_('cache_ro', cache=REPO_CACHE)}", file=sys.stderr)
def _load_repo_cache(cp: str) -> Optional[list]:
"""Wczytuje cache indeksu repo; zwraca None gdy brak albo uszkodzony.
Partial write (crash w trakcie zapisu) mógł zostawić obcięty JSON. Zamiast
zwracać pustą listę pakietów (użytkownik myśli, że repo jest puste)
sygnalizujemy None – wywołujący ponowi pobranie albo pokaże ostrzeżenie."""
if not os.path.exists(cp):
return None
try:
with open(cp, "r", encoding="utf-8") as fh:
data = json.load(fh)
pkgs = data.get("packages")
if not isinstance(pkgs, list):
raise ValueError("brak listy 'packages'")
return pkgs
except (OSError, ValueError) as e:
print(f" ⚠ Uszkodzony cache indeksu {cp}: {e}", file=sys.stderr)
return None
def fetch_repo_index(repo_url, force=False):
cp = _repo_cache_path(repo_url)
ep = _repo_etag_path(repo_url)
tp = _repo_ts_path(repo_url)
if not force and os.path.exists(cp) and os.path.exists(tp):
try:
if time.time() - float(open(tp).read().strip()) < REPO_CACHE_TTL:
cached = _load_repo_cache(cp)
if cached is not None:
return cached
# uszkodzony cache – spróbuj odświeżyć z sieci
except (OSError, ValueError):
pass
# --- Cache tylko-do-odczytu (np. `pag info` jako zwykły user) ---
# /var/cache/pag/repos należy do roota. Nie próbuj odświeżać ani pisać –
# zwykły user i tak nie zapisze indeksu; użyj lokalnej kopii (może być
# nieaktualna). Pełne odświeżenie indeksu: sudo pag sync
if not (os.path.isdir(REPO_CACHE) and os.access(REPO_CACHE, os.W_OK)):
if force:
print(f" ❌ {repo_url}: nie można odświeżyć indeksu – {REPO_CACHE} jest tylko-do-odczytu",
file=sys.stderr)
return None
_cache_ro_notice()
cached = _load_repo_cache(cp)
if cached is not None:
return cached
return None
headers = {"User-Agent": "pag/3.0"}
if os.path.exists(tp) and not force:
try:
lm = datetime.fromtimestamp(float(open(tp).read().strip()), tz=timezone.utc)
# Wymuś lokalizację C/POSIX dla nagłówków HTTP, aby unikać problemów z nazwami dni/miesięcy
try:
old_locale = locale.setlocale(locale.LC_TIME)
locale.setlocale(locale.LC_TIME, 'C')
headers["If-Modified-Since"] = lm.strftime("%a, %d %b %Y %H:%M:%S GMT")
locale.setlocale(locale.LC_TIME, old_locale)
except (locale.Error, ValueError):
# Jeśli ustawienie lokalizacji się nie powiedzie, użyj domyślnej
headers["If-Modified-Since"] = lm.strftime("%a, %d %b %Y %H:%M:%S GMT")
except: pass
if os.path.exists(ep) and not force:
try: headers["If-None-Match"] = open(ep).read().strip()
except: pass
# --- Pobranie indeksu (błędy SIECI nie są błędami zapisu cache) ---
try:
req = Request(f"{repo_url}/repo.json", headers=headers)
with urlopen(req, timeout=30) as resp:
etag = resp.headers.get("ETag","")
raw = resp.read()
data = json.loads(raw.decode())
except HTTPError as e:
if e.code == 304:
# Serwer: indeks bez zmian – odśwież tylko znacznik czasu (best-effort)
try:
open(tp,"w").write(str(time.time()))
except OSError:
pass
cached = _load_repo_cache(cp)
if cached is not None:
return cached
# uszkodzona kopia – potraktuj jak brak (ostrzeżenie niżej)
print(f" ⚠ HTTP {e.code} dla {repo_url}", file=sys.stderr)
return None
except Exception as e:
print(f" ⚠ Błąd pobierania indeksu {repo_url}: {e}", file=sys.stderr)
return _load_repo_cache(cp)
# Indeks pobrany – zapisz SUROWE bajty (nie re-serializuj! podpis GPG jest
# nad oryginalnymi bajtami repo.json z serwera) i zweryfikuj podpis.
# Najpierw zapis tymczasowy + weryfikacja GPG, dopiero potem podmiana cp:
# błąd zapisu (np. pełny dysk) nie niszczy starej, zweryfikowanej kopii
# i NIGDY nie zwracamy danych, które nie przeszły weryfikacji.
tmp_path = cp + ".tmp"
try:
with open(tmp_path, "wb") as f:
f.write(raw)
if not _verify_repo_sig(repo_url, tmp_path):
return None # weryfikacja nie powiodła się – stary cache zostaje
os.replace(tmp_path, cp)
# przenieś podpis obok docelowego pliku (marker „repo ma podpis")
for _ext in (".asc", ".sig"):
if os.path.exists(tmp_path + _ext):
try:
os.replace(tmp_path + _ext, cp + _ext)
except OSError:
pass
break
if etag:
try:
open(ep,"w").write(etag)
except OSError:
pass
try:
open(tp,"w").write(str(time.time()))
except OSError:
pass
return data.get("packages",[])
except OSError as e:
print(f" ⚠ Indeks pobrany, ale nie udało się zapisać cache ({REPO_CACHE}): {e}",
file=sys.stderr)
# cp nie został podmieniony (podmiana jest po weryfikacji) – lokalna kopia
# to wciąż stare, zweryfikowane dane
return _load_repo_cache(cp)
finally:
for _p in (tmp_path, tmp_path + ".asc", tmp_path + ".sig"):
try:
os.unlink(_p)
except OSError:
pass
def _verify_repo_sig(repo_url, cache_path) -> bool:
"""Weryfikuje podpis GPG indeksu repozytorium i przypina fingerprint.
FAIL-CLOSED: brak/nieprawidłowy podpis = False (chyba że PAG_INSECURE=1).
Zwraca True jeśli indeks jest zaufany, False jeśli należy go odrzucić.
Model zaufania (TOFU + pinning):
- Pierwszy raz (brak przypiętego fingerprintu) → klucz jest importowany,
a fingerprint zapisywany w /etc/pag/trusted.json z JAWNYM ostrzeżeniem.
To świadomy kompromis wygody i bezpieczeństwa.
- Kolejne uruchomienia: fingerprint jest porównywany z przypiętym.
Zmiana klucza = ❌ SECURITY ERROR (fail-closed), wymagane ręczne:
pag key-trust <repo_url> (po weryfikacji nowego klucza)
"""
insecure = os.environ.get("PAG_INSECURE", "") == "1"
if not os.path.exists(GPG_HOME):
if insecure:
return True # brak GPG home – tryb insecure, akceptuj
print(f" ❌ {repo_url}: brak kluczy GPG – weryfikacja niemożliwa!")
print(f" Ustaw PAG_INSECURE=1 aby pominąć (niezalecane)")
os.remove(cache_path)
return False
sig_path = cache_path + ".sig"
# Podpisy generowane jako .asc (armored) – próbuj .asc, potem .sig
sig_data = None
sig_ext = ""
for ext in (".asc", ".sig"):
try:
req = Request(f"{repo_url}/repo.json{ext}", headers={"User-Agent":"pag/3.0"})
with urlopen(req, timeout=15) as resp:
sig_data = resp.read()
sig_ext = ext
break
except Exception:
continue
if not sig_data:
if insecure:
return True # tryb insecure – akceptuj bez podpisu
print(f" ❌ {repo_url}: NIE MOŻNA POBRAĆ PODPISU repo.json.asc/.sig!")
print(f" Ustaw PAG_INSECURE=1 aby pominąć (niezalecane)")
os.remove(cache_path)
return False
sig_path = cache_path + sig_ext
with open(sig_path, "wb") as f:
f.write(sig_data)
ok, fingerprint = _gpg_verify_fp(sig_path, cache_path)
if not ok:
# Automatyczny import klucza repo przy pierwszym uruchomieniu (TOFU,
# jak apt) – gdy w keyringu brakuje klucza (No public key).
res = _gpg_run("--verify", sig_path, cache_path,
capture_output=True, text=True, timeout=30)
_stderr = res.stderr.decode(errors="replace") if isinstance(res.stderr, bytes) else (res.stderr or "")
if "public key" in _stderr.lower() and "no public key" in _stderr.lower():
try:
with urlopen(Request(f"{repo_url}/paganos.asc", headers={"User-Agent": "pag/3.0"}), timeout=20) as r:
keydata = r.read()
with tempfile.NamedTemporaryFile(delete=False, suffix=".asc") as tmp:
tmp.write(keydata)
tmp.flush()
_gpg_run("--import", tmp.name, capture_output=True, timeout=30)
os.unlink(tmp.name)
print(f" 🔑 Importowano klucz repo z {repo_url}/paganos.asc")
ok, fingerprint = _gpg_verify_fp(sig_path, cache_path)
except Exception:
pass
if not ok:
if insecure:
print(f" ⚠ {repo_url}: nieprawidłowy podpis GPG (PAG_INSECURE – ignoruję)")
return True
os.remove(cache_path)
if not shutil.which(GPG_BINARY):
print(f" ❌ {repo_url}: GPG nie jest zainstalowane – nie można zweryfikować podpisu!")
print(f" Zainstaluj gnupg lub ustaw PAG_INSECURE=1 (niezalecane)")
else:
print(f" ❌ {repo_url}: NIEPRAWIDŁOWY PODPIS GPG indeksu repozytorium!")
return False
# --- Wymuś przypięty fingerprint (TOFU + pinning) ---
pinned = _repo_pinned_fp(repo_url)
if pinned:
if not fingerprint:
if insecure:
print(f" ⚠ {repo_url}: nie można odczytać fingerprintu (PAG_INSECURE – ignoruję)")
return True
os.remove(cache_path)
print(f" ❌ [SECURITY ERROR] {repo_url}: nie można odczytać fingerprintu podpisu!")
print(f" Przypięty klucz: {pinned} – odrzucam indeks.")
return False
if fingerprint != pinned.upper():
if insecure:
print(f" ⚠ {repo_url}: ZMIENIONY KLUCZ PODPISU (PAG_INSECURE – ignoruję)")
return True
os.remove(cache_path)
print(f" ❌ [SECURITY ERROR] {repo_url}: Klucz podpisujący repo uległ zmianie!")
print(f" Oczekiwany: {pinned}")
print(f" Otrzymany: {fingerprint}")
print(f" Jeśli to celowa rotacja klucza: pag key-trust {repo_url}")
return False
return True
if fingerprint:
# Brak przypiętego fingerprintu → TOFU: zapisz go w bazie zaufania.
db = _load_trust_db()
if db.get(repo_url) != fingerprint:
_save_trust_db({**db, repo_url: fingerprint})
print(f" 🔐 Przypięto fingerprint repo {repo_url}: {fingerprint}")
print(f" (TOFU – pierwsze zaufanie. Gdy klucz się zmieni, pag odmówi aktualizacji.)")
print(f" Aby uniknąć TOFU, dopisz fingerprint w /etc/pag/repos.conf.")
return True
def fetch_all_packages(force=False):
all_pkgs = {}
for repo_url in get_repos():
pkgs = fetch_repo_index(repo_url, force)
if pkgs:
for pdata in pkgs:
name = pdata.get("name", pdata.get("filename","?").split("-")[0])
pkg = PackageInfo(pdata, repo_url)
if name not in all_pkgs or _version_newer(pkg.version, all_pkgs[name].version):
all_pkgs[name] = pkg
return all_pkgs
# =============================================================================
# GPG
# =============================================================================
def _verify_pkg_gpg(pkg_path, repo_url=None):
"""Weryfikuje podpis GPG pakietu i (jeśli znamy repo) przypięty fingerprint.
FAIL-CLOSED: brak podpisu = odrzucenie (chyba że PAG_INSECURE=1).
Zwraca (passed: bool, message: str).
"""
insecure = os.environ.get("PAG_INSECURE", "") == "1"
# Brak gnupg = weryfikacja niemożliwa. Bez tej gałęzi użytkownik dostawał
# mylące „NIEPRAWIDŁOWY PODPIS GPG”, mimo że paczka i podpis są w porządku.
if not shutil.which(GPG_BINARY):
if insecure:
return True, "(gpg missing – PAG_INSECURE)"
return False, _("gpg_missing")
sig_path = pkg_path + ".sig"
if not os.path.exists(sig_path) and os.path.exists(pkg_path + ".asc"):
sig_path = pkg_path + ".asc"
if not os.path.exists(sig_path):
if insecure:
return True, "(no signature – PAG_INSECURE)"
return False, "BRAK PODPISU – pakiet odrzucony (ustaw PAG_INSECURE=1 aby pominąć)"
ok, fp = _gpg_verify_fp(sig_path, pkg_path)
if not ok:
if insecure:
return True, "(invalid signature – PAG_INSECURE)"
return False, "NIEPRAWIDŁOWY PODPIS GPG"
# Opcjonalnie: sprawdź, czy podpis pochodzi od klucza przypiętego dla repo.
if repo_url:
pinned = _repo_pinned_fp(repo_url)
if pinned and fp and fp != pinned.upper():
if insecure:
return True, "(pkg signer mismatch – PAG_INSECURE)"
return False, f"PAKIET PODPISANY INNYM KLUCZEM niż repo (oczekiwano {pinned})"
return True, "GPG verified"
def cmd_key_add(source):
ensure_dirs()
if not shutil.which(GPG_BINARY):
print(f"❌ {_('gpg_missing')}"); return 1
if source.startswith("http"):
try:
with urlopen(Request(source, headers={"User-Agent":"pag/3.0"}), timeout=30) as resp:
keydata = resp.read()
with tempfile.NamedTemporaryFile(delete=False, suffix=".gpg") as tmp:
tmp.write(keydata); tmp.flush()
res = _gpg_run("--import", tmp.name, capture_output=True, timeout=30)
os.unlink(tmp.name)
if res.returncode != 0:
print(f"❌ {_('key_add_failed')}"); return 1
except Exception as e:
print(f"❌ Download error: {e}"); return 1
else:
res = _gpg_run("--import", source, capture_output=True, timeout=30)
if res.returncode != 0:
print(f"❌ {_('key_add_failed')}"); return 1
print(f"✅ {_('key_imported')}")
def cmd_key_list():
if not shutil.which(GPG_BINARY):
print(f"❌ {_('gpg_missing')}"); return
if not os.path.exists(GPG_HOME):
print(_("no_keys")); return
result = _gpg_run("--list-keys", "--keyid-format", "LONG",
capture_output=True, text=True, timeout=30)
if result.returncode != 0:
print(f"❌ {_('gpg_missing')}"); return
print(result.stdout or _("no_keys"))
def cmd_key_remove(key_id):
_gpg_run("--batch", "--yes", "--delete-key", key_id,
capture_output=True, timeout=30)
print(f"✅ {_('key_removed', key_id)}")
def _repo_signer_fp(repo_url):
"""Pobiera repo.json + podpis i zwraca fingerprint podpisującego (bez pinningu)."""
repo_url = repo_url.rstrip("/")
try:
with urlopen(Request(f"{repo_url}/repo.json", headers={"User-Agent":"pag/3.0"}), timeout=30) as r:
data = r.read()
except Exception:
return None
sig = None
sig_ext = ".asc"
for ext in (".asc", ".sig"):
try:
with urlopen(Request(f"{repo_url}/repo.json{ext}", headers={"User-Agent":"pag/3.0"}), timeout=20) as r:
sig = r.read()
sig_ext = ext
break
except Exception:
continue
if not sig:
return None
with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as tf:
tf.write(data); tf.flush()
data_path = tf.name
sig_path = data_path + sig_ext
try:
with open(sig_path, "wb") as f:
f.write(sig)
ok, fp = _gpg_verify_fp(sig_path, data_path)
finally:
for p in (data_path, sig_path):
try: os.unlink(p)
except OSError: pass
return fp if ok else None
def cmd_key_trust(repo_url):
"""Przypina fingerprint klucza podpisującego repo (koniec z TOFU dla tego repo)."""
repo_url = repo_url.rstrip("/")
print(f"🔐 Przypinam klucz repo {repo_url}...")
fp = _repo_signer_fp(repo_url)
if not fp:
print(" ❌ Nie można odczytać fingerprintu podpisu (brak/nieudany).")
print(" Upewnij się, że klucz repo jest w keyringu (pag key-add <url|file>).")
return 1
db = _load_trust_db()
_save_trust_db({**db, repo_url: fp})
print(f" ✅ Przypięto {fp} dla {repo_url}")
print(" Od teraz zmiana klucza zostanie zgłoszona jako SECURITY ERROR.")
return 0
def cmd_key_untrust(repo_url):
"""Usuwa przypięcie fingerprintu dla repo (wraca do TOFU)."""
repo_url = repo_url.rstrip("/")
db = _load_trust_db()
if repo_url not in db:
print(f" ℹ {repo_url} nie ma przypiętego fingerprintu.")
return 0
del db[repo_url]
_save_trust_db(db)
print(f" ✅ Usunięto przypięcie dla {repo_url}.")
return 0
def cmd_key_trusted():
"""Listuje przypięte fingerprinty repozytoriów."""
db = _load_trust_db()
if not db:
print(_("no_keys"))
return
for url, fp in sorted(db.items()):
print(f" {url}\n {fp}")
# =============================================================================
# ATOMOWA INSTALACJA (STAGING)
# =============================================================================
def _safe_rename(src: str, dst: str) -> bool:
"""
Atomowe przeniesienie pliku. Jeśli src i dst są na różnych
systemach plików (EXDEV), kopiuje + usuwa źródło.
"""
try:
os.rename(src, dst)
return True
except OSError as e:
if e.errno == 18: # EXDEV – cross-device link
shutil.copy2(src, dst)
os.remove(src)
return True
raise
def _install_file(src: str, rel: str, data_staging: str, sums: dict,
staging: str, journal: list, installed_files: list,
deploy_dir: str = "", backup_dir: str = "",
backup_journal: Optional[list] = None,
old_checksums: Optional[dict] = None,
privileged=frozenset()) -> bool:
"""
Instaluje pojedynczy plik (zwykły lub symlink).
Obsługuje: cross-device rename, symlinki, weryfikację SHA256.
Jeśli deploy_dir jest podany (tryb immutable), pliki systemowe trafiają
do deploymentu, a współdzielone (/var, /etc, ...) bezpośrednio do /.
Jeśli backup_dir jest podany, a pod dst istnieje już plik (upgrade/reinstall),
stara wersja jest przenoszona do backup_dir, by rollback mógł ją przywrócić.
old_checksums: {ścieżka: sha256 z chwili instalacji}. Gdy podane i plik /etc
został zmodyfikowany przez użytkownika, nowa wersja ląduje jako .pacnew
(stary plik NIE jest ruszany) – wzorzec jak w pacmanie.
"""
# W trybie immutable: pliki współdzielone idą do /, reszta do deploymentu
if deploy_dir and _is_shared_path("/" + rel):
dst_root = PAG_ROOT
elif deploy_dir:
dst_root = deploy_dir
else:
dst_root = PAG_ROOT
dst = os.path.join(dst_root, rel)
# --- SYMLINK ---
if os.path.islink(src):
link_target = os.readlink(src)
# Weryfikuj sums.json dla symlinka (hash ścieżki docelowej)
expected = sums.get("/" + rel, "")
if expected:
link_hash = hashlib.sha256(link_target.encode()).hexdigest()
if expected and link_hash != expected:
return False
os.makedirs(os.path.dirname(dst), exist_ok=True)
# Backup istniejącego symlinka (upgrade) – dla poprawnego rollbacku
if backup_dir and backup_journal is not None and os.path.lexists(dst):
try:
backup_path = os.path.join(backup_dir, rel)
os.makedirs(os.path.dirname(backup_path), exist_ok=True)
os.replace(dst, backup_path)
backup_journal.append((backup_path, "/" + rel))
journal.append(("backup", backup_path, dst))
except OSError:
pass
# Jeśli docelowy symlink już istnieje, usuń go
if os.path.islink(dst) or os.path.exists(dst):
os.remove(dst)
os.symlink(link_target, dst)
journal.append(("symlink", "", dst))
installed_files.append({
"path": "/" + rel,
"sha256": hashlib.sha256(link_target.encode()).hexdigest(),
"size": len(link_target),
"is_symlink": True,
"symlink_target": link_target,
})
return True
# --- ZWYKŁY PLIK ---
# Oblicz SHA256
try:
file_sha = _sha256_file(src)
except Exception:
file_sha = ""
# Weryfikuj sums.json
expected = sums.get("/" + rel, "")
if expected and file_sha and file_sha != expected:
return False
# --- .pacnew: NIE nadpisuj pliku konfiguracyjnego, którego nie wolno zgubić ---
# Robimy `<plik>.pacnew`, gdy:
# a) plik jest śledzony i użytkownik go zmodyfikował (hash != zapisanej sumy),
# b) to wrażliwa konfiguracja (/etc/pam.d, /etc/security, sudoers, shadow…),
# a treść na dysku różni się od tej z pakietu – chroni auth przed
# cichym nadpisaniem przez `install`/`upgrade` (np. przez shadow).
# NIE robimy .pacnew, jeśli plik na dysku jest identyczny z nowym albo był
# niezmieniony od instalacji (wtedy nadpisanie jest bezpieczne).
old_sha = (old_checksums or {}).get("/" + rel, "")
if file_sha and _is_config_path(rel) and os.path.isfile(dst) and not os.path.islink(dst):
tracked_modified = bool(old_sha) and file_sha != old_sha
if tracked_modified or _is_sensitive_config(rel):
try:
cur_sha = _sha256_file(dst)
except OSError:
cur_sha = ""
if cur_sha and cur_sha != file_sha and not (old_sha and cur_sha == old_sha):
pacnew = dst + ".pacnew"
try:
_safe_rename(src, pacnew)
try:
os.chown(pacnew, 0, 0)
except (OSError, PermissionError):
pass
# Journal (do cofnięcia przy nieudanej transakcji) – ale NIE
# zapisujemy .pacnew w bazie plików: to artefakt użytkownika.
journal.append(("file", src, pacnew))
print(f" ⚠ {_('conf_pacnew', path=pacnew)}")
return True
except OSError:
pass # nie udało się – kontynuuj normalną instalację
# Utwórz katalog docelowy
os.makedirs(os.path.dirname(dst), exist_ok=True)
# Backup istniejącego pliku (upgrade) – dla poprawnego rollbacku
if backup_dir and backup_journal is not None and os.path.lexists(dst):
try:
backup_path = os.path.join(backup_dir, rel)
os.makedirs(os.path.dirname(backup_path), exist_ok=True)
os.replace(dst, backup_path)
backup_journal.append((backup_path, "/" + rel))
journal.append(("backup", backup_path, dst))
except OSError:
pass
# Atomowe przeniesienie (z fallbackiem dla cross-device).
# Zachowuje bity uprawnień (SUID/SGID/sticky) – NIE używamy filter='data'.
_safe_rename(src, dst)
if "/" + rel in privileged:
# Plik uprzywilejowany: zostaw właściciela, grupę i bity z pakietu
# (np. sudo 4755 root:root, qemu-bridge-helper 4750 root:kvm).
try:
st = os.stat(dst)
if st.st_mode & 0o6000:
_audit(f"privileged {dst} mode={oct(st.st_mode & 0o7777)} "
f"uid={st.st_uid} gid={st.st_gid}")
except OSError:
pass
else:
# Wymuś właściciela root:root. UWAGA: os.chown() NIE czyści bitów SUID/SGID.
try:
os.chown(dst, 0, 0)
except (OSError, PermissionError):
# Na niektórych systemach plików (tmpfs, fat) chown może się nie powieść
pass
journal.append(("file", src, dst))
installed_files.append({
"path": "/" + rel,
"sha256": file_sha,
"size": os.path.getsize(dst),
"is_symlink": False,
})
return True
def _atomic_install(pkg_path: str, pkg: PackageInfo, deploy_dir: str = "",
backup_dir: str = "",
old_checksums: Optional[dict] = None) -> Tuple[bool, List[dict], List[Tuple[str, str]]]:
"""
Rozpakowuje do staging area, potem atomowo przenosi pliki.
Jeśli deploy_dir podany – instaluje do deploymentu (tryb immutable).
Zwraca (success, [lista plików z SHA256], [(backup_path, dst), ...]).
"""
staging = tempfile.mkdtemp(dir=STAGING_DIR, prefix=f".staging-{pkg.name}-")
# Wbudowana lista + ścieżki zadeklarowane przez recepturę (metadata.json).
priv = PRIVILEGED_PATHS | frozenset(pkg.privileged or ())
journal = []
installed_files = []
backup_journal: List[Tuple[str, str]] = []
try:
# Rozpakuj .pkg.tar.xz → staging (bezpieczne – ochrona Directory Traversal)
with tarfile.open(pkg_path, "r:xz") as tf:
_safe_extractall(tf, staging)
data_tar = os.path.join(staging, "data.tar.xz")
if not os.path.exists(data_tar):
shutil.rmtree(staging, ignore_errors=True)
return False, [], backup_journal
# Rozpakuj data.tar.xz → staging/data (bezpieczne – ochrona Directory Traversal)
data_staging = os.path.join(staging, "data")
os.makedirs(data_staging, exist_ok=True)
with tarfile.open(data_tar, "r:xz") as tf:
_safe_extractall(tf, data_staging, privileged=priv)
# Wczytaj sums.json
sums_path = os.path.join(data_staging, "sums.json")
sums = json.load(open(sums_path)) if os.path.exists(sums_path) else {}
# Hook pre-install (przed przeniesieniem plików do systemu)
_run_hook(os.path.join(staging, "hooks"), "pre-install", pkg)
# Przenieś pliki: staging/data/* → /
for root, dirs, files in os.walk(data_staging):
# Odtwórz katalogi z pakietu – w tym PUSTE (np. /etc/pulse/default.pa.d).
# Pętla plików tworzy tylko rodziców instalowanych plików, przez co
# puste katalogi z data.tar.xz ginęły przy instalacji.
for d in dirs:
src_dir = os.path.join(root, d)
rel_dir = os.path.relpath(src_dir, data_staging)
if deploy_dir and _is_shared_path("/" + rel_dir):
dst_root = PAG_ROOT
elif deploy_dir:
dst_root = deploy_dir
else:
dst_root = PAG_ROOT
dst_dir = os.path.join(dst_root, rel_dir)
if not os.path.isdir(dst_dir):
try:
os.makedirs(dst_dir, exist_ok=True)
except OSError:
pass
for fname in files:
if fname == "sums.json":
continue
src = os.path.join(root, fname)
rel = os.path.relpath(src, data_staging)
ok = _install_file(src, rel, data_staging, sums,
staging, journal, installed_files, deploy_dir,
backup_dir, backup_journal,
old_checksums=old_checksums,
privileged=priv)
if not ok:
# Cofnij wszystkie operacje
_rollback_journal(journal, staging)
return False, [], backup_journal
# Odbuduj cache ikon GTK dla motywów dotkniętych instalacją.
# Bez icon-theme.cache aplikacje GTK nie widzą ikon mimo obecności
# motywu (np. /usr/share/icons/Papirus). Pomijamy, gdy narzędzie
# nie jest zainstalowane.
_icon_dirs = set()
for f in installed_files:
fp = f.get("path", "") or ""
if fp.startswith("/usr/share/icons/"):
_rest = fp[len("/usr/share/icons/"):]
_theme = _rest.split("/", 1)[0]
if _theme:
_icon_dirs.add(os.path.join(PAG_ROOT, "usr/share/icons", _theme))
if _icon_dirs:
try:
subprocess.run(["gtk-update-icon-cache", "--version"],
capture_output=True, timeout=10)
for _d in sorted(_icon_dirs):
if os.path.isdir(_d):
subprocess.run(["gtk-update-icon-cache", "-f", "-q", _d],
capture_output=True, timeout=300)
except Exception:
pass
# Uruchom hooki post-install
hooks_dir = os.path.join(staging, "hooks")
_run_hook(hooks_dir, "post-install", pkg)
# Zachowaj hooki na wypadek usunięcia pakietu (pre/post-remove)
try:
if os.path.isdir(hooks_dir):
persisted = os.path.join(PAG_DB, "hooks", pkg.name)
shutil.rmtree(persisted, ignore_errors=True)
shutil.copytree(hooks_dir, persisted)
except Exception:
pass
# Zapisz do SQLite
_db_record_files(pkg.name, installed_files)
shutil.rmtree(staging, ignore_errors=True)
return True, installed_files, backup_journal
except Exception as e:
_rollback_journal(journal, staging)
return False, [], backup_journal
def _refresh_dynamic_linker_cache(deploy_dir: str = "") -> bool:
"""Odświeża cache ld.so po udanej instalacji pakietów."""
ldconfig = shutil.which("ldconfig")
if not ldconfig:
print(" ⚠ Nie znaleziono ldconfig — cache linkera nie został odświeżony.",
file=sys.stderr)
return False
target_root = deploy_dir or PAG_ROOT
command = [ldconfig]
if target_root != "/":
command.extend(["-r", target_root])
try:
subprocess.run(command, check=True, timeout=60,
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
text=True)
return True
except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
detail = getattr(exc, "stderr", None) or str(exc)
print(f" ⚠ Nie udało się odświeżyć cache'a ld.so: {detail.strip()}",
file=sys.stderr)
return False
def _rollback_journal(journal: list, staging_path: str):
"""Cofa wszystkie operacje z journala (odwrotna kolejność)."""
for entry in reversed(journal):
op = entry[0]
if op == "file":
_, src, dst = entry
try:
if os.path.exists(dst) or os.path.islink(dst):
_safe_rename(dst, src)
except Exception:
pass
elif op == "symlink":
_, _, dst = entry
try:
if os.path.islink(dst) or os.path.exists(dst):
os.remove(dst)
except Exception:
pass
elif op == "backup":
# Przywróć starą wersję pliku z backupu (upgrade)
_, bpath, dst = entry
try:
if os.path.lexists(bpath):
os.replace(bpath, dst)
except Exception:
pass
shutil.rmtree(staging_path, ignore_errors=True)
# =============================================================================
# BEZPIECZNE USUWANIE
# =============================================================================
def _safe_remove_files(pkg_name: str, installed_db: dict) -> Tuple[int, List[str]]:
"""
Usuwa pliki pakietu, ale tylko jeśli NIE są współdzielone z innym pakietem.
Zwraca (liczba usuniętych, [lista usuniętych ścieżek]).
"""
pkg_files = _db_get_package_files(pkg_name)
recorded = _db_get_package_checksums(pkg_name)
removed = []
skipped_shared = []
skipped_sensitive = []
for fpath in pkg_files:
owners = _db_get_file_owners(fpath)
# Sprawdź czy inny ZAINSTALOWANY pakiet też jest właścicielem
other_owners = [o for o in owners if o != pkg_name and o in installed_db]
if other_owners:
# Plik współdzielony – tylko usuń wpis w DB, nie kasuj pliku
skipped_shared.append(fpath)
continue
full = os.path.join(PAG_ROOT, fpath.lstrip("/"))
if os.path.isfile(full) or os.path.islink(full):
# Wrażliwej konfiguracji systemowej (/etc/pam.d, sudoers, shadow…)
# NIE kasujemy przy `remove` – jej brak potrafi zablokować logowanie
# i sudo (np. usunięcie system-auth/system-account). Zostaje wpis w bazie.
if _is_sensitive_config(fpath):
skipped_sensitive.append(fpath)
continue
# .pacsave: zachowaj ZMIENIONY plik konfiguracyjny zamiast kasować
# (porównanie z sumą z chwili instalacji), wzorzec jak w pacmanie.
if not os.path.islink(full) and _is_config_path(fpath):
old_sha = recorded.get(fpath, "")
if old_sha:
try:
cur_sha = _sha256_file(full)
except OSError:
cur_sha = ""
if cur_sha and cur_sha != old_sha:
pacsave = full + ".pacsave"
try:
os.replace(full, pacsave)
print(f" ⚠ {_('conf_pacsave', path=pacsave)}")
removed.append(fpath)
continue
except OSError:
pass
os.remove(full)
removed.append(fpath)
# Usuń puste katalogi (od najgłębszych)
dirs = set()
for fpath in removed + skipped_shared:
parent = os.path.dirname(fpath)
while parent and parent != "/":
dirs.add(parent)
parent = os.path.dirname(parent)
for d in sorted(dirs, key=len, reverse=True):
full_d = os.path.join(PAG_ROOT, d.lstrip("/"))
if os.path.isdir(full_d):
try:
os.rmdir(full_d)
except OSError:
pass # nie jest pusty – OK
# Usuń z SQLite
_db_remove_package_files(pkg_name)
if skipped_shared:
print(f" ⚠ {len(skipped_shared)} plików współdzielonych zachowanych")
if skipped_sensitive:
print(f" ⚠ {len(skipped_sensitive)} wrażliwych plików systemowych zachowanych")
return len(removed) + len(skipped_shared) + len(skipped_sensitive), removed
def _remove_stale_files(pkg_name: str, old_files: List[str], new_paths: List[str],
installed_db: dict, deploy_dir: str = "",
backup_dir: str = "", backup_journal: Optional[list] = None) -> Tuple[int, List[str]]:
"""
Po upgrade usuwa pliki starej wersji, których nie ma w nowej.
- Pliki współdzielone z innym zainstalowanym pakietem są ZACHOWYWANE
(usuwany jest tylko wpis z bazy `files` dla tego pakietu).
- Sprząta puste katalogi i wpisy SQLite starej wersji.
Zwraca (liczba usuniętych, [usunięte ścieżki]).
"""
new_set = set(new_paths)
stale = [f for f in old_files if f not in new_set]
if not stale:
return 0, []
root = deploy_dir or PAG_ROOT
removed = []
skipped = 0
for fpath in stale:
owners = _db_get_file_owners(fpath)
other_owners = [o for o in owners if o != pkg_name and o in installed_db]
if other_owners:
# Współdzielony z innym pakietem – tylko usuń wpis z DB dla tego pakietu
skipped += 1
else:
# Wrażliwych plików systemowych NIE kasujemy, gdy nowy pakiet ich już
# nie dostarcza (np. shadow przestaje pakować system-*). Ich brak
# blokuje login/sudo; usuwamy więc tylko wpis w bazie.
if _is_sensitive_config(fpath):
skipped += 1
else:
# Gdy nowa wersja pliku /etc trafiła do .pacnew, plik użytkownika
# MUSI zostać – nie jest „przestarzały”. Zachowujemy też wpis w bazie
# (stara suma), by kolejny upgrade dalej wykrywał zmiany użytkownika.
if _is_config_path(fpath):
_cfg_full = os.path.join(PAG_ROOT, fpath.lstrip("/"))
if os.path.lexists(_cfg_full + ".pacnew"):
skipped += 1
continue
full = os.path.join(root, fpath.lstrip("/"))
if os.path.isfile(full) or os.path.islink(full):
try:
if backup_dir and backup_journal is not None:
backup_path = os.path.join(backup_dir, fpath.lstrip("/"))
os.makedirs(os.path.dirname(backup_path), exist_ok=True)
os.replace(full, backup_path) # przenieś do backupu (rollback)
backup_journal.append((backup_path, fpath))
else:
os.remove(full)
removed.append(fpath)
except OSError:
pass
# Usuń wpis `files` dla tego pakietu (stara wersja już go nie zawiera)
with _db_session() as db:
db.execute("DELETE FROM files WHERE package=? AND path=?", (pkg_name, fpath))
# Usuń puste katalogi (od najgłębszych)
dirs = set()
for fpath in removed:
parent = os.path.dirname(fpath)
while parent and parent != "/":
dirs.add(parent)
parent = os.path.dirname(parent)
for d in sorted(dirs, key=len, reverse=True):
full_d = os.path.join(root, d.lstrip("/"))
if os.path.isdir(full_d):
try:
os.rmdir(full_d)
except OSError:
pass # nie jest pusty – OK
if removed:
print(f" 🧹 Usunięto {len(removed)} nieaktualnych plików ({pkg_name})")
if skipped:
print(f" ⚠ {skipped} plików współdzielonych zachowanych")
return len(removed), removed
def _new_transaction_backup_root() -> str:
"""Katalog na backupy NADPISYWANYCH plików dla bieżącej transakcji.
Tworzony dla KAŻDEJ transakcji, nie tylko upgrade: „świeża” instalacja
pakietu też potrafi nadpisać pliki spoza bazy pag (baza rootfs/ISO).
Gdy transakcja padnie, rollback MUSI mieć co przywrócić – inaczej kasuje
te pliki (tak zniknął m.in. libpam.so.0 i przestał działać sudo)."""
txn = datetime.now().strftime("%Y%m%dT%H%M%S") + "-" + str(os.getpid())
root = os.path.join(STAGING_DIR, "backups", txn)
os.makedirs(root, exist_ok=True)
return root
def _purge_old_backups(keep_root: str = ""):
"""Usuwa backupy starszych transakcji (zostawia bieżący – dla `pag rollback`)."""
base = os.path.join(STAGING_DIR, "backups")
if not os.path.isdir(base):
return
for entry in os.listdir(base):
p = os.path.join(base, entry)
if p != keep_root and os.path.isdir(p):
shutil.rmtree(p, ignore_errors=True)
# =============================================================================
# HOOKI
# =============================================================================
# Hooki uruchamiają dowolny plik z pakietu jako root — to naturalna cecha
# menedżera pakietów (apt/pacman też tak mają), dlatego MUSISZ ufać repozytorium.
# Aby ograniczyć ryzyko:
# - hook dostaje minimalne, "czyste" środowisko (bez LD_PRELOAD, BASH_ENV itp.)
# - hooki można wyłączyć (PAG_NO_HOOKS=1) i ustawić timeout (PAG_HOOK_TIMEOUT)
# - każde uruchomienie jest logowane do /var/log/pag/audit.log
# - hook ma wersjonowane API (PKG_HOOK_API)
# =============================================================================
# Lista wykonanych hooków — trafia do wpisu transakcji (informacja w rejestrze).
_HOOKS_RUN: List[str] = []
def _hook_env(pkg: PackageInfo, hook_name: str) -> dict:
"""Buduje minimalne środowisko dla hooka (bez niebezpiecznych zmiennych)."""
return {
"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
"HOME": "/root",
"LANG": "C.UTF-8",
"LC_ALL": "C.UTF-8",
"PKG_NAME": pkg.name,
"PKG_VERSION": pkg.version,
"PKG_ACTION": hook_name,
"PKG_HOOK_API": HOOK_API_VERSION,
}
def _hook_timeout() -> int:
try:
return max(1, int(os.environ.get("PAG_HOOK_TIMEOUT", "60")))
except Exception:
return 60
def _run_hook(hooks_dir: str, hook_name: str, pkg: PackageInfo) -> bool:
"""Uruchamia skrypt hooka jeśli istnieje.
Zwraca True jeśli hook został WYKONANY (istniał i uruchomiono go), False w
pozostałych przypadkach (brak pliku, wyłączone hooki, błąd). Obsługuje
ograniczone środowisko, timeout, logowanie do audytu i rejestr w transakcji.
"""
hook_path = os.path.join(hooks_dir, hook_name)
if not os.path.exists(hook_path):
return False
if os.environ.get("PAG_NO_HOOKS", "") == "1":
print(f" ⚠ Hook pominięty (PAG_NO_HOOKS=1): {hook_name} dla {pkg.name}")
_audit(f"hook SKIP {hook_name} {pkg.name}-{pkg.version} (PAG_NO_HOOKS=1)")
return False
os.chmod(hook_path, 0o755)
env = _hook_env(pkg, hook_name)
tag = f"{hook_name} {pkg.name}-{pkg.version}"
try:
result = subprocess.run([hook_path], env=env, timeout=_hook_timeout(),
check=False, capture_output=True, text=True,
cwd="/")
_HOOKS_RUN.append(tag)
if result.returncode != 0:
print(f" ⚠ Hook {hook_name} dla {pkg.name} zakończony z kodem {result.returncode}")
if result.stderr:
print(f" {result.stderr.strip()[-200:]}")
_audit(f"hook FAIL {tag} rc={result.returncode}")
else:
_audit(f"hook OK {tag}")
return True
except subprocess.TimeoutExpired:
print(f" ⚠ Hook {hook_name} dla {pkg.name} przekroczył timeout ({_hook_timeout()}s)")
_audit(f"hook TIMEOUT {tag}")
return False
except Exception as e:
print(f" ⚠ Hook {hook_name} dla {pkg.name}: {e}")
_audit(f"hook ERROR {tag}: {e}")
return False
# =============================================================================
# TRANSAKCJE I ROLLBACK
# =============================================================================
def _record_transaction(action, packages, success, snapshot, file_journal=None, hooks=None,
upgrade_backups=None, upgrade_backup_root=""):
history = load_json(HISTORY_FILE) if os.path.exists(HISTORY_FILE) else []
# Rejestr wykonanych hooków – informacja o tym, że uruchomiono kod pakietu
# jako root. Trafia do historii, by dało się później sprawdzić, co się działo.
executed_hooks = list(_HOOKS_RUN) if hooks is None else hooks
_HOOKS_RUN.clear()
entry = {
"action": action, "packages": packages, "success": success,
"timestamp": datetime.now().isoformat(),
"snapshot": snapshot,
"file_journal": file_journal, # lista plików do wycofania
"hooks": executed_hooks, # wykonane hooki (pre/post-install/remove)
}
if upgrade_backups:
entry["upgrade_backups"] = upgrade_backups # {dst: backup_path}
entry["upgrade_backup_root"] = upgrade_backup_root
history.append(entry)
if len(history) > 50:
history = history[-50:]
save_json(HISTORY_FILE, history)
def cmd_history():
if not os.path.exists(HISTORY_FILE):
print(_("no_history")); return
history = load_json(HISTORY_FILE)
if not history:
print(_("no_history")); return
print(f"Ostatnie transakcje ({len(history)}):")
for i, e in enumerate(reversed(history), 1):
icon = "✅" if e["success"] else "❌"
pkgs = ", ".join(e["packages"][:5])
if len(e["packages"]) > 5: pkgs += f" (+{len(e['packages'])-5})"
print(f" {i}. {icon} {e['action']}: {pkgs}")
print(f" {e['timestamp']}")
def cmd_rollback():
if not os.path.exists(HISTORY_FILE):
print(_("no_history")); return 1
history = load_json(HISTORY_FILE)
if not history:
print(_("no_history")); return 1
last = None
for e in reversed(history):
if e["success"] and e.get("snapshot"):
last = e; break
if not last:
print("❌ No snapshot to restore."); return 1
print(f"⏪ Rolling back: {last['action']} ({last['timestamp']})")
print(f" Packages: {', '.join(last['packages'][:10])}")
if not _ask_confirm():
return 0
# Przywróć installed.json
save_json(INSTALLED_DB, last["snapshot"])
# Wycofaj fizyczne pliki (jeśli zapisano journal)
# Deduplikacja: pakiet może zgłosić ten sam plik więcej niż raz (np. przez
# `provides` lub wspólną ścieżkę), a journal z historii mógł zostać zapisany
# przed dodaniem deduplikacji.
file_journal = list(dict.fromkeys(last.get("file_journal", [])))
upgrade_backups = last.get("upgrade_backups", {}) or {}
backup_root = last.get("upgrade_backup_root", "")
# Przywróć stare wersje z backupów (upgrade) – nadpisane i usunięte stale pliki
for dst, bpath in upgrade_backups.items():
full = os.path.join(PAG_ROOT, dst.lstrip("/"))
if bpath and os.path.lexists(bpath):
try:
os.makedirs(os.path.dirname(full), exist_ok=True)
os.replace(bpath, full)
except OSError:
pass
# Usuń nowe pliki (które nie miały poprzedniej wersji)
backed = set(upgrade_backups)
if file_journal:
for fpath in reversed(file_journal):
if fpath in backed:
continue
full = os.path.join(PAG_ROOT, fpath.lstrip("/"))
if os.path.exists(full) or os.path.islink(full):
os.remove(full)
print(f" {_('rollback_files', len(file_journal))}")
# Sprzątanie pustych katalogów + katalogu backupów
dirs = set()
for fpath in file_journal:
parent = os.path.dirname(fpath)
while parent and parent != "/":
dirs.add(parent)
parent = os.path.dirname(parent)
for d in sorted(dirs, key=len, reverse=True):
full_d = os.path.join(PAG_ROOT, d.lstrip("/"))
if os.path.isdir(full_d):
try:
os.rmdir(full_d)
except OSError:
pass
if backup_root:
shutil.rmtree(backup_root, ignore_errors=True)
print(f"✅ {_('rollback_restored')}")
_record_transaction("rollback", last["packages"], True, None)
return 0
# =============================================================================
# INSTALACJA
# =============================================================================
def _install_local_pkg_files(paths, install_succeeded):
"""Instaluje lokalne pliki .pkg.tar.xz (bez repozytorium).
Zgodnie z _atomic_install każdy plik jest instalowany atomowo.
Zwraca (failed_count, installed_files)."""
failed = 0
all_files = []
for p in paths:
p = os.path.abspath(p)
if not os.path.isfile(p):
print(f" ❌ Nie znaleziono pakietu: {p}")
failed += 1
continue
if not _verify_local_pkg_signature(p):
failed += 1
continue
try:
with tarfile.open(p, "r:xz") as tf:
meta = tf.extractfile("metadata.json")
if meta is None:
print(f" ❌ {p}: brak metadata.json")
failed += 1
continue
data = json.loads(meta.read())
except Exception as e:
print(f" ❌ {p}: nie udało się odczytać pakietu ({e})")
failed += 1
continue
pkg = PackageInfo(data, repo="local")
print(f" ↓ {pkg.name}-{pkg.version} (lokalny) ... ", end="", flush=True)
ok, files, _ = _atomic_install(p, pkg)
if ok:
install_succeeded(pkg, files)
all_files.extend(f["path"] for f in files)
print("✅")
else:
print("❌")
failed += 1
return failed, all_files
def _preflight_disk(total_bytes: int) -> bool:
"""Pre-flight przed transakcją: wolne miejsce + mount read-only.
Zwraca False (przerywa instalację) gdy na partycji docelowej brakuje
miejsca na pakiety albo katalog stagingu jest zamontowany read-only
(inaczej instalacja rwałaby się w połowie, zostawiając uszkodzony system).
"""
target = PAG_ROOT or "/"
try:
st = os.statvfs(target)
free = st.f_bavail * st.f_frsize
except OSError:
return True # nie da się sprawdzić – nie blokuj
need_mb = total_bytes // 1048576
free_mb = free // 1048576
if free < total_bytes:
print(f" ❌ Za mało miejsca na dysku: potrzeba ~{need_mb} MB, "
f"wolne {free_mb} MB ({target})")
return False
if free < total_bytes * 3:
print(f" ⚠ Mało miejsca na dysku: wolne {free_mb} MB, "
f"pakiety ~{need_mb} MB (rozpakowane zajmą więcej)")
# Wykryj mount read-only (test zapisu w stagingu)
try:
probe = os.path.join(STAGING_DIR, ".pag-probe")
with open(probe, "w") as f:
f.write("x")
os.remove(probe)
except OSError:
print(f" ❌ {target} jest zamontowane tylko-do-odczytu – nie można instalować.")
return False
return True
def cmd_install(package_names, as_dep=False, upgrade=False):
ensure_dirs()
installed_db = load_json(INSTALLED_DB)
world = load_world()
pinned = load_json(PINNED_FILE)
# Obsługa lokalnych plików .pkg.tar.xz (zbudowanych przez pagbuild) –
# nie wymaga repozytorium ani GPG.
local_files = [p for p in package_names if p.endswith(PKG_EXT) or
(os.sep in p and os.path.isfile(os.path.abspath(p)))]
if local_files:
_local_need = sum(
os.path.getsize(os.path.abspath(p))
for p in local_files if os.path.isfile(os.path.abspath(p))
)
if not _preflight_disk(_local_need):
return 1
def _ok(pkg, files):
installed_db[pkg.name] = {
"version": pkg.version, "release": pkg.release, "description": pkg.description,
"dependencies": pkg.dependencies, "size_bytes": pkg.size_bytes,
"sha256": pkg.sha256, "installed_at": datetime.now().isoformat(),
"repo": "local",
"provides": getattr(pkg, "provides", None) or [],
"provides_so": getattr(pkg, "provides_so", None) or [],
"requires_so": getattr(pkg, "requires_so", None) or [],
}
world.add(pkg.name)
failed_local, _fl = _install_local_pkg_files(local_files, _ok)
save_json(INSTALLED_DB, installed_db)
save_world(world)
if failed_local:
return 1
_refresh_dynamic_linker_cache()
package_names = [n for n in package_names if n not in
[os.path.abspath(x) for x in local_files] and
n not in local_files]
to_install = []
if not package_names:
return 0
# pozostałe argumenty to nazwy pakietów z repo – kontynuuj
repo_pkgs = fetch_all_packages()
if not repo_pkgs:
print(f"❌ {_('no_index')}"); return 1
for name in list(package_names):
if name in pinned:
print(f"⚠ {name} {_('pinned_to')} {pinned[name]} – skipping")
package_names.remove(name)
to_install, missing_deps = _resolve_deps(package_names, repo_pkgs, installed_db)
# ── Pakiety, których NIE MA w repo ani nie są zainstalowane ──
# Zgłoś od razu zamiast mylącego „Do zainstalowania: N (0.00 MB)”
# i prośby o potwierdzenie (np. `pag install steam` gdy steam nie istnieje).
not_found = []
for n in package_names:
real = _resolve_provides(n, repo_pkgs, installed_db)
if real not in repo_pkgs and real not in installed_db \
and not os.path.exists(os.path.abspath(n)):
not_found.append(n)
if not_found:
print(f"\n ❌ {_('pkg_not_found', ', '.join(not_found))}")
print(f" {_('not_found_hint')}")
return 1
# --- Tryb upgrade: pakiety już zainstalowane MUSZĄ zostać ponownie
# zainstalowane z nowszej wersji (zastąpienie w tej samej transakcji).
if upgrade:
# `pag update` przekazuje tu tylko pakiety z NOWSZĄ wersją (już
# przefiltrowane w _pending_updates), a `pag install -f` wymusza
# reinstalację nawet tej SAMEJ wersji – dlatego nie filtrujemy po
# _version_newer.
upgrade_targets = [
name for name in package_names
if name in repo_pkgs
and name in installed_db
and name not in pinned
]
for name in upgrade_targets:
if name not in to_install:
to_install.append(name)
if not to_install and not missing_deps:
print(f"✅ {_('all_installed')}"); return 0
# ── WERYFIKACJA ZALEŻNOŚCI ──────────────────────────────────────────
fatal_missing = _verify_dependencies(to_install, repo_pkgs, installed_db)
if fatal_missing > 0:
print(f"❌ Nie można kontynuować – {fatal_missing} brakujących zależności.")
print(f" Zainstaluj brakujące pakiety lub dodaj repozytoria.")
return 1
so_missing = _verify_so_deps(to_install, repo_pkgs, installed_db)
if so_missing > 0:
print(" Zainstaluj dostawcę biblioteki lub zaktualizuj repozytorium.")
return 1
if not to_install:
print(f"✅ {_('all_installed')}"); return 0
MAX_MB = MAX_PKG_SIZE // 1048576
for n in to_install:
if not _validate_pkg_name(n):
print(f" {_("sec_badname", name=n)}")
return 1
sz = repo_pkgs[n].size_bytes if n in repo_pkgs else 0
if sz > MAX_PKG_SIZE:
mb = sz // 1048576
print(f" {_("sec_toobig", size_mb=mb, max_mb=MAX_MB)}")
return 1
total_size = sum(repo_pkgs[n].size_bytes for n in to_install if n in repo_pkgs)
if not _preflight_disk(total_size):
return 1
print(f"\n📦 {_('to_install', len(to_install), total_size/1048576)}")
for name in to_install:
p = repo_pkgs.get(name)
if p:
if name in installed_db:
marker = " [upgrade]" if upgrade else ""
else:
marker = f" [{_('new')}]"
print(f" {name}-{p.version}{marker}")
# Pakiet z bazy obrazu ISO (nie w installed.json) → ostrzeż, że pliki z ISO
# zostaną zastąpione pakietem z repo (kopia do `pag rollback`).
_base_warn_overwrite(to_install)
if not as_dep and not upgrade:
if not _ask_confirm():
print(_("cancelled")); return 0
snapshot = copy.deepcopy(installed_db)
all_installed_files = []
failed = []
# Pary (pkg, stare_pliki, nowe_pliki) do usunięcia martwych plików po upgrade
stale_candidates = []
# Katalog backupów nadpisywanych plików – dla poprawnego rollbacku.
# Dla KAŻDEJ transakcji: instalacja „nowego” pakietu także nadpisuje pliki
# spoza bazy pag (baza rootfs/ISO), a rollback bez backupu by je skasował.
backup_root = ""
all_backups: List[Tuple[str, str]] = [] # (backup_path, dst)
if to_install:
backup_root = _new_transaction_backup_root()
# --- Dziennik transakcji (dla pełnej atomowości) ---
# Jeśli którykolwiek pakiet zawiedzie, cofamy WSZYSTKIE zainstalowane
# w tej transakcji przez _rollback_transaction().
transaction_journal: List[Tuple[str, str, str]] = [] # (op, src, dst)
# --- Tryb immutable: utwórz nowy deployment ---
immutable = os.environ.get("PAG_IMMUTABLE", "") == "1"
deploy_dir = ""
deploy_id = ""
if immutable:
print(f"\n 🏗️ Tworzenie nowego deploymentu...")
deploy_dir, deploy_id = _create_deployment(to_install, "upgrade" if upgrade else "install")
target_root = deploy_dir
else:
target_root = ""
# --- Faza 1: Równoległe pobieranie wszystkich pakietów ---
to_download = [repo_pkgs[name] for name in to_install if name in repo_pkgs]
if len(to_download) > 1:
print(f"\n ⏬ Pobieranie {len(to_download)} pakietów równolegle...")
downloaded = _download_packages_parallel(to_download)
else:
downloaded = {}
# --- Faza 1b: WERYFIKACJA WSZYSTKICH paczek PRZED instalacją ---
# KLUCZOWE: GPG i SHA256 sprawdzamy dla WSZYSTKICH paczek, ZANIM
# zainstalujemy choćby jedną. Inaczej instalowany pakiet może podmienić
# bibliotekę używaną przez sam gpg (np. libassuan 3.0.1 zmienił namespace
# symboli LIBASSUAN_1.0 -> LIBASSUAN_2.0) i wtedy weryfikacja kolejnych
# paczek w tej samej transakcji pada, dając fałszywe
# „NIEPRAWIDŁOWY PODPIS GPG".
verified = {}
_vfail = []
for _name in to_install:
_pkg = repo_pkgs.get(_name)
if not _pkg:
_vfail.append((_name, _('not_found')))
continue
_pp = downloaded.get(_name)
if not _pp:
_pp = _download_pkg(_pkg)
if not _pp:
_vfail.append((_name, _('download_fail')))
continue
_gok, _gmsg = _verify_pkg_gpg(_pp, repo_url=_pkg.repo_url)
if not _gok:
_vfail.append((_name, f"{_('gpg_fail')}: {_gmsg[:60]}"))
continue
if _pkg.sha256 and _sha256_file(_pp) != _pkg.sha256:
_vfail.append((_name, _('sha256_mismatch')))
continue
verified[_name] = _pp
if _vfail:
for _n, _m in _vfail:
print(f" ❌ {_n}: {_m}")
print(" ↩ Przerwano PRZED instalacją – żadna paczka nie została zmieniona.")
return 1
# --- Faza 2: Instalacja – JEDNA nadpisywana linia postępu (jak przy
# pobieraniu), bez ściany tekstu na każdy pakiet. W trybie
# nieinteraktywnym (logi, netinstall instalatora) wypisujemy linię na
# pakiet – tam to pożądane do logu.
t0 = time.time()
stderr_tty = sys.stderr.isatty()
stdout_tty = sys.stdout.isatty()
_bar_last = 0
def _bar_draw(idx: int, name: str) -> None:
nonlocal _bar_last
n = len(to_install)
pct = (idx - 1) / n * 100.0
fl = int(25 * pct / 100)
pbar = "█" * fl + "░" * (25 - fl)
eta_s = ""
if idx > 1:
avg = (time.time() - t0) / (idx - 1)
rem = avg * (n - idx + 1)
eta_s = f" ~{rem:.0f}s" if rem < 60 else f" ~{rem/60:.1f}m"
line = f" 📦 [{pbar}] {idx}/{n} ({pct:.0f}%) {name}{eta_s}"
if stderr_tty:
clear = " " * max(0, _bar_last - len(line))
sys.stderr.write(f"\r{line}{clear}")
sys.stderr.flush()
_bar_last = len(line)
else:
print(line, file=sys.stderr, flush=True)
def _bar_end() -> None:
nonlocal _bar_last
if stderr_tty and _bar_last:
sys.stderr.write("\r" + " " * _bar_last + "\r")
sys.stderr.flush()
_bar_last = 0
_pkg_i = 0
for name in to_install:
pkg = repo_pkgs.get(name)
if not pkg:
_bar_end()
print(f" ❌ {name}: {_('not_found')}")
failed.append(name)
break
_pkg_i += 1
_bar_draw(_pkg_i, f"{name}-{pkg.version}")
# Ścieżka pobrana i ZWERYFIKOWANA w fazie 1b – PRZED jakąkolwiek
# instalacją (patrz komentarz przy fazie 1b). Dzięki temu podmiana
# biblioteki używanej przez gpg nie psuje weryfikacji pozostałych.
pkg_path = verified.get(name)
if not pkg_path:
_bar_end()
print(f" ❌ {name}: {_('download_fail')}")
failed.append(name)
break # przerwij transakcję
# Przed instalacją zapamiętaj pliki starej wersji (potrzebne w upgrade)
old_files = _db_get_package_files(name) if name in installed_db else []
# Stare sumy SHA256 – do wykrycia zmian użytkownika w plikach /etc (.pacnew)
old_checksums = _db_get_package_checksums(name) if name in installed_db else None
# Atomowa instalacja (w upgrade backupuje nadpisywane pliki)
ok, files, backup_j = _atomic_install(pkg_path, pkg, deploy_dir,
backup_dir=backup_root,
old_checksums=old_checksums)
if ok:
installed_db[name] = {
"version": pkg.version, "release": pkg.release, "description": pkg.description,
"dependencies": pkg.dependencies, "size_bytes": pkg.size_bytes,
"sha256": pkg.sha256, "installed_at": datetime.now().isoformat(),
"repo": pkg.repo_url,
"provides": getattr(pkg, "provides", None) or [],
"provides_so": getattr(pkg, "provides_so", None) or [],
"requires_so": getattr(pkg, "requires_so", None) or [],
}
if not as_dep and name in package_names:
world.add(name)
if not stdout_tty:
print(f" ✓ {name}-{pkg.version}", flush=True)
all_installed_files.extend(f["path"] for f in files)
all_backups.extend(backup_j)
# Upgrade: zapamiętaj stare pliki, by po sukcesie usunąć te,
# których nie ma już w nowej wersji.
if upgrade and old_files:
stale_candidates.append((name, old_files, [f["path"] for f in files]))
# Po instalacji kernela – przebuduj initramfs
if _is_kernel_package(name):
_rebuild_initramfs(deploy_dir)
else:
_bar_end()
print(f" ❌ {name}: {_('install_failed')}")
failed.append(name)
break # PRZERWIJ – błąd instalacji
# Deduplikacja journalu – ten sam plik może trafić tu dwukrotnie (np. przez
# `provides`); rollback i historia nie potrzebują duplikatów.
all_installed_files = list(dict.fromkeys(all_installed_files))
# --- Rollback całej transakcji jeśli cokolwiek zawiodło ---
if failed:
_bar_end()
print(f"\n ↩ Cofanie transakcji ({len(failed)} błędów)...")
_rollback_transaction(installed_db, snapshot, all_installed_files,
deploy_dir, immutable, backups=all_backups)
if backup_root:
shutil.rmtree(backup_root, ignore_errors=True)
_record_transaction("upgrade" if upgrade else "install", to_install, False, snapshot)
return 1
# --- Po sukcesie transakcji: usuń nieaktualne pliki starych wersji (upgrade).
# Usunięte pliki trafiają do backupu, aby `pag rollback` mógł je przywrócić.
for pkg_name, old_files, new_paths in stale_candidates:
_remove_stale_files(pkg_name, old_files, new_paths, installed_db, deploy_dir,
backup_root, all_backups)
save_json(INSTALLED_DB, installed_db)
save_world(world)
_record_transaction("upgrade" if upgrade else "install", to_install, True, snapshot,
file_journal=all_installed_files,
upgrade_backups={dst: bp for bp, dst in all_backups} if all_backups else None,
upgrade_backup_root=backup_root)
# Zachowaj backupy bieżącej transakcji (dla `pag rollback`), usuń starsze.
if backup_root:
_purge_old_backups(keep_root=backup_root)
_bar_end()
# --- Tryb immutable: przełącz na nowy deployment ---
if immutable and not failed:
_refresh_dynamic_linker_cache(deploy_dir)
print(f"\n 🔄 Przełączanie na deployment {deploy_id}...")
_switch_deployment(deploy_dir)
print(f" ✅ Aktywny deployment: {deploy_id}")
_update_grub_config()
cmd_deploy_cleanup(keep=5) # Zostawia 5 najnowszych deploymentów
print(f" 💡 Restart wymagany do przeładowania systemu.")
else:
_refresh_dynamic_linker_cache()
# Hooki zbiorcze – raz na transakcję (fc-cache itp.), tylko gdy pliki
# trafiły do realnego systemu (nie do deploymentu).
_process_triggers(all_installed_files)
print(f"\n✅ {_('installed', len(to_install))} ({(time.time()-t0):.0f}s)")
return 0
def _rollback_transaction(installed_db: dict, snapshot: dict,
installed_files: List[str],
deploy_dir: str, is_immutable: bool,
backups: Optional[List[Tuple[str, str]]] = None):
"""
Cofa WSZYSTKIE pakiety zainstalowane w bieżącej transakcji.
Przywraca installed_db do stanu sprzed transakcji.
Usuwa fizyczne pliki z systemu (lub deploymentu w trybie immutable).
Jeśli podano `backups` (upgrade) – przywraca stare wersje nadpisanych plików.
"""
# Przywróć installed_db
installed_db.clear()
installed_db.update(snapshot)
root = deploy_dir if is_immutable else PAG_ROOT
backup_map = {dst: src for src, dst in (backups or [])}
# Przywróć stare wersje z backupów (upgrade)
for dst, bpath in backup_map.items():
full = os.path.join(root, dst.lstrip("/"))
if os.path.lexists(bpath):
try:
os.makedirs(os.path.dirname(full), exist_ok=True)
os.replace(bpath, full)
except OSError:
pass
# Usuń nowe pliki (które nie miały poprzedniej wersji)
for fpath in reversed(installed_files):
if fpath in backup_map:
continue
full = os.path.join(root, fpath.lstrip("/"))
if os.path.isfile(full) or os.path.islink(full):
try:
os.remove(full)
except OSError:
pass
# Wyczyść puste katalogi
dirs_to_check = set()
for fpath in installed_files:
parent = os.path.dirname(fpath)
while parent and parent != "/":
dirs_to_check.add(parent)
parent = os.path.dirname(parent)
for d in sorted(dirs_to_check, key=len, reverse=True):
full_d = os.path.join(root, d.lstrip("/"))
if os.path.isdir(full_d):
try:
os.rmdir(full_d)
except OSError:
pass
# W trybie immutable: usuń nieudany deployment
if is_immutable and deploy_dir:
shutil.rmtree(deploy_dir, ignore_errors=True)
save_json(INSTALLED_DB, snapshot)
# =============================================================================
# USUWANIE
# =============================================================================
def cmd_remove(package_names):
installed_db = load_json(INSTALLED_DB)
world = load_world()
snapshot = copy.deepcopy(installed_db)
removed = []
removed_files = []
total = len(package_names)
for i, name in enumerate(package_names, 1):
if name not in installed_db:
print(f" ⚠ {name}: not installed"); continue
# Pasek postępu
pct = (i - 1) / total * 100
filled = int(25 * pct / 100)
print(f" 🗑 [{'█' * filled + '░' * (25 - filled)}] {i}/{total} ({pct:.0f}%) ", end="\r", file=sys.stderr, flush=True)
print(f"🗑 {name}-{installed_db[name]['version']} ...", end=" ", flush=True)
# Pre-remove hook (jeśli dostępny w staging)
_run_hook_for_installed(name, "pre-remove")
count, rm_files = _safe_remove_files(name, installed_db)
del installed_db[name]
world.discard(name)
removed.append(name)
removed_files.extend(rm_files)
print(f"✅ ({count} files)")
# Post-remove hook + sprzątanie zapisanych hooków
_run_hook_for_installed(name, "post-remove")
shutil.rmtree(os.path.join(PAG_DB, "hooks", name), ignore_errors=True)
save_json(INSTALLED_DB, installed_db)
save_world(world)
_record_transaction("remove", removed, True, snapshot)
print(file=sys.stderr) # wyczyść linię paska postępu
if not removed: return 0
print(f"\n✅ Removed {len(removed)}.")
_process_triggers(removed_files)
orphans = _find_orphans(installed_db, world)
if orphans:
print(f"\n💡 {_('orphans_found', len(orphans), ', '.join(sorted(orphans)[:10]))}")
print(" 'pag remove-orphans' to clean up.")
return 0
def _run_hook_for_installed(pkg_name, hook_name):
"""Próbuje uruchomić hook z katalogu pakietu (jeśli został zapisany)."""
hook_dir = os.path.join(PAG_DB, "hooks", pkg_name)
if os.path.isdir(hook_dir):
ver = load_json(INSTALLED_DB).get(pkg_name, {}).get("version", "")
_run_hook(hook_dir, hook_name, PackageInfo({"name": pkg_name, "version": ver}))
# =============================================================================
# TRIGGERS – hooki zbiorcze (raz na transakcję, nie per pakiet)
# =============================================================================
# Wzorem pacman/dpkg: pakiet/administrator deklaruje zainteresowanie ścieżkami,
# a pasujący trigger uruchamia się DOKŁADNIE RAZ na końcu transakcji
# (np. fc-cache, glib-compile-schemas, update-desktop-database) zamiast po
# każdym pakiecie z osobna.
TRIGGERS_DIR = PAG_CONF + "/triggers"
DEFAULT_TRIGGERS = [
{"name": "font-cache", "paths": ["/usr/share/fonts/", "/usr/local/share/fonts/"],
"run": "fc-cache -fs"},
{"name": "glib-schemas", "paths": ["/usr/share/glib-2.0/schemas/"],
"run": "glib-compile-schemas /usr/share/glib-2.0/schemas"},
{"name": "desktop-database", "paths": ["/usr/share/applications/"],
"run": "update-desktop-database -q /usr/share/applications"},
{"name": "mime-database", "paths": ["/usr/share/mime/"],
"run": "update-mime-database /usr/share/mime"},
]
def _load_triggers() -> List[dict]:
"""Ładuje triggery: domyślne (tylko gdy binarka istnieje) + /etc/pag/triggers/*.json."""
out = []
for t in DEFAULT_TRIGGERS:
bin_name = t["run"].split()[0]
if shutil.which(bin_name):
out.append(dict(t))
if os.path.isdir(TRIGGERS_DIR):
for fn in sorted(os.listdir(TRIGGERS_DIR)):
if not fn.endswith(".json"):
continue
try:
with open(os.path.join(TRIGGERS_DIR, fn)) as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
continue
if isinstance(data, dict):
data = [data]
for t in data:
if isinstance(t, dict) and t.get("name") and t.get("paths") and t.get("run"):
out.append(t)
return out
def _process_triggers(touched_paths: List[str]):
"""Uruchamia pasujące triggery RAZ na końcu transakcji (best-effort)."""
if not touched_paths:
return
if os.environ.get("PAG_NO_HOOKS", "") == "1":
return
import shlex as _shlex
matched = []
for trig in _load_triggers():
if any(path.startswith(p) for p in trig["paths"] for path in touched_paths):
matched.append(trig)
for trig in matched:
run = trig["run"]
print(f" ⚡ Trigger: {trig['name']} ({run})")
try:
r = subprocess.run(_shlex.split(run), capture_output=True, text=True, timeout=120)
_audit(f"TRIGGER {trig['name']}: {run} rc={r.returncode}")
if r.returncode != 0:
print(f" ⚠ rc={r.returncode}: {(r.stderr or r.stdout or '').strip()[:160]}")
except subprocess.TimeoutExpired:
print(f" ⚠ trigger {trig['name']} przekroczył limit czasu (120 s)")
_audit(f"TRIGGER {trig['name']} TIMEOUT")
except Exception as e:
print(f" ⚠ trigger {trig['name']}: {e}")
# =============================================================================
# UPDATE / UPGRADE / LIST / SEARCH / INFO / VERIFY
# =============================================================================
def _cleanup_tmp_files(*paths):
"""Usuwa tymczasowe pliki (np. .pag.new) po nieudanej operacji."""
for p in paths:
try:
if os.path.isfile(p):
os.remove(p)
except OSError:
pass
def cmd_self_update():
"""Aktualizuje samego klienta pag z repo (podpisany /stable/pag).
Kolejność: pobierz → weryfikacja GPG (+ fingerprint repo) → SHA256 →
kontrola składni (compile) → backup → atomowe os.replace.
Podmieniamy plik, z którego pag ZOSTAŁ URUCHOMIONY (a nie hardkodowane
/usr/local/bin/pag): gdy pakiet instaluje /usr/bin/pag, a self-update
pisał do /usr/local/bin/pag, powstawały DWIE kopie o różnych wersjach
i zależnie od PATH `pag --version` pokazywał raz jedną, raz drugą."""
# Dedykowana blokada – patrz SelfUpdateLock. Chroni zapis dst.new oraz
# os.replace przed równoległą aktualizacją (np. cron + ręcznie).
with SelfUpdateLock():
return _self_update_impl()
def _self_update_impl():
repos = get_repos()
if not repos:
print("❌ Brak repozytoriów w konfiguracji.")
return 1
# Aktualizuj bieżący plik (Python ≥3.9 ustawia __file__ jako ścieżkę
# bezwzględną). Poza /usr|/usr/local (np. uruchomienie z checkoutu)
# nie nadpisujemy niczego – wracamy do domyślnej lokalizacji instalacji.
dst = "/usr/local/bin/pag"
try:
cand = os.path.realpath(__file__)
if cand.startswith(("/usr/", "/usr/local/")):
dst = cand
except Exception:
pass
base = repos[0]
dst_new = dst + ".new"
dst_bak = dst + ".bak"
print(f"🔄 Sprawdzam aktualizację pag z {base}...")
try:
with urlopen(Request(f"{base}/pag", headers={"User-Agent": "pag/3.0"}), timeout=20) as r:
data = r.read()
with urlopen(Request(f"{base}/pag.asc", headers={"User-Agent": "pag/3.0"}), timeout=20) as r:
sig = r.read()
except Exception as e:
print(f" ❌ Nie można pobrać pag: {e}")
return 1
# Zapisz nową wersję w katalogu docelowym (ta sama partycja → atomowy rename)
with open(dst_new, "wb") as f:
f.write(data)
with open(dst_new + ".asc", "wb") as f:
f.write(sig)
# --- 1. Weryfikacja podpisu GPG – bez tego nie instalujemy ---
insecure = os.environ.get("PAG_INSECURE", "") == "1"
ok, fp = _gpg_verify_fp(dst_new + ".asc", dst_new)
if not ok:
# Automatyczny import klucza (TOFU) – jak w _verify_repo_sig
res = _gpg_run("--verify", dst_new + ".asc", dst_new,
capture_output=True, text=True)
_stderr = res.stderr.decode(errors="replace") if isinstance(res.stderr, bytes) else (res.stderr or "")
if "public key" in _stderr.lower() and "no public key" in _stderr.lower():
try:
with urlopen(Request(f"{base}/paganos.asc", headers={"User-Agent": "pag/3.0"}), timeout=20) as r:
keydata = r.read()
with tempfile.NamedTemporaryFile(delete=False, suffix=".asc") as tmp:
tmp.write(keydata); tmp.flush()
_gpg_run("--import", tmp.name, capture_output=True, timeout=30)
os.unlink(tmp.name)
print(f" 🔑 Importowano klucz repo z {base}/paganos.asc")
ok, fp = _gpg_verify_fp(dst_new + ".asc", dst_new)
except Exception:
pass
if not ok:
if insecure:
print(" ⚠ Nieprawidłowy podpis aktualizacji (PAG_INSECURE – ignoruję)")
else:
print(" ❌ Nieprawidłowy podpis aktualizacji – nie aktualizuję.")
_cleanup_tmp_files(dst_new, dst_new + ".asc")
return 1
# Sprawdź fingerprint względem przypiętego klucza repo
pinned = _repo_pinned_fp(base)
if pinned:
if not fp:
print(" ❌ Nie można potwierdzić fingerprintu podpisu aktualizacji.")
_cleanup_tmp_files(dst_new, dst_new + ".asc")
return 1
if fp != pinned.upper():
if insecure:
print(" ⚠ Podpis aktualizacji innym kluczem (PAG_INSECURE – ignoruję)")
else:
print(" ❌ [SECURITY ERROR] Podpis aktualizacji innym kluczem niż repo!")
print(f" Oczekiwany: {pinned}, Otrzymany: {fp}")
_cleanup_tmp_files(dst_new, dst_new + ".asc")
return 1
# --- 2. Weryfikacja SHA256 (jeśli repo publikuje pag.sha256) ---
try:
with urlopen(Request(f"{base}/pag.sha256", headers={"User-Agent": "pag/3.0"}), timeout=15) as r:
sha = r.read().decode().strip().split()[0]
if sha:
actual = hashlib.sha256(data).hexdigest()
if actual.lower() != sha.lower():
print(f" ❌ SHA256 niezgodny! Oczekiwano {sha}, jest {actual}")
_cleanup_tmp_files(dst_new, dst_new + ".asc")
return 1
print(" ✅ SHA256 zgodny")
except Exception:
# Brak pag.sha256 w repo – opcjonalne; nie blokuj aktualizacji.
pass
# --- 3. Kontrola składni (nie uruchamiaj uszkodzonego/poddanego edycji pliku) ---
try:
compile(data, "pag", "exec")
except SyntaxError as e:
print(f" ❌ Błąd składni w nowym pag: {e}")
_cleanup_tmp_files(dst_new, dst_new + ".asc")
return 1
m = (re.search(rb'PAG_VERSION\s*=\s*"(\d+\.\d+\.\d+[a-z]?)"', data[:3000])
or re.search(rb"v(\d+\.\d+\.\d+[a-z]?)", data[:3000]))
new_ver = m.group(1).decode() if m else "?"
print(f" ✅ Pobrano pag {new_ver} (obecny {PAG_VERSION}), podpis zweryfikowany")
# --- 4. Backup + atomowa podmiana ---
if os.path.exists(dst):
shutil.copy2(dst, dst_bak)
os.chmod(dst_new, 0o755)
os.replace(dst_new, dst) # atomowe na tym samym FS
try:
if os.path.exists(dst_new + ".asc"):
os.remove(dst_new + ".asc")
except OSError:
pass
print(f" ✅ Zainstalowano nowy pag. Stary zachowany jako {dst_bak}")
# Ostrzeż, gdy w PATH pierwsza jest INNA kopia `pag` – inaczej
# `pag --version` dalej pokaże starą wersję i wygląda to jak brak update.
other = shutil.which("pag")
if other and os.path.realpath(other) != os.path.realpath(dst):
print(f" ⚠ W PATH pierwszy jest inny pag: {other} (zaktualizowano {dst})")
print(f" Usuń starą kopię: sudo rm {other}")
print(" Uruchom ponownie pag, aby użyć nowej wersji.")
return 0
def _candidate_newer(rp, inst):
"""Czy pakiet z repo jest nowszy od zainstalowanego.
Porównuje (version, release): sam bump pkgrel (np. auto-rebuild modułów
po aktualizacji jądra: nvidia-kernel-618 610.57.04-1 -> -2) też musi być
widziany przez `pag update`. Stare rekordy instalacji (bez pola release)
traktujemy jak release=1 – nie generują churnu, dopóki nie wrócą do
reinstalacji/zmiany wersji."""
rv = getattr(rp, "version", "0")
iv = inst.get("version", "0")
if _version_newer(rv, iv):
return True
if rv != iv:
return False
rr = int(getattr(rp, "release", 1) or 1)
ir = int(inst.get("release", 1) or 1)
return rr > ir
def _pending_updates() -> List[str]:
"""Zainstalowane pakiety z nowszą wersją/release w repo (bez przypiętych)."""
installed = load_json(INSTALLED_DB)
pinned = load_json(PINNED_FILE)
repo = fetch_all_packages()
if not repo:
return []
return [n for n, i in installed.items()
if n not in pinned and (rp := repo.get(n)) and _candidate_newer(rp, i)]
# =============================================================================
# BAZA Z OBRAZU ISO → REPO (`pag base`) – pełny rolling
# =============================================================================
# ISO jest zbudowane ZE ŹRÓDEŁ (glibc, gcc, gtk, XFCE, …) – pag tych pakietów
# nie instalował, więc nie ma ich w installed.json/files.db. Skutki:
# • `pag upgrade` ich NIE dotyka (nie ma o nich pojęcia),
# • mimo to `pag install <nazwa>` je NADPISZE – pag robi kopię plików spoza
# swojej bazy (patrz _new_transaction_backup_root), więc nie są nietykalne.
#
# /usr/share/pag/base.json (wypalany przez build-pagan.sh) mówi, co jest w bazie
# obrazu i w jakiej wersji. `pag base` pokazuje stan, a `pag base adopt` dogrywa
# te pakiety z repo – po tym zabiegu pag zarządza nimi normalnie, więc
# `pag upgrade` aktualizuje CAŁY system (rolling), a nie tylko to, co sam
# wcześniej zainstalował.
#
# Nazwy z buildera różnią się od nazw pakietów w repo (builder buduje „libX11”,
# repo ma pakiet „xorg-libx11”), stąd aliasy poniżej. Nazwy bez odpowiednika
# w repo zostają w wersji z obrazu – raport mówi o tym wprost, zamiast udawać,
# że są aktualne.
BASE_NAME_ALIASES = {
"atk": "at-spi2-core",
"dejavu-fonts": "dejavu-ttf",
"fontconfig": "xorg-fontconfig",
"gtk": "gtk3",
"iceauth": "xorg-iceauth",
"libX11": "xorg-libx11",
"libXau": "xorg-libxau",
"libXdmcp": "xorg-libxdmcp",
"libXext": "xorg-libxext",
"libXft": "xorg-libxft",
"libdrm": "xorg-libdrm",
"libnspr": "nspr",
"libxkbfile": "xorg-libxkbfile",
"libxklavier": "xorg-libxklavier",
"linux-pam": "pam",
"pixman": "xorg-pixman",
"sdl2": "sdl",
"setxkbmap": "xorg-setxkbmap",
"xauth": "xorg-xauth",
"xf86-input-libinput": "xorg-xf86-input-libinput",
"xf86-video-fbdev": "xorg-xf86-video-fbdev",
"xinit": "xorg-xinit",
"xkbcomp": "xorg-xkbcomp",
"xkeyboard-config": "xorg-xkeyboard-config",
}
def _base_manifest() -> dict:
"""Manifest bazy z obrazu ({} gdy obraz go nie dostarcza)."""
_d = load_json(os.path.join(PAG_ROOT, BASE_MANIFEST.lstrip("/")))
return _d if isinstance(_d, dict) else {}
def _base_entries() -> list:
"""[(nazwa_z_buildera, wersja_z_obrazu), ...] z manifestu."""
_out = []
for _e in (_base_manifest().get("packages") or []):
if isinstance(_e, str):
if _e:
_out.append((_e, "?"))
elif isinstance(_e, dict) and _e.get("name"):
_out.append((str(_e["name"]), str(_e.get("version") or "?")))
return _out
def _base_repo_name(name: str, repo: dict, installed: dict) -> str:
"""Nazwa pakietu w repo dla nazwy z buildera (alias → provides → brak)."""
if name in repo:
return name
_alias = BASE_NAME_ALIASES.get(name)
if _alias and _alias in repo:
return _alias
_real = _resolve_provides(name, repo, installed)
return _real if _real in repo else ""
def _base_summary(repo: Optional[dict] = None) -> dict:
"""Klasyfikacja pakietów bazy z obrazu wobec repo.
repo=None → pobiera indeks repo (potrzebny do listy „do adoptowania”).
repo={} → bez sieci: rozpoznaje tylko to, czym pag już zarządza.
"""
_entries = _base_entries()
if not _entries:
return {"total": 0, "tracked": 0, "adoptable": [], "unmatched": []}
if repo is None:
repo = fetch_all_packages()
installed = load_json(INSTALLED_DB)
tracked, adoptable, unmatched = 0, [], []
for _n, _v in _entries:
_real = _base_repo_name(_n, repo, installed)
if _n in installed or (_real and _real in installed):
tracked += 1
elif _real:
adoptable.append((_n, _real, _v))
else:
unmatched.append(_n)
return {"total": len(_entries), "tracked": tracked,
"adoptable": adoptable, "unmatched": unmatched}
def _base_hint_unmanaged() -> None:
"""Jednozdaniowa informacja dla `pag sync/update/upgrade`."""
try:
_s = _base_summary()
except Exception:
return
if _s["adoptable"]:
print(f"\n ℹ {_('base_unmanaged_hint', len(_s['adoptable']))}")
def _base_warn_overwrite(names) -> None:
"""Ostrzeżenie w `pag install`, gdy pakiet pochodzi z bazy obrazu ISO."""
_entries = _base_entries()
if not _entries:
return
_known = set()
for _n, _v in _entries:
_known.add(_n)
_al = BASE_NAME_ALIASES.get(_n)
if _al:
_known.add(_al)
_hit = [n for n in names if n in _known]
if _hit:
_listed = ", ".join(_hit[:5]) + (" …" if len(_hit) > 5 else "")
print(f" ℹ {_('base_overwrite', _listed)}")
def cmd_base_status():
"""`pag base` / `pag base status` – co w systemie pochodzi z obrazu,
czym pag już zarządza i co da się zaadoptować z repo."""
_man = _base_manifest()
if not _man:
print(f" ℹ {_('base_no_manifest')}")
return 0
_s = _base_summary()
print(f"\n {_c('bold', _('base_title'))}")
print(f" {'─' * 46}")
_de = _man.get("de") or ""
print(f" {_('base_iso_version'):<34} {_man.get('iso_version', '?')}" + (f" ({_de})" if _de else ""))
print(f" {_('base_total'):<34} {_s['total']}")
print(f" {_('base_tracked'):<34} {_s['tracked']}")
print(f" {_('base_adoptable'):<34} {len(_s['adoptable'])}")
print(f" {_('base_unmatched'):<34} {len(_s['unmatched'])}")
if _s["adoptable"]:
print(f"\n {_('base_adopt_hint')}")
for _iso_n, _real, _v in _s["adoptable"][:10]:
_m = "" if _iso_n == _real else f" ({_iso_n})"
print(f" {_real}-{_v}{_m}")
if len(_s["adoptable"]) > 10:
print(f" … {len(_s['adoptable']) - 10} {_('base_more')}")
if _s["unmatched"]:
print(f"\n {_('base_unmatched_hint')}")
print(" " + ", ".join(_s["unmatched"][:12])
+ (f" … +{len(_s['unmatched']) - 12}" if len(_s["unmatched"]) > 12 else ""))
return 0
def cmd_base_adopt(args):
"""`pag base adopt [pkg...] [--dry-run]` – dograj bazę z obrazu z repo.
Pakiety bazy nie ma w installed.json, więc pag traktuje je jak NOWĄ
instalację: pliki z obrazu trafiają do kopii transakcji (działa
`pag rollback`), a po sukcesie pag zarządza nimi jak każdym innym pakietem.
"""
if not _base_manifest():
print(f" ℹ {_('base_no_manifest')}")
return 1
repo = fetch_all_packages()
if not repo:
print(f"❌ {_('no_index')}")
return 1
_want = [a for a in args if not a.startswith("-") and a != "adopt"]
_s = _base_summary(repo)
plan = _s["adoptable"]
if _want:
_sel = set(_want)
plan = [p for p in plan if p[0] in _sel or p[1] in _sel]
if not plan:
print(f" ℹ {_('base_nothing_selected')}")
return 0
if not plan:
print(f"✅ {_('base_all_managed')}")
return 0
_size = sum(repo[r].size_bytes for _i, r, _v in plan if r in repo)
print(f"\n {_('base_adopt_plan', len(plan), _size / 1048576)}")
for _iso_n, _real, _v in plan[:20]:
_rv = repo[_real].version if _real in repo else "?"
_m = "" if _iso_n == _real else f" ({_iso_n} → {_real})"
print(f" {_real}: {_v} → {_rv}{_m}")
if len(plan) > 20:
print(f" … {len(plan) - 20} {_('base_more')}")
if "-n" in args or "--dry-run" in args:
print(f" {_('base_dry_run')}")
return 0
_rc = cmd_install([r for _i, r, _v in plan])
if _rc == 0:
# Wymienione biblioteki/narzędzia (glibc, gcc, GTK…) są już nowe na dysku,
# ale działające procesy wciąż mają stare w pamięci – restart domyka.
print(f"\n ℹ {_('base_reboot_hint')}")
return _rc
def cmd_update(do_upgrade: bool = False):
"""`pag sync` / `pag update` – odświeżenie indeksów + raport aktualizacji.
sync → tylko odświeżenie indeksów + info: „jest X pakietów do
zaktualizowania – wpisz: pag update".
update → odświeżenie indeksów + AKTUALIZACJA PAKIETÓW (pakiety, nie system).
Pomijamy cache TTL (inaczej nowe pakiety/aktualizacje są niewidoczne nawet
przez godzinę). Pełne pobranie + weryfikacja GPG przy każdym odświeżeniu.
"""
force = True
print("🔄 Refreshing indexes...")
for repo_url in get_repos():
pkgs = fetch_repo_index(repo_url, force=force)
cp = _repo_cache_path(repo_url)
has_sig = os.path.exists(cp + ".sig")
print(f" {'✅' if pkgs is not None else '❌'} {repo_url}: {len(pkgs or [])} pkgs {'🔐' if has_sig else '⚠'}")
print(f"✅ {_('indexes_refreshed')}")
# Powiadomienie o nowszej wersji pag (repo.json["pag_version"])
try:
for r in get_repos():
cp = _repo_cache_path(r)
if os.path.exists(cp):
d = json.load(open(cp))
rv = d.get("pag_version", "")
if rv and rv != PAG_VERSION:
print(f" ⚠ Nowa wersja pag {rv} dostępna – uruchom: pag self-update")
except Exception:
pass
# Raport: pakiety do aktualizacji
pending = _pending_updates()
if not pending:
print(f"✅ {_('all_up_to_date')}")
_base_hint_unmanaged()
return 0
print(f"{_('updates_available', len(pending))}")
installed = load_json(INSTALLED_DB)
repo = fetch_all_packages()
for n in pending:
print(f" {n}: {installed.get(n, {}).get('version', '?')} → {repo[n].version}")
if not do_upgrade:
_base_hint_unmanaged()
return 0 # sync: tylko informacja
if not _ask_confirm():
return 0
_rc = cmd_install(pending, upgrade=True)
_base_hint_unmanaged()
return _rc
def _initramfs_stale() -> bool:
"""Czy initramfs jest starszy niż najnowsze jądro (wymaga przebudowy)."""
try:
kernels = [k for k in os.listdir("/boot") if k.startswith("vmlinuz-")] if os.path.isdir("/boot") else []
if not kernels:
return False
newest = max(os.path.getmtime(os.path.join("/boot", k)) for k in kernels)
initrd = "/boot/initramfs.img"
return (not os.path.exists(initrd)) or os.path.getmtime(initrd) < newest
except Exception:
return False
def cmd_upgrade():
"""`pag upgrade` – aktualizacja SYSTEMU: pakiety + kernel/initramfs/GRUB."""
rc = cmd_update(do_upgrade=True)
if rc != 0:
return rc
# System: dopilnuj initramfs (gdyby kernel był nowszy) + GRUB (immutable)
if _initramfs_stale():
print(" 🐧 Przebudowa initramfs (nowsze jądro)...")
_rebuild_initramfs()
try:
if _load_deployments():
_update_grub_config()
except Exception:
pass
return 0
def cmd_list(installed_only=False):
if installed_only:
db = load_json(INSTALLED_DB)
pinned = load_json(PINNED_FILE)
if not db: print("No packages installed."); return
print(f"Installed ({len(db)}):")
for n, i in sorted(db.items()):
pin = " 📌" if n in pinned else ""
print(f" {n}-{i['version']}{pin} – {i.get('description','')}")
else:
pkgs = fetch_all_packages()
installed = load_json(INSTALLED_DB)
pinned = load_json(PINNED_FILE)
print(f"Available ({len(pkgs)}):")
for n, p in sorted(pkgs.items()):
m = "✓" if n in installed else " "
extra = f" [installed: {installed[n]['version']}]" if n in installed else ""
if n in pinned: extra += " 📌"
print(f" [{m}] {n}-{p.version} – {p.description}{extra}")
def cmd_search(query):
pkgs = fetch_all_packages()
results = [(n,p) for n,p in pkgs.items() if query.lower() in n.lower() or query.lower() in p.description.lower()]
if not results: print(f"❌ No results for: {query}"); return
installed = load_json(INSTALLED_DB)
print(f"Results for '{query}' ({len(results)}):")
for n,p in sorted(results):
print(f" [{'✓' if n in installed else ' '}] {n}-{p.version}")
print(f" {p.description}")
def _smart_search(query: str) -> int:
"""
Inteligentne wyszukiwanie: repo PaganOS + Flathub.
Uruchamiane gdy użytkownik wpisze `pag <nazwa>` zamiast `pag install <nazwa>`.
Pokazuje dostępne źródła i sugeruje komendy instalacji.
"""
# 1. Repo PaganOS
try:
pkgs = fetch_all_packages()
except Exception:
pkgs = {}
repo_lower = [(n, p) for n, p in pkgs.items()
if query.lower() in n.lower() or query.lower() in p.description.lower()]
# 2. Flathub (jeśli dostępny)
flat = _flatpak_search_raw(query) if _check_flatpak(quiet=True) else []
if not repo_lower and not flat:
print(f"\n ❌ '{query}' — nie znaleziono.")
print(f" Repo PaganOS: pag search {query}")
if _check_flatpak(quiet=True):
print(f" Flathub: pag flatpak search {query}")
print(f" Dodaj repo: pag repo-add <url>")
return 1
installed = load_json(INSTALLED_DB)
# ── Repo PaganOS ──
if repo_lower:
exact = [(n, p) for n, p in repo_lower if n.lower() == query.lower()]
show = (exact or repo_lower)[:6]
print(f"\n 📦 PaganOS — '{query}':")
for n, p in sorted(show):
mark = "✓" if n in installed else " "
desc = p.description[:70] if len(p.description) > 75 else p.description
print(f" [{mark}] {n}-{p.version}")
if desc:
print(f" {desc}")
if len(repo_lower) > 6:
print(f" ... i {len(repo_lower) - 6} więcej (pag search {query})")
# ── Flathub ──
if flat:
print(f"\n 📦 Flathub — '{query}':")
for r in flat[:5]:
mark = "✓" if r.get("installed") else " "
name = r.get("name") or r.get("application", "?")
desc = (r.get("description") or "")[:65]
print(f" [{mark}] {name}")
if desc:
print(f" {desc}")
if len(flat) > 5:
print(f" ... i {len(flat) - 5} więcej (pag flatpak search {query})")
# ── Sugestie instalacji ──
print()
if repo_lower:
best = sorted(repo_lower, key=lambda x: (x[0].lower() != query.lower(), -len(x[1].name if hasattr(x[1], 'name') else 0)))[0][0]
if best in installed:
print(f" ✓ {best} jest już zainstalowany ({installed[best]['version']})")
else:
print(f" 💡 sudo pag install {best}")
if flat:
best_fp = flat[0].get("application") or flat[0].get("name", query)
print(f" 💡 pag flatpak install {best_fp}")
return 0
def cmd_info(name):
pkgs = fetch_all_packages()
p = pkgs.get(name)
info = load_json(INSTALLED_DB).get(name)
if not p and not info: print(f"❌ '{name}' not found."); return 1
print(f"📦 {name}")
if p:
print(f" Version (repo): {p.version}")
print(f" Description: {p.description}")
print(f" Size: {p.size_bytes/1048576:.1f} MB")
print(f" SHA256: {p.sha256[:32]}...")
print(f" GPG: {p.gpg_fp or 'none'}")
print(f" Dependencies: {', '.join(p.dependencies) if p.dependencies else '(none)'}")
if info:
print(f" Installed: {info['version']} ({info.get('installed_at','?')})")
def cmd_files(name):
if name not in load_json(INSTALLED_DB):
print(f"❌ '{name}' not installed."); return 1
files = _db_get_package_files(name)
print(f"Files in {name} ({len(files)}):")
for f in sorted(files): print(f" {f}")
def cmd_verify(deep=False):
installed = load_json(INSTALLED_DB)
if not installed: print("Nothing to verify."); return
errors = []
# Wczytaj WSZYSTKIE sumy RAZ (nie w pętli!) – przy 50k plików wywołanie
# _db_get_all_file_checksums() per-plik dawało O(n²) i godziny zamiast sekund.
all_checksums = _db_get_all_file_checksums() if deep else {}
for name in installed:
for fpath in _db_get_package_files(name):
full = os.path.join(PAG_ROOT, fpath.lstrip("/"))
if not (os.path.exists(full) or os.path.islink(full)):
errors.append(f" ❌ {name}: missing {fpath}")
elif deep:
expected = all_checksums.get(fpath, "")
if expected:
actual = _sha256_file(full)
if actual != expected:
errors.append(f" ❌ {name}: SHA256 mismatch {fpath}")
if errors:
print(f"❌ {_('verify_errors', len(errors))}")
for e in errors[:50]: print(e)
return 1
total = _db_count_files()
print(f"✅ {_('verify_ok', total)}")
# =============================================================================
# PINNING / CLEAN / ORPHANS / REPO / FLATPAK
# =============================================================================
def cmd_pin(name, version=""):
pinned = load_json(PINNED_FILE)
if version:
pinned[name] = version
else:
info = load_json(INSTALLED_DB).get(name, {})
pinned[name] = info.get("version", "?")
save_json(PINNED_FILE, pinned)
print(f"📌 {name} {_('pinned_to')} {pinned[name]}")
def cmd_unpin(name):
pinned = load_json(PINNED_FILE)
if name in pinned:
del pinned[name]; save_json(PINNED_FILE, pinned)
print(f"🔓 {name} {_('unpinned')}")
else:
print(f"⚠ {name} {_('not_pinned')}")
def cmd_pinned():
pinned = load_json(PINNED_FILE)
if not pinned: print(_("no_pinned")); return
print(_("pinned_list", len(pinned)))
for n,v in sorted(pinned.items()): print(f" 📌 {n} = {v}")
def cmd_clean():
if os.path.isdir(PAG_CACHE):
count = size = 0
for f in os.listdir(PAG_CACHE):
fp = os.path.join(PAG_CACHE, f)
if os.path.isfile(fp):
size += os.path.getsize(fp); os.remove(fp); count += 1
print(f"✅ {_('cache_cleared', count, size/1048576)}")
def cmd_remove_orphans():
installed = load_json(INSTALLED_DB)
world = load_world()
orphans = _find_orphans(installed, world)
if not orphans: print("✅ No orphans."); return
print(f"Orphans ({len(orphans)}):")
for n in sorted(orphans): print(f" {n}-{installed[n]['version']}")
if not _ask_confirm():
return
cmd_remove(list(orphans))
# =============================================================================
# PROVIDES – PAKIETY WIRTUALNE
# =============================================================================
PROVIDES_MAP = {
"pkgconfig(glib-2.0)": "glib",
"pkgconfig(gobject-introspection-1.0)": "gobject-introspection",
"pkgconfig(gtk+-3.0)": "gtk",
"pkgconfig(gtk4)": "gtk",
"pkgconfig(zlib)": "zlib",
"pkgconfig(libffi)": "libffi",
"pkgconfig(expat)": "expat",
"pkgconfig(libsystemd)": "systemd",
"pkgconfig(dbus-1)": "dbus",
"pkgconfig(mount)": "util-linux",
"pkgconfig(blkid)": "util-linux",
"pkgconfig(libcap)": "libcap",
"pkgconfig(liblzma)": "xz",
"pkgconfig(libzstd)": "zstd",
"pkgconfig(bzip2)": "bzip2",
"pkgconfig(libcurl)": "curl",
"pkgconfig(openssl)": "openssl",
"pkgconfig(libpcre2-8)": "pcre2",
"pkgconfig(libxml-2.0)": "libxml2",
"pkgconfig(libxslt)": "libxslt",
"pkgconfig(freetype2)": "freetype",
"pkgconfig(fontconfig)": "fontconfig",
"pkgconfig(harfbuzz)": "harfbuzz",
"pkgconfig(cairo)": "cairo",
"pkgconfig(pango)": "pango",
"pkgconfig(xt)": "xorg-libxt",
"pkgconfig(xmu)": "xorg-libxmu",
"pkgconfig(ice)": "xorg-libice",
"pkgconfig(sm)": "xorg-libsm",
"pkgconfig(x11)": "xorg-libx11",
"pkgconfig(xext)": "xorg-libxext",
"pkgconfig(xrandr)": "xorg-libxrandr",
"pkgconfig(xfixes)": "xorg-libxfixes",
"pkgconfig(xcursor)": "xorg-libxcursor",
"pkgconfig(xinerama)": "xorg-libxinerama",
"pkgconfig(xrender)": "xorg-libxrender",
"pkgconfig(xau)": "xorg-libxau",
"pkgconfig(xcb)": "xorg-libxcb",
"pkgconfig(xdamage)": "xorg-libxdamage",
"pkgconfig(xcomposite)": "xorg-libxcomposite",
"pkgconfig(xft)": "xorg-libxft",
"pkgconfig(xss)": "xorg-libxss",
"pkgconfig(libsoup-3.0)": "libsoup3",
"pkgconfig(libsoup-2.4)": "libsoup2",
"pkgconfig(gdk-pixbuf-2.0)": "gdk-pixbuf2",
"pkgconfig(libpng)": "libpng",
"pkgconfig(libjpeg)": "libjpeg-turbo",
"pkgconfig(libtiff-4)": "libtiff",
"pkgconfig(ffi)": "libffi",
# ── system / baza ──
"pkgconfig(libcrypto)": "openssl",
"pkgconfig(libssl)": "openssl",
"pkgconfig(libudev)": "systemd",
"pkgconfig(libmount)": "util-linux",
"pkgconfig(libblkid)": "util-linux",
"pkgconfig(uuid)": "util-linux",
"pkgconfig(libexpat)": "expat",
"pkgconfig(libpcre)": "pcre",
"pkgconfig(ncursesw)": "ncurses",
"pkgconfig(tinfo)": "ncurses",
"pkgconfig(panel)": "ncurses",
"pkgconfig(readline)": "readline",
"pkgconfig(libseccomp)": "libseccomp",
"pkgconfig(pam)": "linux-pam",
"pkgconfig(libxcrypt)": "libxcrypt",
"pkgconfig(libcrypt)": "libxcrypt",
"pkgconfig(libnsl)": "libnsl",
"pkgconfig(liblz4)": "lz4",
"pkgconfig(libevent)": "libevent",
"pkgconfig(libarchive)": "libarchive",
"pkgconfig(sqlite3)": "sqlite",
"pkgconfig(libpq)": "postgresql",
"pkgconfig(mysqlclient)": "mariadb",
"pkgconfig(json-c)": "json-c",
"pkgconfig(json-glib-1.0)": "json-glib",
"pkgconfig(libunistring)": "libunistring",
"pkgconfig(libidn2)": "libidn2",
"pkgconfig(libpsl)": "libpsl",
"pkgconfig(icu-uc)": "icu",
"pkgconfig(icu-i18n)": "icu",
"pkgconfig(icu-io)": "icu",
"pkgconfig(gnutls)": "gnutls",
"pkgconfig(nettle)": "nettle",
"pkgconfig(hogweed)": "nettle",
"pkgconfig(libgcrypt)": "libgcrypt",
"pkgconfig(libgpg-error)": "libgpg-error",
"pkgconfig(libassuan)": "libassuan",
"pkgconfig(libusb-1.0)": "libusb",
"pkgconfig(libusb)": "libusb",
"pkgconfig(libgudev-1.0)": "libgudev",
"pkgconfig(gudev-1.0)": "libgudev",
"pkgconfig(polkit-gobject-1)": "polkit",
"pkgconfig(polkit-agent-1)": "polkit",
"pkgconfig(libpciaccess)": "libpciaccess",
"pkgconfig(pixman-1)": "pixman",
"pkgconfig(libdrm)": "libdrm",
"pkgconfig(libva)": "libva",
"pkgconfig(libva-drm)": "libva",
"pkgconfig(libva-x11)": "libva",
"pkgconfig(libva-wayland)": "libva",
"pkgconfig(vdpau)": "libvdpau",
"pkgconfig(libvdpau)": "libvdpau",
"pkgconfig(libinput)": "libinput",
"pkgconfig(libevdev)": "libevdev",
"pkgconfig(mtdev)": "mtdev",
# ── grafika / GL / multimedia ──
"pkgconfig(gbm)": "mesa",
"pkgconfig(gl)": "libglvnd",
"pkgconfig(egl)": "libglvnd",
"pkgconfig(glesv2)": "libglvnd",
"pkgconfig(glx)": "libglvnd",
"pkgconfig(vulkan)": "vulkan-loader",
"pkgconfig(libxkbcommon)": "libxkbcommon",
"pkgconfig(xkbcommon)": "libxkbcommon",
"pkgconfig(xkbcommon-x11)": "libxkbcommon",
"pkgconfig(xcb)": "xorg-libxcb",
"pkgconfig(xcb-util)": "xcb-util",
"pkgconfig(xcb-keysyms)": "xcb-util-keysyms",
"pkgconfig(xcb-icccm)": "xcb-util-wm",
"pkgconfig(xcb-cursor)": "xcb-util-cursor",
"pkgconfig(xcb-renderutil)": "xcb-util-renderutil",
"pkgconfig(xcb-image)": "xcb-util-image",
"pkgconfig(xcb-errors)": "xcb-util-errors",
"pkgconfig(wayland-client)": "wayland",
"pkgconfig(wayland-server)": "wayland",
"pkgconfig(wayland-cursor)": "wayland",
"pkgconfig(wayland-egl)": "wayland",
"pkgconfig(wayland-protocols)": "wayland-protocols",
"pkgconfig(gstreamer-1.0)": "gstreamer",
"pkgconfig(gstreamer-base-1.0)": "gstreamer",
"pkgconfig(gstreamer-check-1.0)": "gstreamer",
"pkgconfig(gstreamer-controller-1.0)": "gstreamer",
"pkgconfig(gstreamer-app-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-video-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-audio-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-pbutils-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-fft-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-riff-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-rtp-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-rtsp-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-sdp-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-net-1.0)": "gst-plugins-base",
"pkgconfig(gstreamer-gl-1.0)": "gst-plugins-base",
"pkgconfig(libpulse)": "libpulse",
"pkgconfig(libpulse-simple)": "libpulse",
"pkgconfig(libpulse-mainloop-glib)": "libpulse",
"pkgconfig(alsa)": "alsa-lib",
"pkgconfig(jack)": "jack2",
"pkgconfig(libsamplerate)": "libsamplerate",
"pkgconfig(sndfile)": "libsndfile",
"pkgconfig(libavcodec)": "ffmpeg",
"pkgconfig(libavformat)": "ffmpeg",
"pkgconfig(libavutil)": "ffmpeg",
"pkgconfig(libavfilter)": "ffmpeg",
"pkgconfig(libswscale)": "ffmpeg",
"pkgconfig(libswresample)": "ffmpeg",
"pkgconfig(libpostproc)": "ffmpeg",
"pkgconfig(SDL2)": "sdl2",
"pkgconfig(SDL)": "sdl",
"pkgconfig(SDL2_image)": "sdl2-image",
"pkgconfig(SDL2_ttf)": "sdl2-ttf",
"pkgconfig(SDL2_mixer)": "sdl2-mixer",
"pkgconfig(SDL2_net)": "sdl2-net",
"pkgconfig(libpng16)": "libpng",
"pkgconfig(libwebp)": "libwebp",
"pkgconfig(libwebpmux)": "libwebp",
"pkgconfig(libwebpdemux)": "libwebp",
"pkgconfig(libopenjp2)": "openjpeg2",
"pkgconfig(lcms2)": "lcms2",
"pkgconfig(libheif)": "libheif",
"pkgconfig(libde265)": "libde265",
"pkgconfig(x264)": "x264",
"pkgconfig(x265)": "x265",
# ── glib / gio ──
"pkgconfig(gio-unix-2.0)": "glib",
"pkgconfig(gmodule-2.0)": "glib",
"pkgconfig(gthread-2.0)": "glib",
"pkgconfig(girepository-2.0)": "gobject-introspection",
"pkgconfig(girepository-1.0)": "gobject-introspection",
"pkgconfig(libglib-2.0)": "glib",
"pkgconfig(libgobject-2.0)": "glib",
}
# Ostrzeżenia o wielu dostawcach tej samej wirtualnej nazwy – raz na proces.
_PROVIDES_WARNED = set()
# Cache indeksu provides dla danego obiektu repo: (repo_obj, {virtual: [pkg,...]}).
# Trzymamy referencję do repo, by uniknąć pomyłki przy ponownym użyciu id().
_provides_cache = (None, {})
def _provides_index(repo: dict) -> dict:
"""Buduje (i cache’uje) mapę wirtualna nazwa → lista dostawców w repo.
Pozwala wybrać dostawcę DETERMINISTYCZNIE (posortowanego) zamiast zależeć
od kolejności wstawiania w repo.json, oraz ostrzec o konflikcie provides."""
global _provides_cache
cached_repo, idx = _provides_cache
if cached_repo is repo:
return idx
idx = {}
for _pn, _p in repo.items():
for _prov in (getattr(_p, "provides", None) or []):
idx.setdefault(_prov, []).append(_pn)
for _prov, _pns in idx.items():
if len(_pns) > 1 and _prov not in _PROVIDES_WARNED:
_PROVIDES_WARNED.add(_prov)
_sorted = sorted(_pns)
print(f" ⚠ {_('provides_conflict', name=_prov, providers=', '.join(_sorted), chosen=_sorted[0])}",
file=sys.stderr)
_provides_cache = (repo, idx)
return idx
def _resolve_provides(name: str, repo: dict, installed: Optional[dict] = None) -> str:
"""Rozwija wirtualną nazwę pakietu do rzeczywistej nazwy.
Kolejność: repo → PROVIDES_MAP → wzorce → provides z repo.json →
provides ZAINSTALOWANYCH pakietów (lokalnie zbudowane poza repo też
dostarczają wirtualne zależności) → fallback pkgconfig (czyszczenie nazwy).
"""
if name in repo:
return name
if name in PROVIDES_MAP:
real = PROVIDES_MAP[name]
if real in repo:
return real
# Wzorce: moduły Qt (Qt5Core/Qt6Widgets) i GStreamer (gstreamer-video-1.0)
if name.startswith("pkgconfig(Qt5"):
real = "qt5"
if real in repo:
return real
if name.startswith("pkgconfig(Qt6"):
real = "qt6"
if real in repo:
return real
if name.startswith("pkgconfig(gstreamer-") and name.endswith("-1.0)"):
real = "gstreamer"
if real in repo:
return real
if name.startswith("pkgconfig(gst-"):
real = "gst-plugins-base"
if real in repo:
return real
# Dynamiczne provides z repo.json (sekcja provides: w PAGBUILD.yaml)
# Determinizm: przy wielu dostawcach wybieramy posortowanego pierwszego
# (i ostrzegamy raz), zamiast zależeć od kolejności w repo.json.
_idx = _provides_index(repo)
if name in _idx:
return min(_idx[name])
# provides ZAINSTALOWANYCH pakietów – lokalnie zbudowane (pagbuild, poza
# repo) też dostarczają wirtualne zależności i muszą być rozpoznawane.
if installed:
_inst_cands = [pn for pn, meta in installed.items()
if isinstance(meta, dict) and name in (meta.get("provides") or [])]
if _inst_cands:
return min(_inst_cands)
clean = name
if name.startswith("pkgconfig(") and ")" in name:
clean = name.split("(", 1)[1].rstrip(")")
elif name.startswith("pkgconfig32(") and ")" in name:
clean = name.split("(", 1)[1].rstrip(")")
if clean != name and clean in repo:
return clean
return name
def cmd_why(pkg_name: str):
"""Pokazuje dlaczego pakiet jest zainstalowany."""
installed = load_json(INSTALLED_DB)
world = load_world()
if pkg_name not in installed:
print(f" {pkg_name}: {_('why_not_installed')}"); return 1
if pkg_name in world:
print(f" {pkg_name}-{installed[pkg_name]['version']}: {_('why_explicit')}")
return 0
parents = set()
for w in world:
_find_dep_path(w, pkg_name, installed, set(), [], parents)
if parents:
for pp in sorted(parents):
print(f" {pkg_name}: {_('why_dependency')} {' → '.join(pp)}")
else:
print(f" {pkg_name}: {_('why_dependency')} (unknown/orphan)")
return 0
def _find_dep_path(cur, target, installed, visited, path, results):
if cur in visited: return
visited.add(cur); path.append(cur)
if cur == target:
results.add(tuple(path))
else:
for dep in installed.get(cur, {}).get("dependencies", []):
_find_dep_path(dep, target, installed, visited, path, results)
path.pop(); visited.discard(cur)
def cmd_autoremove():
"""Automatycznie usuwa osierocone zależności bez pytania."""
installed = load_json(INSTALLED_DB)
world = load_world()
orphans = _find_orphans(installed, world)
if not orphans: print(f"✅ {_('autoremove_none')}"); return 0
print(f"🗑 {_('orphans_found', len(orphans), ', '.join(sorted(orphans)[:10]))}")
return cmd_remove(list(orphans))
def cmd_download(package_names):
"""Pobiera pakiety do cache bez instalowania."""
ensure_dirs()
repo = fetch_all_packages()
if not repo: print(f"❌ {_('no_index')}"); return 1
total_size = 0; downloaded = []
for name in package_names:
pkg = repo.get(name)
if not pkg:
print(f" ❌ {name}: {_('not_found')}"); continue
print(f" ↓ {name}-{pkg.version} ...", end=" ", flush=True)
path = _download_pkg(pkg)
if path:
total_size += os.path.getsize(path)
downloaded.append(name)
print(_c("green", "✓"))
else:
print(_c("red", "✗"))
if downloaded:
print(f"\n✅ {_('downloaded', len(downloaded), total_size/1048576)}")
return 0 if len(downloaded) == len(package_names) else 1
def cmd_stats():
"""Wyświetla statystyki PAG."""
installed = load_json(INSTALLED_DB)
history = load_json(HISTORY_FILE) if os.path.exists(HISTORY_FILE) else []
total_size = sum(i.get("size_bytes", 0) for i in installed.values())
total_files = _db_count_files()
cache_size = sum(
os.path.getsize(os.path.join(PAG_CACHE, f))
for f in os.listdir(PAG_CACHE)
if os.path.isfile(os.path.join(PAG_CACHE, f))
) if os.path.isdir(PAG_CACHE) else 0
last_update = "never"
for e in reversed(history):
if e.get("action") in ("install", "upgrade") and e.get("success"):
last_update = e.get("timestamp", "?")[:19]; break
print(f"\n {_c('bold', _('stats_title'))}")
print(f" {'─' * 40}")
print(f" {_('stats_packages'):<30} {len(installed)}")
print(f" {_('stats_files'):<30} {total_files}")
print(f" {_('stats_size'):<30} {total_size/1048576:.1f} MB")
print(f" {_('stats_cache'):<30} {cache_size/1048576:.1f} MB")
print(f" {_('stats_history'):<30} {len(history)}")
print(f" {_('stats_last_update'):<30} {last_update}")
_bs = _base_summary(repo={}) # bez sieci – tylko stan zarządzania bazą
if _bs["total"]:
_bt = f"{_bs['tracked']}/{_bs['total']}"
if _bs["tracked"] < _bs["total"]:
_bt += " (sudo pag base adopt)"
print(f" {_('stats_base'):<30} {_bt}")
by_size = sorted(installed.items(), key=lambda x: x[1].get("size_bytes", 0), reverse=True)[:5]
if by_size:
print(f"\n {_c('dim', 'Top 5:')}")
for n, i in by_size:
print(f" {n}-{i['version']} {i.get('size_bytes',0)/1048576:.1f} MB")
return 0
def cmd_repo_add(url, name=None):
if not url.startswith("https://") and not os.environ.get("PAG_INSECURE"):
print(f" {_('sec_https')}"); return 1
ensure_dirs()
url = url.rstrip("/")
repos = get_repos()
if url in repos: print(f"⚠ {_('repo_exists', url)}"); return
if name:
# Drop-in: /etc/pag/repos/<nazwa>.conf (jak `echo url > .../stable.conf`)
os.makedirs(REPOS_DIR, exist_ok=True)
target = os.path.join(REPOS_DIR, name.rstrip("/").replace("/", "_") + ".conf")
with open(target, "w") as f: f.write(f"{url}\n")
print(f"✅ {_('repo_added', url)} → {target}")
return
with open(REPOS_CONF, "a") as f: f.write(f"{url}\n")
print(f"✅ {_('repo_added', url)}")
def cmd_repo_list():
for i, url in enumerate(get_repos(), 1): print(f" {i}. {url}")
FLATPAK_REMOTE_URL = "https://flathub.org/repo/flathub.flatpakrepo"
def _check_flatpak(quiet: bool = False):
if not shutil.which("flatpak"):
if not quiet:
print(f"❌ {_('flatpak_missing')}")
return False
r = subprocess.run(["flatpak", "remotes"], capture_output=True, text=True)
if "flathub" not in r.stdout:
print(f"⚠ {_('flatpak_adding')}")
# Jako root dodajemy remote systemowy; bez roota – w instalacji użytkownika,
# żeby remote i miejsce instalacji były spójne (patrz _flatpak_do_install).
add = ["flatpak", "remote-add", "--if-not-exists"]
if os.geteuid() != 0:
add.append("--user")
subprocess.run(add + ["flathub", FLATPAK_REMOTE_URL], check=False)
return True
def _spinner(msg: str):
"""Prosty spinner „myślenia” w osobnym wątku. Zwraca funkcję stop()."""
stop = threading.Event()
def _spin():
for c in itertools.cycle("|/-\\"):
if stop.is_set():
break
sys.stdout.write(f"\r {msg} {c}")
sys.stdout.flush()
time.sleep(0.1)
t = threading.Thread(target=_spin, daemon=True)
t.start()
def _stop():
stop.set()
t.join(timeout=0.3)
sys.stdout.write("\r" + " " * (len(msg) + 4) + "\r")
sys.stdout.flush()
return _stop
def _flatpak_search_raw(query: str) -> List[dict]:
"""Szuka we Flathub i zwraca listę wyników jako słowniki."""
if not _check_flatpak():
return []
stop = _spinner("Szukam we Flathub...")
try:
try:
r = subprocess.run(
["flatpak", "search", "--columns=name,description,application,version,branch,remotes", query],
capture_output=True, text=True, timeout=120
)
finally:
stop()
if r.returncode != 0 and "No matches found" not in r.stdout and not r.stdout.strip():
print(f" ⚠ flatpak search: {r.stderr.strip()[:150]}")
results = []
for line in r.stdout.strip().split("\n"):
parts = line.split("\t")
if len(parts) >= 3:
results.append({
"name": parts[0].strip(),
"description": parts[1].strip() if len(parts) > 1 else "",
"app_id": parts[2].strip() if len(parts) > 2 else "",
"version": parts[3].strip() if len(parts) > 3 else "",
"branch": parts[4].strip() if len(parts) > 4 else "stable",
"origin": parts[5].strip() if len(parts) > 5 else "flathub",
})
return results
except Exception as e:
print(f" ⚠ Błąd wyszukiwania: {e}", file=sys.stderr)
return []
def _flatpak_find_best(query: str) -> Optional[dict]:
"""
Szuka we Flathub i próbuje znaleźć najlepsze dopasowanie.
- Jeśli query dokładnie pasuje do app_id → zwraca od razu
- Jeśli query pasuje do nazwy → zwraca pierwsze
- Jeśli wiele wyników → wyświetla listę i pyta użytkownika
- Jeśli brak → zwraca None
"""
results = _flatpak_search_raw(query)
if not results:
return None
# Dokładne dopasowanie app_id
exact = [r for r in results if r["app_id"].lower() == query.lower()]
if exact:
return exact[0]
# Dokładne dopasowanie nazwy
exact_name = [r for r in results if r["name"].lower() == query.lower()]
if exact_name:
return exact_name[0]
# Jednoznaczne dopasowanie (tylko 1 wynik)
if len(results) == 1:
return results[0]
# Wiele wyników – pokaż użytkownikowi
print(f"\n {_('flatpak_found', len(results))}")
for i, r in enumerate(results):
print(f" {i+1}. {_c('bold', r['name'])} ({r['app_id']})")
if r["version"]:
print(f" {_('flatpak_info_version')}: {r['version']}")
if r["description"]:
desc = r["description"][:80] + ("..." if len(r["description"]) > 80 else "")
print(f" {desc}")
try:
choice = input(f"\n Wybierz numer (1-{len(results)}) lub Enter aby anulować: ").strip()
if not choice:
return None
idx = int(choice) - 1
if 0 <= idx < len(results):
return results[idx]
except (EOFError, ValueError, IndexError):
pass
return None
def _flatpak_get_installed_info(app_id: str) -> Optional[dict]:
"""Zwraca info o zainstalowanym flatpaku lub None."""
try:
r = subprocess.run(
["flatpak", "info", "--columns=name,version,branch,origin,installed-size,description", app_id],
capture_output=True, text=True, timeout=10
)
if r.returncode != 0:
return None
parts = r.stdout.strip().split("\t")
if len(parts) < 3:
return None
return {
"name": parts[0].strip(),
"version": parts[1].strip() if len(parts) > 1 else "",
"branch": parts[2].strip() if len(parts) > 2 else "",
"origin": parts[3].strip() if len(parts) > 3 else "",
"size": parts[4].strip() if len(parts) > 4 else "",
"description": parts[5].strip() if len(parts) > 5 else "",
}
except Exception:
return None
def _flatpak_is_installed(app_id: str) -> bool:
"""Sprawdza czy flatpak o danym ID jest zainstalowany."""
try:
r = subprocess.run(
["flatpak", "info", app_id],
capture_output=True, text=True, timeout=10
)
return r.returncode == 0
except Exception:
return False
# =============================================================================
# FLATPAK – KOMENDY GŁÓWNE (zunifikowany interfejs)
# =============================================================================
# pag flatpak <query> → szuka i proponuje instalację (jeśli nie zainstalowany)
# pag flatpak search <query> → tylko szuka
# pag flatpak install <query> → instaluje
# pag flatpak remove <id> → usuwa
# pag flatpak list → lista zainstalowanych
# pag flatpak update → aktualizuje wszystkie
# pag flatpak info <id> → szczegóły flatpaka
def cmd_flatpak(args: list):
"""
Główna komenda flatpak – inteligentnie rozpoznaje intencję:
pag flatpak firefox → szuka i instaluje (jeśli nieznaleziony → szuka)
pag flatpak search firefox → tylko wyszukiwanie
pag flatpak install ... → bezpośrednia instalacja
pag flatpak remove ... → odinstalowanie
pag flatpak list → lista
pag flatpak update → aktualizacja
pag flatpak info ... → szczegóły
"""
if not _check_flatpak():
return 1
if not args:
# Bez argumentów – domyślnie lista
return cmd_flatpak_list()
subcmd = args[0].lower()
rest = args[1:]
# ── Podkomendy jawne ────────────────────────────────────────────────
if subcmd == "search":
if not rest:
print(_("flatpak_usage")); return 1
return cmd_flatpak_search(" ".join(rest))
elif subcmd == "install":
if not rest:
print(_("flatpak_usage")); return 1
return _flatpak_smart_install(rest)
elif subcmd == "remove" or subcmd == "uninstall":
if not rest:
print(_("flatpak_usage")); return 1
return _flatpak_smart_remove(rest)
elif subcmd == "list":
return cmd_flatpak_list()
elif subcmd == "update":
return cmd_flatpak_update()
elif subcmd == "info":
if not rest:
print(_("flatpak_usage")); return 1
return cmd_flatpak_info(rest[0])
else:
# ── Inteligentne wykrywanie: pag flatpak <nazwa> ────────────────
# Sprawdź czy to zainstalowany flatpak → pokaż info
# Jeśli nie → szukaj i zaproponuj instalację
query = " ".join(args)
# Najpierw sprawdź czy już zainstalowany
if _flatpak_is_installed(query):
print(f" 📦 {_c('green', query)} – already installed (use 'pag flatpak info {query}' for details)")
return cmd_flatpak_info(query)
# Szukaj we Flathub
print(f" {_('flatpak_searching', query)}")
best = _flatpak_find_best(query)
if not best:
print(f" ❌ '{query}' – {_('flatpak_not_found')}")
return 1
print(f"\n {_c('cyan', best['name'])} ({best['app_id']})")
if best["version"]:
print(f" {_('flatpak_info_version')}: {best['version']}")
if best["description"]:
print(f" {best['description']}")
try:
ans = input(f"\n {_('flatpak_install_prompt', best['name'])}").strip().lower()
except (EOFError, KeyboardInterrupt):
print(f"\n ⚠ {_('no_tty')}")
return 0
if ans and ans not in ("t", "y"):
print(_("cancelled"))
return 0
return _flatpak_do_install(best["app_id"])
def _flatpak_smart_install(names: list) -> int:
"""Instaluje flatpaki – obsługuje nazwy częściowe (wyszukuje przed instalacją)."""
failed = 0
for name in names:
if "." in name and "/" not in name:
# Wygląda na pełne app_id (np. org.mozilla.firefox)
app_id = name
else:
# Szukaj najlepszego dopasowania
best = _flatpak_find_best(name)
if not best:
print(f" ❌ '{name}' – {_('flatpak_not_found')}")
failed += 1
continue
app_id = best["app_id"]
print(f" → {best['name']} ({app_id})")
if _flatpak_do_install(app_id) != 0:
failed += 1
return 1 if failed else 0
def _flatpak_export_dirs() -> List[str]:
"""Katalogi eksportów Flatpaka (system + użytkownika) obecne na dysku.
Instalacja użytkownika roota (/root/.local/share/flatpak) jest świadomie
POMIJANA – to wewnętrzny artefakt roota, którego sesja użytkownika nigdy
nie zobaczy (patrz _flatpak_do_install)."""
dirs = ["/var/lib/flatpak/exports"]
home = os.path.expanduser("~")
if home and home not in ("/root", "/"):
dirs.append(os.path.join(home, ".local", "share", "flatpak", "exports"))
return [d for d in dirs if os.path.isdir(d)]
def _flatpak_refresh_caches() -> None:
"""Odświeża cache pulpitu i ikon po zmianie w Flatpaku (best-effort).
Flatpak robi to sam, ale tylko gdy `update-desktop-database` jest w PATH.
Bez tego nowo zainstalowana aplikacja bywa nieobecna w menu."""
for exports in _flatpak_export_dirs():
apps = os.path.join(exports, "share", "applications")
if os.path.isdir(apps) and shutil.which("update-desktop-database"):
subprocess.run(["update-desktop-database", apps],
capture_output=True, check=False, timeout=60)
icons = os.path.join(exports, "share", "icons", "hicolor")
if os.path.isdir(icons) and shutil.which("gtk-update-icon-cache"):
subprocess.run(["gtk-update-icon-cache", "-q", "-t", "-f", icons],
capture_output=True, check=False, timeout=120)
def _flatpak_session_sees_exports() -> bool:
"""Czy bieżąca sesja ma eksporty Flatpaka w XDG_DATA_DIRS."""
raw = os.environ.get("XDG_DATA_DIRS", "")
dirs = {d for d in raw.split(":") if d} or {"/usr/local/share", "/usr/share"}
return any(os.path.join(e, "share") in dirs for e in _flatpak_export_dirs())
def _flatpak_warn_if_invisible() -> None:
"""Ostrzega, gdy sesja nie widzi eksportów – inaczej wygląda to jak
„zainstalowało się, ale nie ma go w menu / na PC”."""
if not _flatpak_session_sees_exports():
print(f" ⚠ {_('flatpak_menu_hint')}")
print(f" {_('flatpak_menu_fix')}")
def _flatpak_do_install(app_id: str) -> int:
"""Wykonuje właściwą instalację flatpaka.
Jako root wymuszamy instalację SYSTEMOWĄ (--system). Bez tego `flatpak`
uruchomiony jako root potrafi zainstalować aplikację w instalacji
użytkownika roota (/root/.local/share/flatpak) – widocznej dla roota, ale
nie dla zalogowanego użytkownika. To dokładnie efekt „instaluje, ale nie
mam tego na PC”. Bez roota próbujemy instalacji systemowej (polkit),
a w razie niepowodzenia cofamy się do instalacji użytkownika (--user).
Zakres (scope) ma znaczenie: instalacja SYSTEMOWA (--system) jest widoczna
dla wszystkich użytkowników, a --user tylko dla bieżącego (dla roota:
/root/.local/... → niewidoczna dla sesji). `flatpak list` bez flag pokazuje
oba zakresy, więc systemowa instalacja roota jest widoczna też dla
zwykłego użytkownika."""
print(f" {_('flatpak_installing', app_id)}")
if os.geteuid() == 0:
result = subprocess.run(
["flatpak", "install", "-y", "--system", "flathub", app_id],
check=False, timeout=600)
else:
result = subprocess.run(
["flatpak", "install", "-y", "flathub", app_id],
check=False, timeout=600)
if result.returncode != 0:
# Flathub tylko dla użytkownika / brak agenta polkit – instalacja
# lokalna użytkownika jest lepsza niż twardy błąd.
subprocess.run(
["flatpak", "remote-add", "--if-not-exists", "--user",
"flathub", FLATPAK_REMOTE_URL], check=False, timeout=60)
result = subprocess.run(
["flatpak", "install", "-y", "--user", "flathub", app_id],
check=False, timeout=600)
if result.returncode != 0:
print(f" ❌ {_('download_fail')}: {app_id}")
return 1
_flatpak_refresh_caches()
print(f" ✅ {_('flatpak_installed', app_id)}")
_flatpak_warn_if_invisible()
return 0
def _flatpak_smart_remove(names: list) -> int:
"""Usuwa flatpaki – obsługuje nazwy częściowe."""
# Pobierz listę zainstalowanych
try:
r = subprocess.run(
["flatpak", "list", "--columns=application,name"],
capture_output=True, text=True, timeout=10
)
installed = {}
for line in r.stdout.strip().split("\n"):
parts = line.split("\t")
if len(parts) >= 2:
installed[parts[0].strip()] = parts[1].strip()
except Exception:
installed = {}
failed = 0
for name in names:
app_id = name
# Jeśli nie podano pełnego ID – spróbuj dopasować
if name not in installed:
matches = {aid: aname for aid, aname in installed.items()
if name.lower() in aid.lower() or name.lower() in aname.lower()}
if len(matches) == 0:
print(f" ❌ '{name}' – {_('flatpak_not_installed', name)}")
failed += 1
continue
elif len(matches) == 1:
app_id = list(matches.keys())[0]
print(f" → {matches[app_id]} ({app_id})")
else:
print(f"\n Wiele dopasowań dla '{name}':")
for i, (aid, aname) in enumerate(sorted(matches.items()), 1):
print(f" {i}. {aname} ({aid})")
try:
choice = input(f"\n Wybierz numer (1-{len(matches)}) lub Enter: ").strip()
if not choice:
failed += 1
continue
aid_list = sorted(matches.keys())
app_id = aid_list[int(choice) - 1]
except (EOFError, ValueError, IndexError):
failed += 1
continue
print(f" 🗑 {app_id} ...", end=" ", flush=True)
result = subprocess.run(
["flatpak", "uninstall", "-y", app_id],
capture_output=True, text=True, timeout=120
)
if result.returncode == 0:
print("✅")
print(f" {_('flatpak_removed', app_id)}")
else:
print("❌")
failed += 1
if not failed:
_flatpak_refresh_caches()
return 1 if failed else 0
def cmd_flatpak_search(q: str):
"""Wyszukuje we Flathub i wyświetla wyniki (z możliwością wyboru do instalacji)."""
if not _check_flatpak():
return 1
results = _flatpak_search_raw(q)
if not results:
print(f" ❌ '{q}' – {_('flatpak_not_found')}")
return 1
print(f"\n {_('flatpak_found', len(results))}")
shown = results[:30] # max 30 wyników
for i, r in enumerate(shown, 1):
installed = "📦 " if _flatpak_is_installed(r["app_id"]) else " "
print(f" {i:>2}. {installed}{_c('bold', r['name'])} ({r['app_id']})")
if r["version"]:
print(f" {_('flatpak_info_version')}: {r['version']} | {_('flatpak_info_branch')}: {r['branch']}")
if r["description"]:
desc = r["description"][:100] + ("..." if len(r["description"]) > 100 else "")
print(f" {_c('dim', desc)}")
if len(results) > 30:
print(f" ... i {len(results) - 30} więcej. Doprecyzuj zapytanie.")
# Interaktywny wybór – wpisz numer, aby zainstalować (Enter = anuluj)
try:
ans = input(f"\n Wybierz numer do zainstalowania (1-{len(shown)}) lub Enter aby anulować: ").strip()
except (EOFError, KeyboardInterrupt):
return 0
if ans:
try:
idx = int(ans) - 1
if 0 <= idx < len(shown):
return _flatpak_do_install(shown[idx]["app_id"])
print(_("cancelled"))
except (ValueError, IndexError):
print(_("cancelled"))
return 0
def cmd_flatpak_list():
"""Wyświetla zainstalowane flatpaki."""
if not _check_flatpak():
return 1
r = subprocess.run(
["flatpak", "list", "--columns=application,name,version,origin,installation,installed-size"],
capture_output=True, text=True, timeout=10
)
lines = [l for l in r.stdout.strip().split("\n") if l.strip()]
if not lines:
print(" (brak zainstalowanych flatpaków)")
return 0
print(f" Zainstalowane flatpaki ({len(lines)}):")
for line in lines:
parts = line.split("\t")
if len(parts) >= 3:
app_id, name, version = parts[0], parts[1], parts[2]
scope = parts[4] if len(parts) > 4 else ""
size = parts[5] if len(parts) > 5 else ""
size_str = f" ({size})" if size else ""
scope_str = f" [{scope}]" if scope else ""
print(f" 📦 {_c('bold', name)} {version}{scope_str}{size_str}")
print(f" {_c('dim', app_id)}")
return 0
def cmd_flatpak_update():
"""Aktualizuje wszystkie flatpaki."""
if not _check_flatpak():
return 1
print(" 🔄 Aktualizacja flatpaków...")
result = subprocess.run(["flatpak", "update", "-y"], check=False, timeout=600)
if result.returncode == 0:
_flatpak_refresh_caches()
print(f" ✅ {_('flatpak_updated')}")
return result.returncode
def cmd_flatpak_info(app_id: str):
"""Wyświetla szczegóły flatpaka (zainstalowanego lub z Flathub)."""
if not _check_flatpak():
return 1
# Najpierw sprawdź zainstalowany
info = _flatpak_get_installed_info(app_id)
if info:
print(f"\n 📦 {_c('bold', info['name'])} {_c('green', '[zainstalowany]')}")
print(f" {'─' * 45}")
print(f" {_('flatpak_info_id'):<16} {app_id}")
print(f" {_('flatpak_info_version'):<16} {info['version']}")
print(f" {_('flatpak_info_branch'):<16} {info['branch']}")
print(f" {_('flatpak_info_origin'):<16} {info['origin']}")
if info["size"]:
print(f" {_('flatpak_info_size'):<16} {info['size']}")
if info["description"]:
print(f" {_('flatpak_info_desc'):<16} {info['description']}")
return 0
# Szukaj we Flathub
results = _flatpak_search_raw(app_id)
exact = [r for r in results if r["app_id"].lower() == app_id.lower()]
if not exact:
# Spróbuj częściowego dopasowania
if results:
exact = [results[0]]
else:
print(f" ❌ '{app_id}' – {_('flatpak_not_found')}")
return 1
r = exact[0]
print(f"\n 📦 {_c('bold', r['name'])} (Flathub)")
print(f" {'─' * 45}")
print(f" {_('flatpak_info_id'):<16} {r['app_id']}")
print(f" {_('flatpak_info_version'):<16} {r['version']}")
if r["description"]:
print(f" {_('flatpak_info_desc'):<16} {r['description']}")
print(f"\n 💡 Aby zainstalować: pag flatpak install {r['app_id']}")
return 0
# =============================================================================
# IMMUTABLE OS – KOMENDY DEPLOYMENTOWE
# =============================================================================
# Pakiety jądra – po ich instalacji trzeba przebudować initramfs
KERNEL_PACKAGE_PATTERNS = ["linux", "kernel", "linux-kernel", "linux-lts"]
def _is_kernel_package(name: str) -> bool:
"""Sprawdza czy pakiet to jądro (wymaga przebudowy initramfs)."""
name_lower = name.lower()
return any(pattern in name_lower for pattern in KERNEL_PACKAGE_PATTERNS)
def _rebuild_initramfs(deploy_dir: str = "") -> bool:
"""
Przebudowuje initramfs dla aktywnego (lub podanego) deploymentu.
Używa skryptu pag-initramfs lub ręcznego cpio.
"""
if deploy_dir:
root = deploy_dir
else:
root = _get_deployment_root()
if root == PAG_ROOT:
# Zwykły system – użyj dracut jeśli dostępny
if shutil.which("dracut"):
print(" 🔧 Przebudowa initramfs (dracut)...")
result = subprocess.run(
["dracut", "--force", "/boot/initramfs.img"],
capture_output=True, text=True, timeout=120
)
return result.returncode == 0
elif shutil.which("mkinitcpio"):
print(" 🔧 Przebudowa initramfs (mkinitcpio)...")
result = subprocess.run(
["mkinitcpio", "-g", "/boot/initramfs.img"],
capture_output=True, text=True, timeout=120
)
return result.returncode == 0
else:
print(" ⚠ Brak dracut/mkinitcpio – initramfs nie został przebudowany")
return False
# Tryb immutable – budujemy initramfs dla deploymentu
print(" 🔧 Budowanie initramfs dla deploymentu...")
# Sprawdź czy mamy nasz skrypt init
pag_init_script = "/usr/share/pag/initramfs-init"
if not os.path.exists(pag_init_script):
# Szukaj w źródłach (developerski fallback)
alt_paths = [
os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts", "initramfs-init"),
"/usr/share/pag/init",
]
for p in alt_paths:
if os.path.exists(p):
pag_init_script = p
break
if not os.path.exists(pag_init_script):
print(" ⚠ Nie znaleziono pag-initramfs-init – pomijam budowę initramfs")
return False
boot_dir = os.path.join(root, "boot")
os.makedirs(boot_dir, exist_ok=True)
# Znajdź jądro (vmlinuz-*)
kernels = sorted(
[f for f in os.listdir(boot_dir) if f.startswith("vmlinuz-")],
reverse=True
) if os.path.exists(boot_dir) else []
if not kernels:
print(" ⚠ Nie znaleziono vmlinuz-* w /boot deploymentu")
return False
kernel_ver = kernels[0].replace("vmlinuz-", "")
print(f" 🐧 Jądro: {kernel_ver}")
# Buduj initramfs ręcznie (cpio)
tmpdir = tempfile.mkdtemp(prefix="pag-initramfs-")
try:
# Podstawowa struktura
for d in ["bin", "sbin", "dev", "proc", "sys", "run", "new_root",
"usr/bin", "usr/sbin", "lib", "lib64", "etc"]:
os.makedirs(os.path.join(tmpdir, d), exist_ok=True)
# Skopiuj init
shutil.copy2(pag_init_script, os.path.join(tmpdir, "init"))
os.chmod(os.path.join(tmpdir, "init"), 0o755)
# Skopiuj niezbędne binaria (busybox lub podstawowe narzędzia)
busybox_paths = [
os.path.join(root, "usr/bin/busybox"),
os.path.join(root, "bin/busybox"),
"/usr/bin/busybox",
"/bin/busybox",
]
busybox = None
for bp in busybox_paths:
if os.path.exists(bp):
busybox = bp
break
if busybox:
shutil.copy2(busybox, os.path.join(tmpdir, "bin/busybox"))
# Utwórz symlinki dla podstawowych komend
for cmd in ["sh", "mount", "umount", "ls", "cat", "echo", "sleep",
"readlink", "mkdir", "switch_root", "cp", "rm"]:
link = os.path.join(tmpdir, "bin", cmd)
if not os.path.exists(link):
os.symlink("busybox", link)
# /bin/sh → busybox
if not os.path.exists(os.path.join(tmpdir, "bin/sh")):
os.symlink("busybox", os.path.join(tmpdir, "bin/sh"))
else:
# Bez busybox – kopiuj podstawowe narzędzia z deploymentu
for tool in ["bash", "mount", "umount", "readlink", "mkdir", "cat", "sleep", "cp", "rm"]:
src = os.path.join(root, "usr/bin", tool)
if not os.path.exists(src):
src = os.path.join(root, "bin", tool)
if os.path.exists(src):
dest = os.path.join(tmpdir, "bin", os.path.basename(tool))
shutil.copy2(src, dest)
# Kopiuj zależności .so
_copy_libs_for_binary(src, tmpdir, root)
# Dodaj moduły jądra (opcjonalnie – dla sterowników dyskowych)
modules_src = os.path.join(root, "lib/modules", kernel_ver)
if os.path.isdir(modules_src):
modules_dst = os.path.join(tmpdir, "lib/modules", kernel_ver)
# Kopiuj tylko niezbędne (fs, block, drivers/ata, drivers/nvme)
for sub in ["kernel/fs", "kernel/drivers/ata", "kernel/drivers/nvme",
"kernel/drivers/scsi", "kernel/drivers/virtio",
"modules.order", "modules.builtin"]:
src_sub = os.path.join(modules_src, sub)
if os.path.exists(src_sub):
dst_sub = os.path.join(modules_dst, sub)
os.makedirs(os.path.dirname(dst_sub), exist_ok=True)
if os.path.isdir(src_sub):
try:
shutil.copytree(src_sub, dst_sub, dirs_exist_ok=True, symlinks=True,
ignore_dangling_symlinks=True)
except (FileNotFoundError, PermissionError):
print(f" ⚠ Pomijam niedostępne pliki: {sub}")
else:
try:
shutil.copy2(src_sub, dst_sub)
except (FileNotFoundError, PermissionError):
print(f" ⚠ Pomijam niedostępny plik: {sub}")
# Pakuj do initramfs.img
initramfs_path = os.path.join(boot_dir, "initramfs.img")
old_cwd = os.getcwd()
os.chdir(tmpdir)
try:
with open(initramfs_path + ".tmp", "wb") as out:
_run_cpio_pipeline(tmpdir, out)
os.rename(initramfs_path + ".tmp", initramfs_path)
finally:
os.chdir(old_cwd)
size_mb = os.path.getsize(initramfs_path) / 1048576
print(f" ✅ initramfs.img ({size_mb:.1f} MB) → {initramfs_path}")
return True
except Exception as e:
print(f" ❌ Błąd budowy initramfs: {e}")
return False
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
def _run_cpio_pipeline(tmpdir: str, out):
"""find . -print0 | cpio --null -oH newc | gzip — bez shell=True.
Buduje pipeline przez subprocess.Popen, unikając pośrednika powłoki
(brak ryzyka injection i niepotrzebnego procesu sh). Wykonuje się w cwd=tmpdir.
Separatory NUL (\0): plik/katalog ze znakiem nowej linii w nazwie nie
rozjeżdża cpio (inaczej uszkodzone archiwum → kernel panic przy rozruchu).
"""
find = subprocess.Popen(["find", ".", "-print0"], cwd=tmpdir, stdout=subprocess.PIPE)
cpio = subprocess.Popen(["cpio", "--null", "-oH", "newc"], cwd=tmpdir,
stdin=find.stdout, stdout=subprocess.PIPE)
find.stdout.close() # zwolnij uchwyt – cpio dostanie SIGPIPE po zakończeniu find
gzip = subprocess.Popen(["gzip"], stdin=cpio.stdout, stdout=out)
cpio.stdout.close()
try:
gzip.wait(timeout=120)
if gzip.returncode != 0:
raise subprocess.CalledProcessError(gzip.returncode, ["gzip"])
cpio.wait(timeout=30)
find.wait(timeout=30)
except subprocess.TimeoutExpired:
for p in (gzip, cpio, find):
p.kill()
raise
finally:
for p in (find, cpio, gzip):
if p.poll() is None:
p.kill()
# Skontroluj też kody procesów pośrednich (cpio/find mogą zawieść, a gzip zwrócić 0)
if cpio.returncode != 0:
raise subprocess.CalledProcessError(cpio.returncode, ["cpio"])
if find.returncode != 0:
raise subprocess.CalledProcessError(find.returncode, ["find"])
def _copy_libs_for_binary(binary: str, dest_dir: str, root: str):
"""Kopiuje zależności .so dla binarki do initramfs (uproszczone ldd)."""
try:
result = subprocess.run(
["ldd", binary], capture_output=True, text=True, timeout=10
)
for line in result.stdout.split("\n"):
m = re.search(r'=>\s+(/\S+)', line)
if m:
lib_path = m.group(1)
lib_rel = lib_path.lstrip("/")
lib_dest = os.path.join(dest_dir, lib_rel)
if not os.path.exists(lib_dest):
os.makedirs(os.path.dirname(lib_dest), exist_ok=True)
# Szukaj w deployment root lub systemie
if os.path.exists(lib_path):
shutil.copy2(lib_path, lib_dest)
else:
alt = os.path.join(root, lib_rel)
if os.path.exists(alt):
shutil.copy2(alt, lib_dest)
except Exception:
pass
def cmd_initramfs_update():
"""Ręcznie przebudowuje initramfs dla bieżącego deploymentu."""
ensure_dirs()
deploy_dir = _get_deployment_root()
if deploy_dir != PAG_ROOT:
print(f"🏗️ Deployment: {os.path.basename(deploy_dir)}")
ok = _rebuild_initramfs(deploy_dir)
if ok:
print("✅ Initramfs zaktualizowany.")
# Po initramfs – zaktualizuj też GRUB
_update_grub_config()
else:
print("❌ Błąd aktualizacji initramfs.")
return 0 if ok else 1
def _update_grub_config():
"""
Generuje wpisy GRUB dla wszystkich deploymentów.
Każdy deployment dostaje własny wpis – rollback możliwy z bootloadera.
"""
grub_cfg = "/boot/grub/grub.cfg"
if not os.path.exists(os.path.dirname(grub_cfg)):
return # brak GRUB
deployments = _load_deployments()
root_dev = _detect_root_device()
if not root_dev:
print(" ⚠ Nie udało się wykryć partycji root – wpisy GRUB nie dostaną root=.",
file=sys.stderr)
print(" Ustaw PAG_ROOT_DEVICE=/dev/... (lub PAG_GRUB_ROOT) i powtórz.",
file=sys.stderr)
root_arg = f" root={root_dev}" if root_dev else ""
lines = [
"# =====================================================================",
"# Pagan Linux – GRUB config (wygenerowane przez pag grub-update)",
f"# Data: {datetime.now().isoformat()}",
"# =====================================================================",
"",
]
# Domyślny – ostatni (najnowszy) deployment
if deployments:
latest = deployments[-1]["id"]
lines.append(f"set default=0")
lines.append(f"set timeout=5")
else:
lines.append("set default=0")
lines.append("set timeout=5")
lines.append("")
# Wpisy dla każdego deploymentu (od najnowszego)
entry_num = 0
for d in reversed(deployments):
deploy_dir = os.path.join(DEPLOYMENTS_DIR, d["id"])
boot_dir = os.path.join(deploy_dir, "boot")
kernels = sorted(
[f for f in os.listdir(boot_dir) if f.startswith("vmlinuz-")],
reverse=True
) if os.path.isdir(boot_dir) else []
kernel_path = f"/.deployments/{d['id']}/boot/{kernels[0]}" if kernels else ""
initrd_path = f"/.deployments/{d['id']}/boot/initramfs.img"
initrd_line = f"initrd {initrd_path}" if os.path.exists(os.path.join(boot_dir, "initramfs.img")) else ""
active_mark = " [AKTYWNY]" if d.get("active") else ""
pkg_list = ", ".join(d.get("packages", [])[:3])
label = f"Pagan Linux – {d['id']}{active_mark}"
lines.append(f"menuentry '{label}' {{")
if kernel_path:
lines.append(f" linux {kernel_path}{root_arg} rw quiet")
else:
lines.append(f" # Brak jądra w tym deploymencie")
if initrd_line:
lines.append(f" {initrd_line}")
lines.append("}")
lines.append("")
entry_num += 1
# Wpis fallback: zwykły root (gdyby wszystko padło).
# Uwaga: GRUB NIE rozwija globów – trzeba podać KONKRETNY plik jądra
# (wcześniejsze `linux /boot/vmlinuz-*` było niepoprawne).
fallback_kernels = sorted(
[f for f in os.listdir("/boot") if f.startswith("vmlinuz-")],
reverse=True
) if os.path.isdir("/boot") else []
lines.append("menuentry 'Pagan Linux – fallback (zwykły root)' {")
if fallback_kernels:
lines.append(f" linux /boot/{fallback_kernels[0]}{root_arg} rw quiet")
else:
lines.append(" # Brak jądra w /boot")
lines.append(f" initrd /boot/initramfs.img")
lines.append("}")
lines.append("")
# Zapisz
os.makedirs(os.path.dirname(grub_cfg), exist_ok=True)
with open(grub_cfg, "w") as f:
f.write("\n".join(lines))
print(" 📋 GRUB config zaktualizowany – wpisy dla każdego deploymentu")
def _root_device_from_fstab(path: str = "/etc/fstab") -> str:
"""Zwraca DEVICE z wpisu dla '/' w fstab (pomija komentarze)."""
try:
with open(path, "r", encoding="utf-8", errors="replace") as fh:
for line in fh:
line = line.split("#", 1)[0].strip()
if not line:
continue
parts = line.split()
if len(parts) >= 2 and parts[1] == "/":
return parts[0]
except OSError:
pass
return ""
def _in_chroot() -> bool:
"""Heurystyka: czy działamy w chrocie (build ISO/IMG)?
W chrocie /proc/1/root to root HOSTA – inny system plików niż nasz '/'.
Na zwykłym systemie PID 1 ma root równy '/'. PAGAN_ROOT ustawiają skrypty
budujące i jest dziedziczony przez chroot (dodatkowa wskazówka).
"""
if os.environ.get("PAG_IN_CHROOT") == "1":
return True
if os.environ.get("PAGAN_ROOT"):
return True
try:
return os.stat("/").st_dev != os.stat("/proc/1/root").st_dev
except OSError:
return False
def _detect_root_device() -> str:
"""Wykrywa urządzenie/identyfikator partycji root dla GRUB (root=...).
Kolejność (od najbardziej wiarygodnego w danym kontekście):
1) PAG_ROOT_DEVICE / PAG_GRUB_ROOT – jawny override (build ISO/IMG,
instalator); najpewniejszy, bo nie zgadujemy.
2) /etc/fstab TARGETU – w chrocie findmnt '/' zwraca partycję HOSTA,
a fstab opisuje target (np. root=/dev/sda2 dla obrazu IMG).
3) findmnt '/' – na żywym systemie (obsługuje UUID/LUKS/subvol).
Zwraca "" gdy nie da się ustalić – wtedy GRUB nie dostaje błędnego root=.
"""
explicit = (os.environ.get("PAG_ROOT_DEVICE")
or os.environ.get("PAG_GRUB_ROOT") or "").strip()
if explicit:
return explicit
fstab_dev = _root_device_from_fstab()
if _in_chroot() and fstab_dev:
return fstab_dev
try:
result = subprocess.run(
["findmnt", "-n", "-o", "SOURCE", "/"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0 and result.stdout.strip():
return result.stdout.strip()
except Exception:
pass
# Ostatnia deska: fstab targetu (lepsze niż hardkodowane /dev/sda1,
# które na sprzęcie z NVMe dawało niebootowalny system).
return fstab_dev
def cmd_grub_update():
"""Ręcznie regeneruje konfigurację GRUB (wpisy dla deploymentów)."""
ensure_dirs()
print("📋 Aktualizacja konfiguracji GRUB...")
_update_grub_config()
print("✅ GRUB zaktualizowany.")
return 0
def cmd_deploy_list():
"""Wyświetla listę wszystkich deploymentów."""
deployments = _load_deployments()
if not deployments:
print(_("no_deployments")); return
print(_("deployments_list", len(deployments)))
active = os.readlink(ACTIVE_LINK) if os.path.islink(ACTIVE_LINK) else ""
for d in reversed(deployments):
marker = f" ◀ {_('active_deployment')}" if d.get("active") or d["id"] == os.path.basename(active) else ""
print(f" {d['id']}{marker}")
print(f" {d['action']}: {', '.join(d['packages'][:5])}")
if len(d.get('packages', [])) > 5:
print(f" +{len(d['packages']) - 5} więcej...")
print(f" {d['timestamp']}")
def cmd_deploy_rollback():
"""Przełącza na poprzedni deployment."""
deployments = _load_deployments()
active_indices = [i for i, d in enumerate(deployments) if d.get("active")]
if len(deployments) < 2:
print(f"❌ {_('deploy_rollback_fail')}"); return 1
current_idx = active_indices[0] if active_indices else len(deployments) - 1
prev_idx = current_idx - 1 if current_idx > 0 else -1
if prev_idx < 0:
print(f"❌ {_('deploy_rollback_fail')}"); return 1
prev = deployments[prev_idx]
prev_dir = os.path.join(DEPLOYMENTS_DIR, prev["id"])
if not os.path.isdir(prev_dir):
print(f"❌ Deployment {prev['id']} nie istnieje na dysku"); return 1
print(f"⏪ Przywracanie deploymentu: {prev['id']}")
print(f" {prev['action']}: {', '.join(prev['packages'][:5])}")
if not _ask_confirm():
return 0
_switch_deployment(prev_dir)
for d in deployments:
d["active"] = (d["id"] == prev["id"])
_save_deployments(deployments)
_update_grub_config()
print(f"✅ {_('deploy_rollback_ok', prev['id'])}")
print(" 💡 Restart wymagany do przeładowania systemu.")
return 0
def cmd_deploy_cleanup(keep: int = 3):
"""Usuwa stare deploymenty, zachowując ostatnie `keep`."""
deployments = _load_deployments()
if len(deployments) <= keep:
print(f"✅ {_('deploy_cleanup_none', keep)}"); return 0
to_remove = deployments[:-keep]
removed = 0
for d in to_remove:
deploy_dir = os.path.join(DEPLOYMENTS_DIR, d["id"])
if os.path.isdir(deploy_dir):
shutil.rmtree(deploy_dir, ignore_errors=True)
removed += 1
remaining = deployments[-keep:]
_save_deployments(remaining)
print(f"✅ {_('deploy_cleanup_ok', removed)}")
return 0
# =============================================================================
# POMOCNICZE
# =============================================================================
# Pakiety dostarczane przez bazowy system (zawsze "zainstalowane"). Wchodzą do
# rootfs z ISO przez data.tar.xz, więc NIE ma ich w repo.json ani w bazie pag –
# każda zależność od nich (glibc, bash, python3…) wyglądałaby na "brakującą".
# Współdzielone przez rozwijanie i weryfikację zależności.
SYSTEM_BASE = {
"glibc", "libc", "gcc", "g++", "make", "binutils", "coreutils", "bash",
"linux-api-headers", "kernel-headers", "zlib", "pkg-config", "pkgconf",
"tar", "gzip", "xz", "bzip2", "findutils", "grep", "sed", "gawk", "awk",
"diffutils", "patch", "file", "m4", "perl", "python3", "sh",
}
def _resolve_deps(names, repo, installed):
resolved, visited = [], set()
missing = [] # zależności których nie ma ani w repo ani zainstalowane
def _is_base(dep):
"""True, gdy zależność dostarcza bazowy system (także przez provides)."""
return dep in SYSTEM_BASE \
or _resolve_provides(dep, repo, installed) in SYSTEM_BASE
def visit(name, explicit=False):
if name in visited: return
# Rozwijanie wirtualnych zależności przez provides
target = _resolve_provides(name, repo, installed)
# Zależność dostarczana przez bazowy system – nie ma jej w repo ani w
# bazie pag, więc nie "brakuje" jej i nie próbujemy jej instalować.
# Wyjątek: nazwa podana wprost przez użytkownika (explicit=True).
if not explicit and target in SYSTEM_BASE:
return
if target in visited: return
visited.add(target)
if target in repo:
for dep in repo[target].dependencies:
real_dep = _resolve_provides(dep, repo, installed)
real_target = real_dep if real_dep in repo else dep
if _is_base(dep):
continue
# Sprawdź czy zależność jest dostępna
if real_target not in installed and real_target not in repo:
if dep not in missing:
missing.append(dep)
if dep not in installed:
visit(real_target)
elif target not in installed:
# Pakiet nie istnieje ani w repo ani zainstalowany
if target not in missing:
missing.append(target)
if target not in installed and target not in resolved:
resolved.append(target)
for name in names:
visit(name, explicit=True)
# Zwróć brakujące (do sprawdzenia przez wywołującego)
return resolved, missing
def _verify_dependencies(to_install: list, repo: dict, installed: dict) -> int:
"""
Sprawdza czy wszystkie zależności pakietów do instalacji są spełnione.
Zwraca liczbę brakujących zależności.
"""
all_missing = []
all_warnings = []
for pkg_name in to_install:
pkg = repo.get(pkg_name)
if not pkg:
continue
for dep in pkg.dependencies:
if dep in SYSTEM_BASE:
continue # bazowy system dostarcza tę zależność
real_dep = _resolve_provides(dep, repo, installed)
# Sprawdź czy zależność jest dostępna (w repo lub już zainstalowana)
in_repo = real_dep in repo
in_installed = real_dep in installed
will_be_installed = real_dep in to_install
if not in_repo and not in_installed and not will_be_installed:
if dep not in all_missing:
all_missing.append((pkg_name, dep))
elif in_repo and not in_installed and not will_be_installed:
if dep not in [w[1] for w in all_warnings]:
all_warnings.append((pkg_name, dep, real_dep))
if all_missing:
print(f"\n❌ {_c('red', 'BRAKUJĄCE ZALEŻNOŚCI')} – nie można zainstalować:")
for pkg, dep in all_missing:
print(f" {pkg} → potrzebuje {_c('red', dep)} (brak w repozytoriach)")
print()
if all_warnings:
print(f"\n⚠ {_c('yellow', 'NIESPEŁNIONE ZALEŻNOŚCI')} – zostaną doinstalowane:")
for pkg, dep, real in all_warnings:
print(f" {pkg} → {dep} ({_c('green', real)} – będzie pobrane)")
print()
return len(all_missing)
# Biblioteki bazowe (glibc/gcc runtime) – zawsze dostępne, nie wymagają pakietu
BASE_SO = {
"libc.so.6", "libm.so.6", "libpthread.so.0", "libdl.so.2", "librt.so.1",
"libutil.so.1", "libresolv.so.2", "libnsl.so.1", "libcrypt.so.1",
"ld-linux.so.2", "ld-linux-x86-64.so.2", "ld-linux-aarch64.so.1",
"libgcc_s.so.1", "linux-vdso.so.1",
}
def _verify_so_deps(to_install: list, repo: dict, installed: dict) -> int:
"""Sprawdza wymagania ABI (provides_so / requires_so z metadata.json).
Fail-closed TYLKO gdy metadata jawnie deklaruje requires_so, a żaden pakiet
(bazowy, zainstalowany lub instalowany w tej transakcji) nie dostarcza
wymaganej wersji biblioteki. Stare pakiety bez tych pól są pomijane.
"""
provided = set(BASE_SO)
for n in to_install:
p = repo.get(n)
if p:
provided.update(p.provides_so or [])
for n, info in installed.items():
provided.update(info.get("provides_so", []) or [])
missing = []
for n in sorted(to_install):
p = repo.get(n)
if not p:
continue
for so in (p.requires_so or []):
if so not in provided:
missing.append((n, so))
if missing:
print(f"\n❌ {_c('red', 'BRAK WYMAGANYCH BIBLIOTEK (ABI so-name)')}:")
for n, so in missing:
print(f" {n} → wymaga {_c('red', so)} – żaden pakiet nie dostarcza tej wersji")
print()
return len(missing)
def _download_pkg(pkg, attempts: int = 3):
url = f"{pkg.repo_url}/{pkg.filename}"
dest = os.path.join(PAG_CACHE, pkg.filename)
if os.path.exists(dest) and (not pkg.sha256 or _sha256_file(dest) == pkg.sha256):
_download_pkg_sig(pkg, dest) # upewnij się, że sygnatura jest w cache
return dest
last_err = None
for attempt in range(1, attempts + 1):
try:
req = Request(url, headers={"User-Agent":"pag/3.0"})
with urlopen(req, timeout=600) as resp:
total = int(resp.headers.get("Content-Length", 0))
bar = DownloadBar(pkg.filename, total)
with open(dest, "wb") as f:
while True:
chunk = resp.read(65536)
if not chunk:
break
f.write(chunk)
bar.update(len(chunk))
bar.close()
if pkg.sha256 and _sha256_file(dest) != pkg.sha256:
last_err = "SHA256 mismatch"
try:
os.remove(dest)
except OSError:
pass
if attempt < attempts:
print(f" ↻ Ponawiam pobieranie {pkg.filename} "
f"({attempt}/{attempts - 1})...", file=sys.stderr)
time.sleep(attempt)
continue
return None
_download_pkg_sig(pkg, dest)
return dest
except Exception as e:
last_err = e
# Usuń częściowy plik – bez tego mógłby zostać użyty jako „cache”
# (gdy pakiet nie ma sha256) albo mylić kolejną próbę.
try:
if os.path.exists(dest):
os.remove(dest)
except OSError:
pass
if attempt < attempts:
print(f" ↻ Ponawiam pobieranie {pkg.filename} "
f"({attempt}/{attempts - 1})...", file=sys.stderr)
time.sleep(attempt)
continue
print(f" ⚠ Błąd pobierania {pkg.filename}: {last_err}", file=sys.stderr)
return None
def _download_pkg_sig(pkg, dest):
"""Zapewnia AKTUALNY podpis pakietu (.asc, fallback .sig) w cache.
Istniejący podpis jest używany tylko wtedy, gdy faktycznie weryfikuje TĘ
paczkę. Inaczej po przebudowie tej samej wersji (ten sam plik, nowy sha256)
stary podpis zostawał obok nowej paczki i weryfikacja dawała fałszywe
„NIEPRAWIDŁOWY PODPIS GPG”.
"""
for ext in (".asc", ".sig"):
sig_dest = dest + ext
if os.path.exists(sig_dest):
ok, _fp = _gpg_verify_fp(sig_dest, dest)
if ok:
return
for ext in (".asc", ".sig"):
sig_dest = dest + ext
try:
req = Request(f"{pkg.repo_url}/{pkg.filename}{ext}", headers={"User-Agent":"pag/3.0"})
with urlopen(req, timeout=30) as resp:
data = resp.read()
except Exception:
continue
# nie mieszaj rozszerzeń – zostaje tylko ten wariant podpisu
for other in (".asc", ".sig"):
if other != ext:
try:
os.remove(dest + other)
except OSError:
pass
with open(sig_dest, "wb") as f:
f.write(data)
return
# Nie udało się pobrać podpisu – usuń nieaktualny z cache, żeby weryfikacja
# nie porównywała paczki z podpisem od innej wersji (czytelny „BRAK PODPISU”).
for other in (".asc", ".sig"):
try:
os.remove(dest + other)
except OSError:
pass
def _download_packages_parallel(pkgs: List[PackageInfo], max_workers: int = 4) -> Dict[str, Optional[str]]:
"""
Równoległe pobieranie wielu pakietów przez ThreadPoolExecutor.
Znacząco przyspiesza przy dużych aktualizacjach (50+ pakietów).
Zwraca słownik {nazwa_pakietu: ścieżka_lub_None}.
"""
results = {}
total = len(pkgs)
completed = 0
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_pkg = {executor.submit(_download_pkg, pkg): pkg for pkg in pkgs}
for future in as_completed(future_to_pkg):
pkg = future_to_pkg[future]
try:
results[pkg.name] = future.result()
except Exception:
results[pkg.name] = None
completed += 1
# Pasek postępu
pct = completed / total * 100
filled = int(20 * pct / 100)
bar = "█" * filled + "░" * (20 - filled)
print(f"\r ⏬ [{bar}] {completed}/{total} ({pct:.0f}%)", end="", file=sys.stderr, flush=True)
print(file=sys.stderr) # nowa linia po zakończeniu
return results
def load_world():
if not os.path.exists(WORLD_FILE): return set()
return {l.strip() for l in open(WORLD_FILE) if l.strip()}
def save_world(w):
with open(WORLD_FILE,"w") as f:
for n in sorted(w): f.write(f"{n}\n")
def _find_orphans(installed, world):
needed = set(world)
changed = True
while changed:
changed = False
for n in list(needed):
for dep in installed.get(n,{}).get("dependencies",[]):
if dep not in needed and dep in installed:
needed.add(dep); changed = True
return {n for n in installed if n not in needed}
# =============================================================================
# MAIN
# =============================================================================
def cmd_sbom(argv):
"""pag sbom export [spdx|cyclonedx] – manifest SBOM zainstalowanych pakietów.
Wypisuje na stdout JSON (SPDX 2.3 lub CycloneDX 1.5) z listą
zainstalowanych pakietów, wersji, licencji i sum SHA256.
"""
fmt = (argv[0] if argv else "spdx").lower()
if fmt not in ("spdx", "cyclonedx"):
print("❌ Format: spdx | cyclonedx")
return 1
installed = load_json(INSTALLED_DB)
if not installed:
print("{}") if fmt == "cyclonedx" else print("{\"packages\": []}")
return 0
# metadata repo (licencje) – best-effort
try:
repo = fetch_all_packages()
except Exception:
repo = {}
names = sorted(installed)
created = datetime.now().astimezone().isoformat(timespec="seconds")
def _license_of(name):
p = repo.get(name)
lic = getattr(p, "license", None) or []
if isinstance(lic, list):
lic = ", ".join(x for x in lic if x)
return lic or "NOASSERTION"
if fmt == "spdx":
doc = {
"spdxVersion": "SPDX-2.3",
"dataLicense": "CC0-1.0",
"SPDXID": "SPDXRef-DOCUMENT",
"name": "PaganOS-installed",
"documentNamespace": f"https://repo.paganlinux.eu/sbom/installed-{int(time.time())}",
"creationInfo": {
"created": created,
"creators": [f"Tool: pag-{PAG_VERSION}"],
},
"packages": [],
}
for i, n in enumerate(names):
info = installed[n]
doc["packages"].append({
"SPDXID": f"SPDXRef-Package-{i+1}",
"name": n,
"versionInfo": info.get("version", ""),
"downloadLocation": info.get("repo", "NOASSERTION"),
"filesAnalyzed": False,
"licenseConcluded": _license_of(n),
"checksums": [{"algorithm": "SHA256", "checksumValue": info.get("sha256", "")}],
})
else: # cyclonedx
doc = {
"bomFormat": "CycloneDX",
"specVersion": "1.5",
"serialNumber": f"urn:uuid:{str(uuid.uuid4())}",
"version": 1,
"metadata": {
"timestamp": created,
"tools": [{"vendor": "PaganOS", "name": "pag", "version": PAG_VERSION}],
},
"components": [],
}
for n in names:
info = installed[n]
lic = _license_of(n)
comp = {
"type": "library",
"name": n,
"version": info.get("version", ""),
"hashes": [{"alg": "SHA-256", "content": info.get("sha256", "")}],
}
if lic != "NOASSERTION":
comp["licenses"] = [{"license": {"id": lic}}]
doc["components"].append(comp)
print(json.dumps(doc, indent=2, ensure_ascii=False))
return 0
USAGE_EN = """pag v3 – Pagan Linux Package Manager
BASIC:
pag install <pkg>... Install packages
pag remove <pkg>... Remove packages
pag update Update PACKAGES (refreshes indexes first)
pag sync Refresh indexes + show pending package updates
pag upgrade Update SYSTEM (packages + kernel/initramfs/GRUB)
pag list [--installed] List available / installed
pag search <query> Search packages
pag info <pkg> Package details
pag files <pkg> List package files
pag verify [--deep] Verify integrity (--deep = SHA256 per file)
pag clean Clear download cache
pag stats System statistics
pag download <pkg>... Download packages to cache (offline prep)
ROLLING (base baked into the image):
pag base [status] Image packages vs. repo (what pag manages)
pag base adopt [pkg...] Adopt base packages from the repo (--dry-run = preview)
SECURITY:
pag key-add <url|file> Import GPG key
pag key-list List trusted keys
pag key-remove <id> Remove key
pag key-trust <repo> Pin repo signing key fingerprint (no TOFU)
pag key-untrust <repo> Forget repo fingerprint (back to TOFU)
pag key-trusted List pinned repo fingerprints
ADVANCED:
pag why <pkg> Show why a package is installed
pag autoremove Auto-remove orphaned dependencies
pag pin <pkg> [ver] Pin package version
pag unpin <pkg> Unpin
pag pinned List pinned
pag history Transaction history
pag rollback Rollback last transaction
pag remove-orphans Remove orphaned deps
pag repo-add <url> [name] Add repository (drop-in /etc/pag/repos/)
pag repo-list List repositories
pag sbom export [fmt] SBOM manifest (spdx|cyclonedx)
FLATPAK:
pag flatpak [<query>] Search & install (smart)
pag flatpak search <q> Search Flathub
pag flatpak install <id> Install flatpak
pag flatpak remove <id> Remove flatpak
pag flatpak list List installed flatpaks
pag flatpak update Update all flatpaks
pag flatpak info <id> Show flatpak details
IMMUTABLE OS (PAG_IMMUTABLE=1):
pag deploy-list List all deployments
pag deploy-rollback Switch to previous deployment
pag deploy-cleanup [N] Remove old deployments (keep last N, default 3)
pag initramfs-update Rebuild initramfs for current kernel/deployment
pag grub-update Regenerate GRUB entries for all deployments
"""
USAGE_PL = """pag v3 – Pagan Linux Package Manager
PODSTAWOWE:
pag install <pkg>... Instalacja pakietów
pag remove <pkg>... Usuwanie pakietów
pag update Aktualizacja PAKIETÓW (odświeża indeksy)
pag sync Odśwież indeksy + info o aktualizacjach
pag upgrade Aktualizacja SYSTEMU (pakiety + kernel/initramfs/GRUB)
pag list [--installed] Lista dostępnych / zainstalowanych
pag search <query> Szukaj pakietów
pag info <pkg> Szczegóły pakietu
pag files <pkg> Lista plików pakietu
pag verify [--deep] Weryfikacja integralności
pag clean Wyczyść cache pobierania
pag stats Statystyki systemu
pag download <pkg>... Pobierz do cache (offline)
ROLLING (baza wypalona w obrazie):
pag base [status] Pakiety z obrazu vs repo (czym pag zarządza)
pag base adopt [pkg...] Adoptuj pakiety bazy z repo (--dry-run = podgląd)
BEZPIECZEŃSTWO:
pag key-add <url|file> Importuj klucz GPG
pag key-list Lista zaufanych kluczy
pag key-remove <id> Usuń klucz
pag key-trust <repo> Przypnij fingerprint klucza repo (bez TOFU)
pag key-untrust <repo> Zapomnij fingerprint repo (powrót do TOFU)
pag key-trusted Lista przypiętych fingerprintów repo
ZAAWANSOWANE:
pag why <pkg> Dlaczego pakiet jest zainstalowany
pag autoremove Usuń osierocone zależności
pag pin <pkg> [ver] Przypnij wersję pakietu
pag unpin <pkg> Odepnij
pag pinned Lista przypiętych
pag history Historia transakcji
pag rollback Cofnij ostatnią transakcję
pag remove-orphans Usuń osierocone zależności
pag repo-add <url> [nazwa] Dodaj repozytorium (drop-in w /etc/pag/repos/)
pag repo-list Lista repozytoriów
pag sbom export [fmt] Manifest SBOM (spdx|cyclonedx)
FLATPAK:
pag flatpak [<query>] Szukaj i instaluj
pag flatpak search <q> Szukaj na Flathub
pag flatpak install <id> Zainstaluj flatpak
pag flatpak remove <id> Usuń flatpak
pag flatpak list Lista zainstalowanych
pag flatpak update Aktualizuj wszystkie
pag flatpak info <id> Szczegóły flatpaka
IMMUTABLE OS (PAG_IMMUTABLE=1):
pag deploy-list Lista wdrożeń
pag deploy-rollback Przełącz na poprzednie wdrożenie
pag deploy-cleanup [N] Usuń stare wdrożenia (zachowaj N, domyślnie 3)
pag initramfs-update Przebuduj initramfs
pag grub-update Regeneruj wpisy GRUB"""
def _get_usage():
# Plik językowy może dostarczyć klucz "usage" – wtedy wygrywa z wbudowanym.
_u = T.get(LANG, {}).get("usage")
if _u:
return _u
if LANG == "pl":
return USAGE_PL
return USAGE_EN
def _extract_lang(outdir: str) -> int:
"""Eksport wbudowanych tłumaczeń do outdir/{pl,en}.json (+ klucz "usage").
Używane przez recepturę pakietu (pag.pag), żeby tłumaczenia jechały RAZEM
z wersją paga – po `pag install/upgrade pag` i self-update są zawsze zgodne.
"""
os.makedirs(outdir, exist_ok=True)
for _code in ("pl", "en"):
_d = dict(T.get(_code, {}))
_u = globals().get(f"USAGE_{_code.upper()}", "")
if _u:
_d["usage"] = _u
_p = os.path.join(outdir, f"{_code}.json")
with open(_p, "w", encoding="utf-8") as _fh:
json.dump(_d, _fh, ensure_ascii=False, indent=2, sort_keys=True)
print(f" ✓ {_p} ({len(_d)} kluczy)")
return 0
def main():
# Ukryte (używane przy budowie pakietu): eksport tłumaczeń do plików
if len(sys.argv) >= 3 and sys.argv[1] == "--lang-extract":
sys.exit(_extract_lang(sys.argv[2]))
if len(sys.argv) >= 2 and sys.argv[1] in ("--version", "-V", "version"):
print(f"pag {PAG_VERSION}")
sys.exit(0)
if len(sys.argv) == 2 and sys.argv[1] in ("--help", "-h", "help"):
print(_get_usage()); sys.exit(0)
if len(sys.argv) < 2:
print(_get_usage()); sys.exit(0)
cmd = sys.argv[1]
args = sys.argv[2:]
# Python bez modułu ssl = brak HTTPS w urllib. Powiedz to wprost, zamiast
# pokazywać mylące „unknown url type: https” przy każdej operacji sieciowej.
if not _ssl_ok():
print(f" ⚠ {_('ssl_broken')}", file=sys.stderr)
# --- Komendy TYLKO DO ODCZYTU (nie wymagają roota) ---
READ_ONLY = {
"list": lambda: cmd_list("--installed" in args),
"search": lambda: cmd_search(args[0]) if args else print("Usage: pag search <query>"),
"info": lambda: cmd_info(args[0]) if args else print("Usage: pag info <pkg>"),
"files": lambda: cmd_files(args[0]) if args else print("Usage: pag files <pkg>"),
"verify": lambda: cmd_verify("--deep" in args),
"why": lambda: cmd_why(args[0]) if args else print("Usage: pag why <pkg>"),
"stats": cmd_stats,
"pinned": cmd_pinned,
"history": cmd_history,
"repo-list": cmd_repo_list,
"key-list": cmd_key_list,
"key-trusted": cmd_key_trusted,
"flatpak": lambda: cmd_flatpak(args),
"flatpak-search": lambda: cmd_flatpak_search(args[0]) if args else print("Usage: pag flatpak-search <query>"),
"flatpak-list": cmd_flatpak_list,
"flatpak-info": lambda: cmd_flatpak_info(args[0]) if args else print("Usage: pag flatpak-info <id>"),
"deploy-list": cmd_deploy_list,
"deploy": cmd_deploy_list,
"sbom": lambda: cmd_sbom(args),
}
if cmd in READ_ONLY:
sys.exit(READ_ONLY[cmd]() or 0)
# --- `pag base` / `pag base status`: tylko odczyt (bez roota i bez locka) ---
# `pag base adopt` NIE wpada tutaj – idzie ścieżką komend zapisu (roota + lock
# pilnuje wspólna obsługa niżej), żeby transakcja była bezpieczna.
if cmd in ("base", "base-status") and not (args and args[0] == "adopt"):
sys.exit(cmd_base_status() or 0)
# --- Smart search: `pag <nazwa-pakietu>` → repo + Flathub + sugestie ---
WRITE_CMDS = {
"install", "remove", "update", "sync", "upgrade", "clean", "download",
"autoremove", "remove-orphans", "pin", "unpin", "rollback",
"repo-add", "key-add", "key-remove", "key-trust", "key-untrust",
"self-update",
"flatpak", "flatpak-install", "flatpak-remove", "flatpak-update",
"deploy-rollback", "deploy-cleanup", "initramfs-update", "grub-update",
"base", "base-adopt",
}
if cmd not in WRITE_CMDS:
# Literówka komendy? (np. `pag instal steam` zamiast `pag install`) –
# zasugeruj poprawną komendę ZAMIAST wpadać w smart search (który
# potrafi wisieć na `flatpak search` aż do Ctrl-C).
_known = set(READ_ONLY) | set(WRITE_CMDS)
_close = difflib.get_close_matches(cmd, _known, n=1, cutoff=0.75)
if _close:
print(f"❌ Nieznana komenda: '{cmd}'. Czy chodziło o '{_close[0]}'?")
print(f" Uruchom 'pag' bez argumentów, aby zobaczyć listę komend.")
sys.exit(1)
sys.exit(_smart_search(" ".join([cmd] + args)) or 0)
# Obsługa flag globalnych (-y/--yes)
global_args = []
for a in args:
if a in ("-y", "--yes"):
os.environ["PAG_YES"] = "1"
else:
global_args.append(a)
args = global_args
# --- Komendy ZAPISU (wymagają roota) ---
if os.geteuid() != 0:
print(f"❌ {_('root_required')}", file=sys.stderr); sys.exit(1)
ensure_dirs()
with DatabaseLock():
WRITE_COMMANDS = {
"install": lambda: cmd_install(
[a for a in args if a not in ("-f", "--force", "--")],
upgrade=("-f" in args or "--force" in args)),
"remove": lambda: cmd_remove(args),
"update": lambda: cmd_update(do_upgrade=True),
"sync": lambda: cmd_update(do_upgrade=False),
"upgrade": cmd_upgrade,
"clean": cmd_clean,
"download": lambda: cmd_download(args),
"autoremove": cmd_autoremove,
"remove-orphans": cmd_remove_orphans,
"pin": lambda: cmd_pin(args[0], args[1] if len(args)>1 else ""),
"unpin": lambda: cmd_unpin(args[0]) if args else print("Usage: pag unpin <pkg>"),
"rollback": cmd_rollback,
"repo-add": lambda: cmd_repo_add(args[0], args[1] if len(args) > 1 else "") if args else print("Usage: pag repo-add <url> [name]"),
"key-add": lambda: cmd_key_add(args[0]) if args else print("Usage: pag key-add <url|file>"),
"key-remove": lambda: cmd_key_remove(args[0]) if args else print("Usage: pag key-remove <id>"),
"key-trust": lambda: cmd_key_trust(args[0]) if args else print("Usage: pag key-trust <repo_url>"),
"key-untrust": lambda: cmd_key_untrust(args[0]) if args else print("Usage: pag key-untrust <repo_url>"),
"self-update": cmd_self_update,
"flatpak": lambda: cmd_flatpak(args),
"flatpak-install": lambda: _flatpak_smart_install(args) if args else print("Usage: pag flatpak-install <app>"),
"flatpak-remove": lambda: _flatpak_smart_remove(args) if args else print("Usage: pag flatpak-remove <app>"),
"flatpak-update": cmd_flatpak_update,
"deploy-rollback": cmd_deploy_rollback,
"deploy-cleanup": lambda: cmd_deploy_cleanup(int(args[0]) if args else 3),
"initramfs-update": cmd_initramfs_update,
"grub-update": cmd_grub_update,
"base": lambda: cmd_base_adopt(args),
"base-adopt": lambda: cmd_base_adopt(args),
}
fn = WRITE_COMMANDS.get(cmd)
if fn:
sys.exit(fn() or 0)
# Should never reach here – _smart_search handles unknowns
sys.exit(_smart_search(" ".join([cmd] + args)) or 0)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
# Ctrl-C (np. podczas flatpak search / pobierania) – bez tracebacka
print("\n ⚠ Przerwano (Ctrl-C).")
sys.exit(130)