#!/usr/bin/env python3 """undo_proof2.py APP MAIN_SVC FROM(svc=ref,…) TO(svc=ref,…) — a big step's UNDO, proven on scratch 9202 through the product's own guarded Update (generalised from audits/security-app-updates-2026-10-27/part3/tools/undo_proof.py, Jellyfin): 1. the drill's APP must stand at FROM (this tool checks, never guesses); install, seed with the box fixture, read back (C1); 2. a temporary drill commit moves the images to TO AND makes the checks fail (MAIN_SVC's compose healthcheck `false`, the .felhom.yml probe on a path that does not exist) — so TO starts on the data, runs its migrations, and the Update fails in its check; 3. press the guarded Update: the final phase must be `undone`; MAIN_SVC's log is saved while TO verifies; 4. after the undo: the running images are FROM and the seed reads back; 5. the temporary drill commit is reverted (the template equal to before), the app removed. usage: SC= TARGET=9202 EV= python3 undo_proof2.py APP MAIN_SVC FROM TO [DEFINITION_DIR]""" import json, os, re, subprocess, sys, threading, time sys.path.insert(0, "/mnt/5_hdd/felhom.eu/git/app-catalog-felhom.eu/scripts") import box_walk as w import upgrade_fixtures_box as fb import upgrade_fixtures_box28 as f28 APP, MAIN = sys.argv[1], sys.argv[2] FROM = dict(x.split("=", 1) for x in sys.argv[3].split(",")) TO = dict(x.split("=", 1) for x in sys.argv[4].split(",")) DEF = sys.argv[5] if len(sys.argv) > 5 else None # a definition step: TO is that whole definition (box_walk --to-definition) FX = dict(f28.FIXTURES28); FX.update(fb.FIXTURES) fx = FX[APP] SUB = getattr(fx, "sub", APP) D = w.DRILL EV = os.environ["EV"] os.makedirs(EV, exist_ok=True) OUT = os.path.join(EV, "undo-proof-%s.json" % APP) rec = {"app": APP, "from": FROM, "to": TO, "started": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} def git(*a): return subprocess.run(["git", "-C", D] + list(a), capture_output=True, text=True, timeout=120) def finish(verdict, why): rec["verdict"], rec["why"] = verdict, why w.say(f"RESULT {verdict} — {why}") def failing_to(c, y): if DEF: c = open(os.path.join(DEF, "docker-compose.yml")).read() y = open(os.path.join(DEF, ".felhom.yml")).read() assert w.images_of(c) == TO, (w.images_of(c), TO) for svc in ({} if DEF else FROM): if FROM[svc] != TO[svc]: assert f"image: {FROM[svc]}" in c, svc c = c.replace(f"image: {FROM[svc]}", f"image: {TO[svc]}") # MAIN's own block only (never past the next service), comment lines allowed between healthcheck: and test: m = re.search(r"(?m)^ %s:\n(?:(?! \S)[^\n]*\n)*? healthcheck:\n(?:\s*#[^\n]*\n)*(\s+)test: [^\n]*\n" % re.escape(MAIN), c) if m: c = c[:m.start()] + m.group(0).replace(m.group(0).split("\n")[-2], m.group(1) + 'test: ["CMD", "false"]') + c[m.end():] else: # no test line of our own (the image's own): add a failing healthcheck under the service c = re.sub(r"(?m)^( %s:\n)" % re.escape(MAIN), r'\1 healthcheck:\n test: ["CMD", "false"]\n', c, count=1) # port 1: nothing listens there, so the probe fails on ANY app — an app that answers 200 on every path (a single-page # app: Termix, 2026-10-28) passed a „no such path" probe and the update ended `done` y2 = re.sub(r"(?m)^(healthcheck:\n checks:\n - type: )\w+(\n port: )\d+\n(?: path: [^\n]*\n)?(?: expect:\n status: \d+\n)?", r'\1api\g<2>1\n path: "/undo-proof-no-such-path"\n expect:\n status: 200\n', y, count=1) return c, y2 w.login() git("pull", "-q", "--rebase", "origin", "main") head0 = git("rev-parse", "HEAD").stdout.strip() rec["drill_head_before"] = head0 comp, fy = f"{D}/templates/{APP}/docker-compose.yml", f"{D}/templates/{APP}/.felhom.yml" commit = None try: if w.images_of(open(comp).read()) != FROM: finish("inconclusive", f"the drill's {APP} is not at FROM"); raise SystemExit(1) if w.stack(APP).get("deployed"): w.remove(APP) w.refresh_box_mirror() w.sync_rescan() if not w.deploy(APP, SUB): finish("inconclusive", "the install at FROM did not complete"); raise SystemExit(1) tok = fx.seed(w, SUB, w.say) if not tok or not fx.verify(w, SUB, tok, w.say): finish("inconclusive", "C1: the seed did not read back before the update"); raise SystemExit(1) c, y = failing_to(open(comp).read(), open(fy).read()) rec["failing_compose_main_healthcheck"] = re.findall(r'test: \["CMD", "false"\]', c) rec["failing_probe"] = "/undo-proof-no-such-path" in y open(comp, "w").write(c); open(fy, "w").write(y) git("add", f"templates/{APP}") git("commit", "-q", "-m", f"UNDO-PROOF: {APP} at TO whose checks fail (temporary, reverted at the end)") git("push", "-q", "origin", "main") commit = git("rev-parse", "--short=12", "HEAD").stdout.strip() w.say(f" drill commit {commit}: {APP} TO with failing checks") w.refresh_box_mirror() main_to = TO[MAIN] w.sync_rescan(APP, main_to, tries=40, delay=10) seen = {"lines": []} stop = threading.Event() def watch(): while not stop.is_set(): out = w.guest(f"docker inspect -f '{{{{.Config.Image}}}}' {MAIN} 2>/dev/null; docker logs --tail 300 {MAIN} 2>&1 | tail -40") if main_to.split(":")[-1] in out.split("\n")[0]: seen["lines"] = out.strip().split("\n") time.sleep(5) th = threading.Thread(target=watch, daemon=True) th.start() rec["update"] = u = w.press_update(APP, poll=1.0, cap_s=1800) stop.set(); th.join(timeout=30) rec["to_log_while_verifying"] = seen["lines"][-40:] time.sleep(15) rec["observables_after"] = w.observables(APP) running = {s: w.guest(f"docker inspect -f '{{{{.Config.Image}}}}' {s} 2>/dev/null").strip() for s in FROM} rec["running_images_after"] = running if u.get("final_phase") != "undone": finish("failed", f"the update ended {u.get('final_phase')!r}, not 'undone'"); raise SystemExit(1) if any(running[s].split("@")[0] != FROM[s] for s in FROM): finish("failed", f"after the undo the running images are {running}, not FROM"); raise SystemExit(1) rec["seed_after_undo"] = ok = fx.verify(w, SUB, tok, w.say) if not ok: finish("failed", "the undo put the old images back but the seed does not read back"); raise SystemExit(1) finish("proven", "TO started on the data and failed its check; the box undid it: FROM runs again and the seed reads back") finally: if w.stack(APP).get("deployed"): w.remove(APP) if commit: git("pull", "-q", "--rebase", "origin", "main") r = git("revert", "--no-edit", commit) git("push", "-q", "origin", "main") w.refresh_box_mirror() w.say(f" drill revert {commit}: rc={r.returncode}") rec["drill_head_after"] = git("rev-parse", "HEAD").stdout.strip() rec["drill_tree_equal_to_before"] = git("diff", "--quiet", head0, "HEAD", "--", f"templates/{APP}").returncode == 0 w.say(f" drill templates/{APP} equal to before: {rec['drill_tree_equal_to_before']}") rec["log"] = w.LOG json.dump(rec, open(OUT, "w"), indent=2)