#!/usr/bin/env python3 """Genesis V1 : ouvrir, déposer une seed, installer et lancer.""" import argparse, bz2, hashlib, json, lzma, os, queue, shutil, struct, subprocess, sys, tempfile, threading, zlib from pathlib import Path LIMIT = 256 * 1024**2 BASE = Path(os.environ.get("GENESIS_INSTALL_DIR", Path.home()/".local/share/genesis-refonte-install")) def unique(pairs): result = {} for key,value in pairs: if key in result: raise ValueError("Clé JSON en double") result[key] = value return result def inflate(data, codec, limit): if codec == 0: if len(data) > limit: raise ValueError("Budget dépassé") return data d = {1: zlib.decompressobj, 2: lambda: lzma.LZMADecompressor(memlimit=256*1024**2), 3: bz2.BZ2Decompressor}[codec]() out = d.decompress(data, limit+1) if len(out) > limit or not d.eof or d.unused_data: raise ValueError("Seed tronquée ou trop volumineuse") return out def contents(seed): if seed.stat().st_size > LIMIT: raise ValueError("Seed trop volumineuse") raw = seed.read_bytes(); magic = raw[:4] if magic in (b"GWV2", b"GWS1"): h = struct.Struct(">4sBBBBQQ32s") _, version, transform, codec, flags, size, packed, digest = h.unpack_from(raw) data = raw[h.size:] if version != 1 or flags or transform > 3 or size > LIMIT or len(data) != packed: raise ValueError("En-tête Wave invalide") data = inflate(data, codec, size); out = bytearray(data) for i in range(1, len(out)): if transform == 1: out[i] = (data[i]+out[i-1]) & 255 elif transform == 2: out[i] = (data[i]+2*out[i-1]-(out[i-2] if i > 1 else out[0])) & 255 elif transform == 3: out[i] = data[i]^out[i-1] data = bytes(out) elif magic == b"GSE1": h = struct.Struct(">4sBBQQ32s") _, version, count, size, packed, digest = h.unpack_from(raw) stages = raw[h.size:h.size+count]; data = raw[h.size+count:] if version != 1 or count > 8 or size > LIMIT or len(data) != packed: raise ValueError("En-tête GSEED invalide") for stage in reversed(stages): data = inflate(data, {90:1,76:2,66:3}[stage], min(LIMIT,max(size,packed)+1024**2)) else: raise ValueError("Seed .gwave ou .gseed compatible attendue") if len(data) != size or hashlib.sha256(data).digest() != digest: raise ValueError("Intégrité de la seed incorrecte") if data[:5] != b"GPAK1": raise ValueError("Cette seed ne contient pas un logiciel installable") end = 13+int.from_bytes(data[5:13],"big") if end > len(data) or end > 8*1024**2: raise ValueError("Manifeste invalide") manifest = json.loads(data[13:end],object_pairs_hook=unique); records = []; seen = set(); cursor = end if manifest.get("format") != "GPAK1" or len(manifest.get("files",[])) > 20000: raise ValueError("Paquet incompatible") for entry in manifest["files"]: name = entry["path"]; size = entry["size"] if not isinstance(name,str) or any(p in ("",".","..") for p in name.split("/")) or "\\" in name or "\0" in name or ":" in name.split("/")[0] or name in seen: raise ValueError("Chemin interdit") if type(size) is not int or size < 0 or cursor+size > len(data): raise ValueError("Taille incorrecte") block = data[cursor:cursor+size]; cursor += size; seen.add(name) if hashlib.sha256(block).hexdigest() != entry["sha256"]: raise ValueError("Fichier corrompu : "+name) records.append((name,block)) if cursor != len(data) or any("/".join(n.split("/")[:i]) in seen for n in seen for i in range(1,len(n.split("/")))): raise ValueError("Contenu du paquet invalide") return records def install(seed, args, report=lambda s: None, cancel=lambda:False): report("Vérification de la seed…"); records = contents(seed) if cancel(): raise InterruptedError("Installation annulée avant modification") meta = json.loads(dict(records).get("install.json",b"{}")) if meta.get("format") != "Genesis-Install-1" or meta.get("launcher") != 1: raise ValueError("Seed incompatible avec le lanceur V1") if args.inspect: return meta report("Préparation de Genesis "+meta["version"]+"…") with tempfile.TemporaryDirectory(prefix="genesis-v1-") as tmp: root = Path(tmp) for name,data in records: if cancel(): raise InterruptedError("Installation annulée avant modification") path = root/name; path.parent.mkdir(parents=True,exist_ok=True); path.write_bytes(data) result = subprocess.run([sys.executable,str(root/"app/genesis_install.py"),str(root),str(args.destination)], capture_output=True,text=True,timeout=120) if result.returncode: raise RuntimeError(result.stderr.strip() or result.stdout.strip()) receipt = json.loads(result.stdout) report("Installation terminée.") receipt["launch_cancelled"] = cancel() if not args.no_launch and not cancel(): launch(args.destination) return receipt def launch(base): state = json.loads((base/"current.json").read_text()) release = base/"releases"/state["release"] if release.parent.resolve() != (base/"releases").resolve(): raise ValueError("Version installée invalide") return subprocess.Popen([sys.executable,str(release/"app/genesis_app.py")],cwd=release/"app",start_new_session=True) def main(): parser = argparse.ArgumentParser(description=__doc__); parser.add_argument("seed",nargs="?",type=Path) parser.add_argument("--destination",type=Path,default=BASE); parser.add_argument("--headless",action="store_true") parser.add_argument("--no-launch",action="store_true"); parser.add_argument("--inspect",action="store_true") args = parser.parse_args() if args.headless or args.inspect: if not args.seed: parser.error("Seed requise") print(json.dumps(install(args.seed,args),ensure_ascii=False)); return if not args.seed and (args.destination/"current.json").is_file(): launch(args.destination); return import tkinter as tk from tkinter import filedialog,ttk,messagebox try: from tkinterdnd2 import TkinterDnD,DND_FILES window = TkinterDnD.Tk() except ImportError: window = tk.Tk(); DND_FILES = None window.title("Genesis · Installation V1"); window.geometry("520x250"); window.resizable(False,False) body = ttk.Frame(window,padding=24); body.pack(fill="both",expand=True) ttk.Label(body,text="Genesis",font=("Sans",24,"bold")).pack(anchor="w") ttk.Label(body,text="Déposez votre seed .gwave ou .gseed pour installer.").pack(anchor="w",pady=12) state = tk.StringVar(value="Prêt · installation locale"); messages = queue.Queue(); busy = [False]; cancelled = threading.Event() def begin(path): if not path or busy[0]: return busy[0] = True; cancelled.clear(); button.configure(state="disabled") def worker(): try: messages.put(("done",install(Path(path),args,lambda s:messages.put(("status",s)),cancelled.is_set))) except Exception as e: messages.put(("error",str(e))) threading.Thread(target=worker,daemon=True).start() button = ttk.Button(body,text="Choisir une seed…",command=lambda:begin(filedialog.askopenfilename(filetypes=[("Seeds Genesis","*.gwave *.gseed"),("Tous","*")]))); button.pack(fill="x",pady=8) ttk.Label(body,textvariable=state,wraplength=470).pack(anchor="w",pady=10) ttk.Button(body,text="Annuler",command=lambda:(cancelled.set(),state.set("Annulation demandée ; fin de l'étape d'installation engagée…")) if busy[0] else window.destroy()).pack(anchor="e") if DND_FILES: window.drop_target_register(DND_FILES); window.dnd_bind("<>",lambda e:begin(window.tk.splitlist(e.data)[0])) def poll(): try: while True: kind,value = messages.get_nowait() if kind == "status": state.set(value) elif kind == "error": busy[0]=False; button.configure(state="normal"); state.set("Installation interrompue"); messagebox.showerror("Genesis",value) else: state.set("Genesis est installé."); window.after(800,window.destroy); return except queue.Empty: pass window.after(100,poll) window.protocol("WM_DELETE_WINDOW",lambda:(cancelled.set(),state.set("Annulation demandée…")) if busy[0] else window.destroy()) window.after(100,poll) if args.seed: window.after(200,lambda:begin(args.seed)) window.mainloop() if __name__ == "__main__": try: main() except Exception as error: print("Genesis V1 : "+str(error),file=sys.stderr); sys.exit(1)